构建稳定车辆GPS数据处理系统:SpringBoot + Redis + Kafka + MongoDB
myzbx 2025-10-23 08:31 6 浏览
系统概述
随着车联网和智能交通的快速发展，车辆GPS数据处理成为智能交通系统的核心。本文介绍如何基于SpringBoot、Redis、Kafka和MongoDB构建高稳定性、高吞吐量的车辆GPS数据处理系统。
核心需求
- 高并发处理：支持每秒万级GPS点位处理
- 低延迟：实时位置更新<100ms
- 数据可靠性：确保GPS数据不丢失
- 高可用性：系统组件故障自动恢复
- 弹性扩展：根据流量动态扩容
系统架构设计
核心实现详解
1. 数据接收层 - 高并发接入
@RestController
@RequestMapping("/api/v1/gps")
@Slf4j
public class GpsDataController {
@Autowired
private KafkaTemplate<String, GpsData> kafkaTemplate;
@Autowired
private RateLimiterService rateLimiterService;
@PostMapping("/upload")
public ResponseEntity<ApiResponse<String>> uploadGpsData(
@RequestBody @Valid GpsDataRequest request) {
// 1. 限流保护
if (!rateLimiterService.tryAcquire(request.getDeviceId())) {
return ResponseEntity.status(HttpStatus.TOO_MANY_REQUESTS)
.body(ApiResponse.error("请求频率过高"));
}
// 2. 数据校验
if (!validateGpsData(request)) {
return ResponseEntity.badRequest()
.body(ApiResponse.error("数据格式错误"));
}
// 3. 转换为领域模型
GpsData gpsData = convertToDomain(request);
// 4. 异步发送到Kafka
CompletableFuture<SendResult<String, GpsData>> future =
kafkaTemplate.send("gps-raw-topic", gpsData.getDeviceId(), gpsData);
// 5. 异步处理发送结果
future.whenComplete((result, ex) -> {
if (ex != null) {
log.error("GPS数据发送Kafka失败: {}", gpsData.getDeviceId(), ex);
// 可降级到本地存储或重试队列
degradeToLocalStorage(gpsData);
} else {
log.debug("GPS数据发送成功: {}", result.getRecordMetadata().offset());
}
});
return ResponseEntity.ok(ApiResponse.success("接收成功"));
}
// 数据校验逻辑
private boolean validateGpsData(GpsDataRequest request) {
return request.getLatitude() >= -90 && request.getLatitude() <= 90 &&
request.getLongitude() >= -180 && request.getLongitude() <= 180 &&
request.getTimestamp() > 0 &&
StringUtils.isNotBlank(request.getDeviceId());
}
}2. Kafka配置 - 高可用消息队列
# application-kafka.yml
spring:
kafka:
producer:
bootstrap-servers: ${KAFKA_SERVERS:localhost:9092}
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
compression-type: gzip
acks: all
retries: 3
batch-size: 16384
linger-ms: 10
buffer-memory: 33554432
consumer:
bootstrap-servers: ${KAFKA_SERVERS:localhost:9092}
group-id: gps-consumer-group
auto-offset-reset: latest
enable-auto-commit: false
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
properties:
spring.json.trusted.packages: "com.example.gps.domain"
listener:
concurrency: 4
ack-mode: manual3. 数据处理层 - 核心业务逻辑
@Service
@Slf4j
public class GpsDataProcessService {
@Autowired
private RedisTemplate<String, VehicleLocation> redisTemplate;
@Autowired
private MongoTemplate mongoTemplate;
@Autowired
private GeofenceService geofenceService;
@KafkaListener(topics = "gps-raw-topic")
public void processGpsData(ConsumerRecord<String, GpsData> record,
Acknowledgment ack) {
try {
GpsData gpsData = record.value();
String deviceId = gpsData.getDeviceId();
// 1. 数据去重校验
if (isDuplicateData(deviceId, gpsData)) {
ack.acknowledge();
return;
}
// 2. 数据清洗和增强
enhanceGpsData(gpsData);
// 3. 更新Redis最新位置
updateRedisLocation(deviceId, gpsData);
// 4. 地理围栏判断
checkGeofence(deviceId, gpsData);
// 5. 批量写入MongoDB
saveToMongoDB(gpsData);
// 6. 确认消息消费
ack.acknowledge();
log.debug("GPS数据处理完成: {}", deviceId);
} catch (Exception e) {
log.error("GPS数据处理异常, 设备ID: {}", record.key(), e);
// 发送到死信队列进行后续处理
sendToDlq(record);
}
}
/**
* Redis去重校验：防止重复数据处理
*/
private boolean isDuplicateData(String deviceId, GpsData newData) {
String redisKey = buildRedisKey(deviceId);
VehicleLocation lastLocation = redisTemplate.opsForValue().get(redisKey);
if (lastLocation == null) {
return false;
}
// 相同位置且时间相近的数据视为重复
return newData.getTimestamp() - lastLocation.getTimestamp() < 1000 &&
distanceBetween(lastLocation, newData) < 10; // 10米内视为相同位置
}
/**
* 数据增强：计算速度、方向等
*/
private void enhanceGpsData(GpsData currentData) {
String redisKey = buildRedisKey(currentData.getDeviceId());
VehicleLocation lastLocation = redisTemplate.opsForValue().get(redisKey);
if (lastLocation != null) {
// 计算速度(km/h)
double distance = calculateDistance(lastLocation, currentData);
double timeDiff = (currentData.getTimestamp() - lastLocation.getTimestamp()) / 1000.0 / 3600; // 小时
double speed = timeDiff > 0 ? distance / timeDiff : 0;
currentData.setSpeed(speed);
// 计算方向
double direction = calculateDirection(lastLocation, currentData);
currentData.setDirection(direction);
}
// 数据质量标记
currentData.setQuality(calculateDataQuality(currentData));
}
/**
* 更新Redis中的最新位置
*/
private void updateRedisLocation(String deviceId, GpsData gpsData) {
String redisKey = buildRedisKey(deviceId);
VehicleLocation location = convertToVehicleLocation(gpsData);
// 设置5分钟过期，避免内存泄漏
redisTemplate.opsForValue().set(redisKey, location, Duration.ofMinutes(5));
// 更新GEO索引用于附近车辆查询
redisTemplate.opsForGeo().add("vehicle:geo",
new Point(gpsData.getLongitude(), gpsData.getLatitude()), deviceId);
}
}4. MongoDB数据存储优化
@Document(collection = "gps_tracks")
@CompoundIndex(name = "device_timestamp_idx", def = "{'deviceId': 1, 'timestamp': -1}")
@CompoundIndex(name = "timestamp_idx", def = "{'timestamp': 1}")
public class GpsTrackDocument {
@Id
private String id;
@Indexed
private String deviceId;
private double latitude;
private double longitude;
private double speed;
private double direction;
private String quality;
@Indexed(direction = IndexDirection.DESCENDING)
private long timestamp;
private Date createTime;
// TTL索引，自动清理90天前数据
@Indexed(expireAfterSeconds = 7776000)
private Date expireAt;
// 分片键
@Sharded
private String shardKey;
}
@Repository
public class GpsTrackRepository {
@Autowired
private MongoTemplate mongoTemplate;
/**
* 批量插入GPS数据
*/
public void batchInsert(List<GpsTrackDocument> tracks) {
if (tracks.isEmpty()) return;
BulkOperations bulkOps = mongoTemplate.bulkOps(BulkMode.ORDERED, GpsTrackDocument.class);
for (GpsTrackDocument track : tracks) {
bulkOps.insert(track);
}
bulkOps.execute();
}
/**
* 查询车辆轨迹
*/
public List<GpsTrackDocument> findTracks(String deviceId, long startTime, long endTime) {
Query query = new Query(Criteria.where("deviceId").is(deviceId)
.and("timestamp").gte(startTime).lte(endTime))
.with(Sort.by(Sort.Direction.ASC, "timestamp"));
return mongoTemplate.find(query, GpsTrackDocument.class);
}
/**
* 查询区域内车辆轨迹
*/
public List<GpsTrackDocument> findTracksInArea(double minLng, double minLat,
double maxLng, double maxLat,
long startTime, long endTime) {
Query query = new Query(Criteria
.where("timestamp").gte(startTime).lte(endTime)
.and("longitude").gte(minLng).lte(maxLng)
.and("latitude").gte(minLat).lte(maxLat));
return mongoTemplate.find(query, GpsTrackDocument.class);
}
}构建高稳定车辆GPS数据处理系统：SpringBoot + Redis + Kafka + MongoDB实战
系统概述
随着车联网和智能交通的快速发展，车辆GPS数据处理成为智能交通系统的核心。本文介绍如何基于SpringBoot、Redis、Kafka和MongoDB构建高稳定性、高吞吐量的车辆GPS数据处理系统。
核心需求
- 高并发处理：支持每秒万级GPS点位处理
- 低延迟：实时位置更新<100ms
- 数据可靠性：确保GPS数据不丢失
- 高可用性：系统组件故障自动恢复
- 弹性扩展：根据流量动态扩容
系统架构设计
业务流程图
A[GPS设备] -->|HTTP/WebSocket| B(SpringBoot API网关)
B -->|异步写入| C[Kafka集群]
C -->|消息队列| D[数据处理服务]
D -->|缓存最新位置| E[Redis集群]
D -->|存储历史轨迹| F[MongoDB分片集群]
E --> G[实时查询服务]
F --> H[轨迹查询服务]
F --> I[地理围栏分析]
F --> J[数据统计分析]
K[监控告警] -->|系统监控| B
K -->|性能监控| D
K -->|业务监控| G核心实现详解
1. 数据接收层 - 高并发接入
@RestController
@RequestMapping("/api/v1/gps")
@Slf4j
public class GpsDataController {
@Autowired
private KafkaTemplate<String, GpsData> kafkaTemplate;
@Autowired
private RateLimiterService rateLimiterService;
@PostMapping("/upload")
public ResponseEntity<ApiResponse<String>> uploadGpsData(
@RequestBody @Valid GpsDataRequest request) {
// 1. 限流保护
if (!rateLimiterService.tryAcquire(request.getDeviceId())) {
return ResponseEntity.status(HttpStatus.TOO_MANY_REQUESTS)
.body(ApiResponse.error("请求频率过高"));
}
// 2. 数据校验
if (!validateGpsData(request)) {
return ResponseEntity.badRequest()
.body(ApiResponse.error("数据格式错误"));
}
// 3. 转换为领域模型
GpsData gpsData = convertToDomain(request);
// 4. 异步发送到Kafka
CompletableFuture<SendResult<String, GpsData>> future =
kafkaTemplate.send("gps-raw-topic", gpsData.getDeviceId(), gpsData);
// 5. 异步处理发送结果
future.whenComplete((result, ex) -> {
if (ex != null) {
log.error("GPS数据发送Kafka失败: {}", gpsData.getDeviceId(), ex);
// 可降级到本地存储或重试队列
degradeToLocalStorage(gpsData);
} else {
log.debug("GPS数据发送成功: {}", result.getRecordMetadata().offset());
}
});
return ResponseEntity.ok(ApiResponse.success("接收成功"));
}
// 数据校验逻辑
private boolean validateGpsData(GpsDataRequest request) {
return request.getLatitude() >= -90 && request.getLatitude() <= 90 &&
request.getLongitude() >= -180 && request.getLongitude() <= 180 &&
request.getTimestamp() > 0 &&
StringUtils.isNotBlank(request.getDeviceId());
}
}2. Kafka配置 - 高可用消息队列
# application-kafka.yml
spring:
kafka:
producer:
bootstrap-servers: ${KAFKA_SERVERS:localhost:9092}
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
compression-type: gzip
acks: all
retries: 3
batch-size: 16384
linger-ms: 10
buffer-memory: 33554432
consumer:
bootstrap-servers: ${KAFKA_SERVERS:localhost:9092}
group-id: gps-consumer-group
auto-offset-reset: latest
enable-auto-commit: false
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
properties:
spring.json.trusted.packages: "com.example.gps.domain"
listener:
concurrency: 4
ack-mode: manual3. 数据处理层 - 核心业务逻辑
@Service
@Slf4j
public class GpsDataProcessService {
@Autowired
private RedisTemplate<String, VehicleLocation> redisTemplate;
@Autowired
private MongoTemplate mongoTemplate;
@Autowired
private GeofenceService geofenceService;
@KafkaListener(topics = "gps-raw-topic")
public void processGpsData(ConsumerRecord<String, GpsData> record,
Acknowledgment ack) {
try {
GpsData gpsData = record.value();
String deviceId = gpsData.getDeviceId();
// 1. 数据去重校验
if (isDuplicateData(deviceId, gpsData)) {
ack.acknowledge();
return;
}
// 2. 数据清洗和增强
enhanceGpsData(gpsData);
// 3. 更新Redis最新位置
updateRedisLocation(deviceId, gpsData);
// 4. 地理围栏判断
checkGeofence(deviceId, gpsData);
// 5. 批量写入MongoDB
saveToMongoDB(gpsData);
// 6. 确认消息消费
ack.acknowledge();
log.debug("GPS数据处理完成: {}", deviceId);
} catch (Exception e) {
log.error("GPS数据处理异常, 设备ID: {}", record.key(), e);
// 发送到死信队列进行后续处理
sendToDlq(record);
}
}
/**
* Redis去重校验：防止重复数据处理
*/
private boolean isDuplicateData(String deviceId, GpsData newData) {
String redisKey = buildRedisKey(deviceId);
VehicleLocation lastLocation = redisTemplate.opsForValue().get(redisKey);
if (lastLocation == null) {
return false;
}
// 相同位置且时间相近的数据视为重复
return newData.getTimestamp() - lastLocation.getTimestamp() < 1000 &&
distanceBetween(lastLocation, newData) < 10; // 10米内视为相同位置
}
/**
* 数据增强：计算速度、方向等
*/
private void enhanceGpsData(GpsData currentData) {
String redisKey = buildRedisKey(currentData.getDeviceId());
VehicleLocation lastLocation = redisTemplate.opsForValue().get(redisKey);
if (lastLocation != null) {
// 计算速度(km/h)
double distance = calculateDistance(lastLocation, currentData);
double timeDiff = (currentData.getTimestamp() - lastLocation.getTimestamp()) / 1000.0 / 3600; // 小时
double speed = timeDiff > 0 ? distance / timeDiff : 0;
currentData.setSpeed(speed);
// 计算方向
double direction = calculateDirection(lastLocation, currentData);
currentData.setDirection(direction);
}
// 数据质量标记
currentData.setQuality(calculateDataQuality(currentData));
}
/**
* 更新Redis中的最新位置
*/
private void updateRedisLocation(String deviceId, GpsData gpsData) {
String redisKey = buildRedisKey(deviceId);
VehicleLocation location = convertToVehicleLocation(gpsData);
// 设置5分钟过期，避免内存泄漏
redisTemplate.opsForValue().set(redisKey, location, Duration.ofMinutes(5));
// 更新GEO索引用于附近车辆查询
redisTemplate.opsForGeo().add("vehicle:geo",
new Point(gpsData.getLongitude(), gpsData.getLatitude()), deviceId);
}
}4. MongoDB数据存储优化
@Document(collection = "gps_tracks")
@CompoundIndex(name = "device_timestamp_idx", def = "{'deviceId': 1, 'timestamp': -1}")
@CompoundIndex(name = "timestamp_idx", def = "{'timestamp': 1}")
public class GpsTrackDocument {
@Id
private String id;
@Indexed
private String deviceId;
private double latitude;
private double longitude;
private double speed;
private double direction;
private String quality;
@Indexed(direction = IndexDirection.DESCENDING)
private long timestamp;
private Date createTime;
// TTL索引，自动清理90天前数据
@Indexed(expireAfterSeconds = 7776000)
private Date expireAt;
// 分片键
@Sharded
private String shardKey;
}
@Repository
public class GpsTrackRepository {
@Autowired
private MongoTemplate mongoTemplate;
/**
* 批量插入GPS数据
*/
public void batchInsert(List<GpsTrackDocument> tracks) {
if (tracks.isEmpty()) return;
BulkOperations bulkOps = mongoTemplate.bulkOps(BulkMode.ORDERED, GpsTrackDocument.class);
for (GpsTrackDocument track : tracks) {
bulkOps.insert(track);
}
bulkOps.execute();
}
/**
* 查询车辆轨迹
*/
public List<GpsTrackDocument> findTracks(String deviceId, long startTime, long endTime) {
Query query = new Query(Criteria.where("deviceId").is(deviceId)
.and("timestamp").gte(startTime).lte(endTime))
.with(Sort.by(Sort.Direction.ASC, "timestamp"));
return mongoTemplate.find(query, GpsTrackDocument.class);
}
/**
* 查询区域内车辆轨迹
*/
public List<GpsTrackDocument> findTracksInArea(double minLng, double minLat,
double maxLng, double maxLat,
long startTime, long endTime) {
Query query = new Query(Criteria
.where("timestamp").gte(startTime).lte(endTime)
.and("longitude").gte(minLng).lte(maxLng)
.and("latitude").gte(minLat).lte(maxLat));
return mongoTemplate.find(query, GpsTrackDocument.class);
}
}数据使用场景实现
1. 实时位置查询服务
@Service
public class RealTimeLocationService {
@Autowired
private RedisTemplate<String, VehicleLocation> redisTemplate;
/**
* 获取车辆实时位置
*/
public VehicleLocation getRealTimeLocation(String deviceId) {
String redisKey = buildRedisKey(deviceId);
return redisTemplate.opsForValue().get(redisKey);
}
/**
* 批量获取车辆位置
*/
public Map<String, VehicleLocation> batchGetLocations(List<String> deviceIds) {
Map<String, VehicleLocation> result = new HashMap<>();
for (String deviceId : deviceIds) {
VehicleLocation location = getRealTimeLocation(deviceId);
if (location != null) {
result.put(deviceId, location);
}
}
return result;
}
/**
* 查询附近车辆
*/
public List<NearbyVehicle> findNearbyVehicles(double lng, double lat, double radiusKm) {
Circle circle = new Circle(new Point(lng, lat), new Distance(radiusKm, Metrics.KILOMETERS));
RedisGeoCommands.GeoRadiusCommandArgs args = RedisGeoCommands.GeoRadiusCommandArgs
.newGeoRadiusArgs().includeDistance().includeCoordinates().sortAscending();
GeoResults<RedisGeoCommands.GeoLocation<String>> results =
redisTemplate.opsForGeo().radius("vehicle:geo", circle, args);
return results.getContent().stream()
.map(this::convertToNearbyVehicle)
.collect(Collectors.toList());
}
}2. 轨迹回放服务
@Service
public class TrackPlaybackService {
@Autowired
private GpsTrackRepository trackRepository;
/**
* 轨迹回放
*/
public TrackPlaybackResult playbackTrack(String deviceId, long startTime, long endTime,
int intervalSeconds) {
// 查询原始轨迹数据
List<GpsTrackDocument> rawTracks = trackRepository.findTracks(deviceId, startTime, endTime);
if (rawTracks.isEmpty()) {
return TrackPlaybackResult.empty();
}
// 轨迹压缩和抽稀
List<GpsTrackDocument> simplifiedTracks = simplifyTrack(rawTracks, intervalSeconds);
// 计算轨迹统计信息
TrackStatistics statistics = calculateTrackStatistics(rawTracks);
return TrackPlaybackResult.builder()
.deviceId(deviceId)
.tracks(simplifiedTracks)
.statistics(statistics)
.build();
}
/**
* 轨迹抽稀算法（Douglas-Peucker）
*/
private List<GpsTrackDocument> simplifyTrack(List<GpsTrackDocument> tracks, int interval) {
if (tracks.size() <= 2) return tracks;
List<GpsTrackDocument> result = new ArrayList<>();
result.add(tracks.get(0));
long lastTime = tracks.get(0).getTimestamp();
for (int i = 1; i < tracks.size() - 1; i++) {
if (tracks.get(i).getTimestamp() - lastTime >= interval * 1000) {
result.add(tracks.get(i));
lastTime = tracks.get(i).getTimestamp();
}
}
result.add(tracks.get(tracks.size() - 1));
return result;
}
}3. 地理围栏服务
@Service
public class GeofenceService {
@Autowired
private RedisTemplate<String, String> redisTemplate;
@Autowired
private MongoTemplate mongoTemplate;
/**
* 检查车辆地理围栏状态
*/
public GeofenceCheckResult checkGeofence(String deviceId, GpsData gpsData) {
// 1. 从Redis获取车辆最近的围栏状态
String lastGeofenceKey = buildGeofenceKey(deviceId);
String lastGeofenceId = redisTemplate.opsForValue().get(lastGeofenceKey);
// 2. 查询相关的围栏
List<Geofence> relevantGeofences = findRelevantGeofences(gpsData.getLongitude(),
gpsData.getLatitude());
GeofenceCheckResult result = new GeofenceCheckResult();
result.setDeviceId(deviceId);
result.setTimestamp(System.currentTimeMillis());
for (Geofence geofence : relevantGeofences) {
boolean isInside = isPointInGeofence(gpsData.getLongitude(),
gpsData.getLatitude(), geofence);
if (isInside) {
if (!geofence.getId().equals(lastGeofenceId)) {
// 进入围栏
result.getEntered().add(geofence);
// 记录围栏进入事件
recordGeofenceEvent(deviceId, geofence, "ENTER", gpsData);
}
} else {
if (geofence.getId().equals(lastGeofenceId)) {
// 离开围栏
result.getExited().add(geofence);
// 记录围栏离开事件
recordGeofenceEvent(deviceId, geofence, "EXIT", gpsData);
}
}
}
// 更新最近的围栏状态
if (!result.getEntered().isEmpty()) {
String currentGeofenceId = result.getEntered().get(0).getId();
redisTemplate.opsForValue().set(lastGeofenceKey, currentGeofenceId,
Duration.ofHours(24));
} else if (result.getExited().isEmpty()) {
// 不在任何围栏内，清除状态
redisTemplate.delete(lastGeofenceKey);
}
return result;
}
/**
* 记录围栏事件
*/
private void recordGeofenceEvent(String deviceId, Geofence geofence,
String eventType, GpsData gpsData) {
GeofenceEvent event = GeofenceEvent.builder()
.deviceId(deviceId)
.geofenceId(geofence.getId())
.geofenceName(geofence.getName())
.eventType(eventType)
.longitude(gpsData.getLongitude())
.latitude(gpsData.getLatitude())
.timestamp(System.currentTimeMillis())
.build();
mongoTemplate.insert(event, "geofence_events");
// 发送事件到Kafka供其他系统消费
kafkaTemplate.send("geofence-events-topic", deviceId, event);
}
}4. 数据统计与分析
@Service
public class GpsDataAnalysisService {
@Autowired
private MongoTemplate mongoTemplate;
/**
* 车辆行驶统计
*/
public VehicleStatistics getVehicleStatistics(String deviceId, long startTime, long endTime) {
// 使用MongoDB聚合框架进行复杂统计
TypedAggregation<GpsTrackDocument> aggregation = Aggregation.newAggregation(
GpsTrackDocument.class,
Aggregation.match(Criteria.where("deviceId").is(deviceId)
.and("timestamp").gte(startTime).lte(endTime)),
Aggregation.sort(Sort.by("timestamp").ascending()),
Aggregation.group("deviceId")
.first("timestamp").as("startTime")
.last("timestamp").as("endTime")
.count().as("pointCount")
.avg("speed").as("avgSpeed")
.max("speed").as("maxSpeed")
.sum("distance").as("totalDistance")
);
AggregationResults<VehicleStats> results = mongoTemplate.aggregate(
aggregation, "gps_tracks", VehicleStats.class);
VehicleStats stats = results.getUniqueMappedResult();
return convertToVehicleStatistics(stats);
}
/**
* 区域热力图分析
*/
public HeatmapData getAreaHeatmap(double minLng, double minLat,
double maxLng, double maxLat,
long startTime, long endTime) {
// 网格化统计
int gridSize = 100; // 100x100网格
// MongoDB地理空间查询和统计
Query query = new Query(Criteria
.where("timestamp").gte(startTime).lte(endTime)
.and("longitude").gte(minLng).lte(maxLng)
.and("latitude").gte(minLat).lte(maxLat));
List<GpsTrackDocument> points = mongoTemplate.find(query, GpsTrackDocument.class);
// 生成热力图数据
return generateHeatmapData(points, minLng, minLat, maxLng, maxLat, gridSize);
}
}高可用保障措施
1. 熔断降级配置
@Service
@Slf4j
public class GpsDataServiceWithCircuitBreaker {
@CircuitBreaker(name = "redisService", fallbackMethod = "redisFallback")
public void updateRedisLocation(String deviceId, GpsData gpsData) {
// Redis操作
redisTemplate.opsForValue().set(buildRedisKey(deviceId),
convertToVehicleLocation(gpsData),
Duration.ofMinutes(5));
}
@CircuitBreaker(name = "mongoService", fallbackMethod = "mongoFallback")
public void saveToMongoDB(GpsData gpsData) {
mongoTemplate.insert(convertToDocument(gpsData), "gps_tracks");
}
// 降级方法
public void redisFallback(String deviceId, GpsData gpsData, Throwable t) {
log.warn("Redis服务降级, 设备ID: {}", deviceId);
// 降级策略：写入本地文件或内存队列，后续补偿
degradeToLocalQueue(gpsData);
}
public void mongoFallback(GpsData gpsData, Throwable t) {
log.warn("MongoDB服务降级, 设备ID: {}", gpsData.getDeviceId());
// 降级策略：写入本地文件，后续批量补偿
degradeToLocalFile(gpsData);
}
}高可用保障措施
1. 熔断降级配置
@Service
@Slf4j
public class GpsDataServiceWithCircuitBreaker {
@CircuitBreaker(name = "redisService", fallbackMethod = "redisFallback")
public void updateRedisLocation(String deviceId, GpsData gpsData) {
// Redis操作
redisTemplate.opsForValue().set(buildRedisKey(deviceId),
convertToVehicleLocation(gpsData),
Duration.ofMinutes(5));
}
@CircuitBreaker(name = "mongoService", fallbackMethod = "mongoFallback")
public void saveToMongoDB(GpsData gpsData) {
mongoTemplate.insert(convertToDocument(gpsData), "gps_tracks");
}
// 降级方法
public void redisFallback(String deviceId, GpsData gpsData, Throwable t) {
log.warn("Redis服务降级, 设备ID: {}", deviceId);
// 降级策略：写入本地文件或内存队列，后续补偿
degradeToLocalQueue(gpsData);
}
public void mongoFallback(GpsData gpsData, Throwable t) {
log.warn("MongoDB服务降级, 设备ID: {}", gpsData.getDeviceId());
// 降级策略：写入本地文件，后续批量补偿
degradeToLocalFile(gpsData);
}
}2. 监控告警体系
# Prometheus监控配置
management:
endpoints:
web:
exposure:
include: health,info,metrics,prometheus
endpoint:
health:
show-details: always
metrics:
enabled: true
prometheus:
enabled: true
# 自定义监控指标
@Component
public class GpsDataMetrics {
private final Counter gpsReceivedCounter;
private final Counter gpsProcessedCounter;
private final Counter gpsErrorCounter;
private final Gauge kafkaLagGauge;
public GpsDataMetrics(MeterRegistry registry) {
this.gpsReceivedCounter = Counter.builder("gps.data.received")
.description("接收到的GPS数据数量")
.register(registry);
this.gpsProcessedCounter = Counter.builder("gps.data.processed")
.description("成功处理的GPS数据数量")
.register(registry);
this.gpsErrorCounter = Counter.builder("gps.data.errors")
.description("处理失败的GPS数据数量")
.register(registry);
}
public void incrementReceived() {
gpsReceivedCounter.increment();
}
public void incrementProcessed() {
gpsProcessedCounter.increment();
}
public void incrementError() {
gpsErrorCounter.increment();
}
}性能优化策略
1. 批量操作优化
@Service
public class BatchProcessingService {
private final List<GpsData> batchBuffer = new ArrayList<>();
private final int BATCH_SIZE = 1000;
private final long BATCH_TIMEOUT = 5000; // 5秒
@Scheduled(fixedDelay = BATCH_TIMEOUT)
public void batchProcess() {
if (batchBuffer.isEmpty()) return;
List<GpsData> currentBatch;
synchronized (batchBuffer) {
currentBatch = new ArrayList<>(batchBuffer);
batchBuffer.clear();
}
if (!currentBatch.isEmpty()) {
// 批量写入MongoDB
List<GpsTrackDocument> documents = currentBatch.stream()
.map(this::convertToDocument)
.collect(Collectors.toList());
mongoTemplate.insert(documents, GpsTrackDocument.class);
log.info("批量写入{}条GPS数据", documents.size());
}
}
public void addToBatch(GpsData data) {
synchronized (batchBuffer) {
batchBuffer.add(data);
if (batchBuffer.size() >= BATCH_SIZE) {
batchProcess();
}
}
}
}2. Redis Pipeline优化
@Service
public class RedisPipelineService {
@Autowired
private RedisTemplate<String, VehicleLocation> redisTemplate;
public void batchUpdateLocations(Map<String, VehicleLocation> locationMap) {
redisTemplate.executePipelined(new RedisCallback<Object>() {
@Override
public Object doInRedis(RedisConnection connection) throws DataAccessException {
for (Map.Entry<String, VehicleLocation> entry : locationMap.entrySet()) {
String key = buildRedisKey(entry.getKey());
byte[] keyBytes = redisTemplate.getStringSerializer().serialize(key);
byte[] valueBytes = redisTemplate.getValueSerializer().serialize(entry.getValue());
connection.setEx(keyBytes, 300, valueBytes); // 5分钟过期
}
return null;
}
});
}
}部署架构
# docker-compose.yml 简化版
version: '3.8'
services:
zookeeper:
image: confluentinc/cp-zookeeper:latest
environment:
ZOOKEEPER_CLIENT_PORT: 2181
kafka:
image: confluentinc/cp-kafka:latest
depends_on:
- zookeeper
environment:
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
redis:
image: redis:6.2-alpine
command: redis-server --appendonly yes
volumes:
- redis_data:/data
mongodb:
image: mongo:4.4
environment:
MONGO_INITDB_ROOT_USERNAME: admin
MONGO_INITDB_ROOT_PASSWORD: password
volumes:
- mongo_data:/data/db
gps-api:
image: gps-api:latest
environment:
SPRING_PROFILES_ACTIVE: prod
KAFKA_SERVERS: kafka:9092
REDIS_HOST: redis
MONGODB_URI: mongodb://admin:password@mongodb:27017
deploy:
replicas: 3
gps-processor:
image: gps-processor:latest
environment:
SPRING_PROFILES_ACTIVE: prod
deploy:
replicas: 4
volumes:
redis_data:
mongo_data:总结
本文详细介绍了基于SpringBoot + Redis + Kafka + MongoDB的高稳定车辆GPS数据处理系统，具备：
核心优势
- 高可用架构：多组件集群部署，故障自动转移
- 弹性扩展：水平扩展应对流量波动
- 数据可靠性：多级保障确保数据不丢失
- 实时性能：毫秒级位置更新，秒级轨迹查询
- 丰富的数据应用：实时查询、轨迹回放、地理围栏、统计分析
性能指标
- 数据处理能力：10,000+ TPS
- 位置查询延迟：< 50ms
- 系统可用性：99.9%
- 数据可靠性：99.99%
该系统已在多个智能交通项目中成功应用，为车辆监控、调度优化、安全预警等业务场景提供了坚实的数据基础。
相关推荐
- 如何设计一个优秀的电子商务产品详情页
-
加入人人都是产品经理【起点学院】产品经理实战训练营，BAT产品总监手把手带你学产品电子商务网站的产品详情页面无疑是设计师和开发人员关注的最重要的网页之一。产品详情页面是客户作出“加入购物车”决定的页面...
- 怎么在JS中使用Ajax进行异步请求?
-
大家好，今天我来分享一项JavaScript的实战技巧，即如何在JS中使用Ajax进行异步请求，让你的网页速度瞬间提升。Ajax是一种在不刷新整个网页的情况下与服务器进行数据交互的技术，可以实现异步加...
- 中小企业如何组建,管理团队_中小企业应当如何开展组织结构设计变革
-
前言写了太多关于产品的东西觉得应该换换口味.从码农到架构师,从前端到平面再到UI、UE,最后走向了产品这条不归路,其实以前一直再给你们讲.产品经理跟项目经理区别没有特别大,两个岗位之间有很...
- 前端监控 SDK 开发分享_前端监控系统 开源
-
一、前言随着前端的发展和被重视，慢慢的行业内对于前端监控系统的重视程度也在增加。这里不对为什么需要监控再做解释。那我们先直接说说需求。对于中小型公司来说，可以直接使用三方的监控，比如自己搭建一套免费的...
- Ajax 会被 fetch 取代吗?Axios 怎么办?
-
大家好，很高兴又见面了，我是"高级前端进阶"，由我带着大家一起关注前端前沿、深入前端底层技术，大家一起进步，也欢迎大家关注、点赞、收藏、转发！今天给大家带来的主题是ajax、fetch...
- 前端面试题《AJAX》_前端面试ajax考点汇总
-
1.什么是ajax？ajax作用是什么？AJAX=异步JavaScript和XML。AJAX是一种用于创建快速动态网页的技术。通过在后台与服务器进行少量数据交换,AJAX可以使网页实...
- Ajax 详细介绍_ajax
-
1、ajax是什么?asynchronousjavascriptandxml:异步的javascript和xml。ajax是用来改善用户体验的一种技术，其本质是利用浏览器内置的一个特殊的...
- 6款可替代dreamweaver的工具_替代powerdesigner的工具
-
dreamweaver对一个web前端工作者来说，再熟悉不过了，像我07年接触web前端开发就是用的dreamweaver，一直用到现在，身边的朋友有跟我推荐过各种更好用的可替代dreamweaver...
- 我敢保证，全网没有再比这更详细的Java知识点总结了，送你啊
-
接下来你看到的将是全网最详细的Java知识点总结，全文分为三大部分：Java基础、Java框架、Java+云数据小编将为大家仔细讲解每大部分里面的详细知识点，别眨眼，从小白到大佬、零基础到精通，你绝...
- 福斯《死侍》发布新剧照 "小贱贱"韦德被改造前造型曝光
-
时光网讯福斯出品的科幻片《死侍》今天发布新剧照，其中一张是较为罕见的死侍在被改造之前的剧照，其余两张剧照都是死侍在执行任务中的状态。据外媒推测，片方此时发布剧照，预计是为了给不久之后影片发布首款正式预...
- 2021年超详细的java学习路线总结—纯干货分享
-
本文整理了java开发的学习路线和相关的学习资源，非常适合零基础入门java的同学，希望大家在学习的时候，能够节省时间。纯干货，良心推荐！第一阶段：Java基础重点知识点：数据类型、核心语法、面向对象...
- 不用海淘，真黑五来到你身边:亚马逊15件热卖爆款推荐!
-
Fujifilm富士instaxMini8小黄人拍立得相机（黄色/蓝色）扫二维码进入购物页面黑五是入手一个轻巧可爱的拍立得相机的好时机，此款是mini8的小黄人特别版，除了颜色涂装成小黄人...
- 2025 年 Python 爬虫四大前沿技术:从异步到 AI
-
作为互联网大厂的后端Python爬虫开发，你是否也曾遇到过这些痛点：面对海量目标URL，单线程爬虫爬取一周还没完成任务；动态渲染的SPA页面，requests库返回的全是空白代码；好不容易...
- 最贱超级英雄《死侍》来了!_死侍超燃
-
死侍Deadpool(2016)导演:蒂姆·米勒编剧:略特·里斯/保罗·沃尼克主演:瑞恩·雷诺兹/莫蕾娜·巴卡林/吉娜·卡拉诺/艾德·斯克林/T·J·米勒类型:动作/...
- 停止javascript的ajax请求，取消axios请求，取消reactfetch请求
-
一、Ajax原生里可以通过XMLHttpRequest对象上的abort方法来中断ajax。注意abort方法不能阻止向服务器发送请求，只能停止当前ajax请求。停止javascript的ajax请求...
- 一周热门
- 最近发表
- 标签列表
-
- HTML 简介 (30)
- HTML 响应式设计 (31)
- HTML URL 编码 (32)
- HTML Web 服务器 (31)
- HTML 表单属性 (32)
- HTML 音频 (31)
- HTML5 支持 (33)
- HTML API (36)
- HTML 总结 (32)
- HTML 全局属性 (32)
- HTML 事件 (31)
- HTML 画布 (32)
- HTTP 方法 (30)
- 键盘快捷键 (30)
- CSS 语法 (35)
- CSS 轮廓宽度 (31)
- CSS 谷歌字体 (33)
- CSS 链接 (31)
- CSS 定位 (31)
- CSS 图片库 (32)
- CSS 图像精灵 (31)
- SVG 文本 (32)
- 时钟启动 (33)
- HTML 游戏 (34)
- JS Loop For (32)