From c7060ce32740d6a5d703e9c8c7f4a81de094642e Mon Sep 17 00:00:00 2001 From: sladro Date: Tue, 10 Feb 2026 22:10:53 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=8C=E6=88=90=E4=BA=86contactcross?= =?UTF-8?q?=E7=9A=84=E5=88=9D=E6=AD=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../collision/common/model/MovingObject.java | 5 ++ .../datacollector/dao/DataCollectorDao.java | 21 ++++- .../service/DataCollectorService.java | 51 ++++++++++- .../AdxpFlightServiceWebSocketClient.java | 9 +- .../service/DataProcessingService.java | 6 +- .../WebSocketMessageBroadcaster.java | 58 ++++++++++--- .../websocket/event/PositionUpdateEvent.java | 86 +++++++------------ .../message/PositionUpdatePayload.java | 47 +++++----- 8 files changed, 180 insertions(+), 103 deletions(-) diff --git a/qaup-collision/src/main/java/com/qaup/collision/common/model/MovingObject.java b/qaup-collision/src/main/java/com/qaup/collision/common/model/MovingObject.java index 9064208..b933d8c 100644 --- a/qaup-collision/src/main/java/com/qaup/collision/common/model/MovingObject.java +++ b/qaup-collision/src/main/java/com/qaup/collision/common/model/MovingObject.java @@ -47,6 +47,11 @@ public class MovingObject { */ private Double altitude; + /** + * Source timestamp from upstream payload, epoch milliseconds. + */ + private Long sourceTimestampMs; + /** * 对象类型枚举 - 四种基本分类 */ diff --git a/qaup-collision/src/main/java/com/qaup/collision/datacollector/dao/DataCollectorDao.java b/qaup-collision/src/main/java/com/qaup/collision/datacollector/dao/DataCollectorDao.java index 8b45872..9ba9f57 100644 --- a/qaup-collision/src/main/java/com/qaup/collision/datacollector/dao/DataCollectorDao.java +++ b/qaup-collision/src/main/java/com/qaup/collision/datacollector/dao/DataCollectorDao.java @@ -121,6 +121,7 @@ public class DataCollectorDao { aircraft.setCurrentSpeed(0.0); aircraft.setCurrentHeading(0.0); aircraft.setAltitude(0.0); // 假设默认高度为0，如果API有提供可以设置 + aircraft.setSourceTimestampMs(normalizeUpstreamTimestampToMs(rawData.getTime())); return aircraft; }).filter(java.util.Objects::nonNull).collect(Collectors.toList()); @@ -167,6 +168,7 @@ public class DataCollectorDao { vehicle.setCurrentSpeed(0.0); // 暂时设置为0.0，DataCollectorService会处理 vehicle.setCurrentHeading(0.0); vehicle.setAltitude(0.0); // 假设默认高度为0，如果API有提供可以设置 + vehicle.setSourceTimestampMs(normalizeUpstreamTimestampToMs(rawData.getTime())); return vehicle; }).filter(java.util.Objects::nonNull).collect(Collectors.toList()); @@ -183,6 +185,21 @@ public class DataCollectorDao { // 内部 DTOs，精确匹配外部 API 接口返回的 JSON 结构 + private Long normalizeUpstreamTimestampToMs(Double rawTime) { + if (rawTime == null || rawTime.isNaN() || rawTime.isInfinite()) { + return null; + } + + double absolute = Math.abs(rawTime); + if (absolute < 100_000_000_000d) { + return Math.round(rawTime * 1000d); + } + if (absolute < 100_000_000_000_000d) { + return Math.round(rawTime); + } + return Math.round(rawTime / 1000d); + } + @Data @NoArgsConstructor @AllArgsConstructor @@ -195,7 +212,7 @@ public class DataCollectorDao { @JsonProperty("latitude") private Double latitude; @JsonProperty("time") - private Long time; + private Double time; // 其他可能存在的但我们不关心的字段将被 @JsonIgnoreProperties 忽略 } @@ -211,7 +228,7 @@ public class DataCollectorDao { @JsonProperty("latitude") private Double latitude; @JsonProperty("time") - private Long time; + private Double time; } diff --git a/qaup-collision/src/main/java/com/qaup/collision/datacollector/service/DataCollectorService.java b/qaup-collision/src/main/java/com/qaup/collision/datacollector/service/DataCollectorService.java index fa20a77..58ac766 100644 --- a/qaup-collision/src/main/java/com/qaup/collision/datacollector/service/DataCollectorService.java +++ b/qaup-collision/src/main/java/com/qaup/collision/datacollector/service/DataCollectorService.java @@ -27,6 +27,8 @@ import org.springframework.stereotype.Service; import java.time.Instant; import java.time.LocalDateTime; import java.time.ZoneId; +import java.util.Collections; +import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Set; @@ -423,6 +425,42 @@ public class DataCollectorService { * 定时采集航空器数据 * - 不进行数据持久化 */ + private List deduplicateAircraftByLatestTimestamp(List aircrafts) { + if (aircrafts == null || aircrafts.isEmpty()) { + return Collections.emptyList(); + } + + Map latestByFlightNo = new HashMap<>(); + for (Aircraft aircraft : aircrafts) { + if (aircraft == null || aircraft.getObjectId() == null || aircraft.getObjectId().isBlank()) { + continue; + } + + Aircraft existing = latestByFlightNo.get(aircraft.getObjectId()); + if (existing == null || isCandidateNewer(existing, aircraft)) { + latestByFlightNo.put(aircraft.getObjectId(), aircraft); + } + } + + return List.copyOf(latestByFlightNo.values()); + } + + private boolean isCandidateNewer(Aircraft existing, Aircraft candidate) { + Long existingTs = existing.getSourceTimestampMs(); + Long candidateTs = candidate.getSourceTimestampMs(); + + if (existingTs == null && candidateTs == null) { + return true; + } + if (existingTs == null) { + return true; + } + if (candidateTs == null) { + return false; + } + return candidateTs >= existingTs; + } + @Scheduled(fixedRateString = "${data.collector.interval}") public void collectAircraftData() { if (collectorDisabled) { @@ -444,7 +482,13 @@ public class DataCollectorService { log.info("采集到 {} 条航空器数据，用于实时处理", newAircrafts.size()); - for (Aircraft aircraft : newAircrafts) { + List deduplicatedAircrafts = deduplicateAircraftByLatestTimestamp(newAircrafts); + if (deduplicatedAircrafts.size() != newAircrafts.size()) { + log.info("Aircraft batch deduplicated by latest source time: raw={}, deduped={}", + newAircrafts.size(), deduplicatedAircrafts.size()); + } + + for (Aircraft aircraft : deduplicatedAircrafts) { try { if (aircraft.getCurrentPosition() == null) { log.warn("航空器 {} 位置信息缺失，跳过处理。", aircraft.getObjectId()); @@ -464,6 +508,7 @@ public class DataCollectorService { .currentSpeed(null) // 不在采集阶段计算速度 .currentHeading(null) // 不在采集阶段计算方向 .altitude(aircraft.getAltitude()) + .sourceTimestampMs(aircraft.getSourceTimestampMs()) .build(); upsertActiveMovingObject(movingObject); @@ -479,7 +524,7 @@ public class DataCollectorService { } } - log.info("航空器数据处理和事件发布完成，处理数量: {}", newAircrafts.size()); + log.info("航空器数据处理和事件发布完成，处理数量: {}", deduplicatedAircrafts.size()); } catch (Exception e) { log.error("采集航空器数据异常", e); @@ -557,6 +602,7 @@ public class DataCollectorService { .currentSpeed(null) // 不在采集阶段计算速度 .currentHeading(null) // 不在采集阶段计算方向 .altitude(vehicle.getAltitude()) + .sourceTimestampMs(vehicle.getSourceTimestampMs()) .build(); // 将最新数据更新到缓存（不发送WebSocket消息，统一在周期性检测中发送） @@ -1295,4 +1341,3 @@ public class DataCollectorService { log.info("缓存已清理"); } } - diff --git a/qaup-collision/src/main/java/com/qaup/collision/datacollector/websocket/AdxpFlightServiceWebSocketClient.java b/qaup-collision/src/main/java/com/qaup/collision/datacollector/websocket/AdxpFlightServiceWebSocketClient.java index 6ecf681..b157476 100644 --- a/qaup-collision/src/main/java/com/qaup/collision/datacollector/websocket/AdxpFlightServiceWebSocketClient.java +++ b/qaup-collision/src/main/java/com/qaup/collision/datacollector/websocket/AdxpFlightServiceWebSocketClient.java @@ -559,9 +559,11 @@ public class AdxpFlightServiceWebSocketClient implements WebSocketHandler { // TISFLIGHT 缺 BizKey 时，先用 FlightId 关联 FuId（FuId 视作可复用业务键） String fuNorm = normalizeBizKey(fuIdRaw); String normalizedFlightId = normalizeFlightNo(flightId); - if (fuNorm != null && normalizedFlightId != null) { + String normalizedResolvedFlightNo = normalizeFlightNo(resolvedFlightNo); + String flightNoForMatch = normalizedFlightId != null ? normalizedFlightId : normalizedResolvedFlightNo; + if (fuNorm != null && flightNoForMatch != null) { String fuFlightNo = firstSegmentFromBizLike(fuNorm); - if (normalizedFlightId.equalsIgnoreCase(fuFlightNo)) { + if (flightNoForMatch.equalsIgnoreCase(fuFlightNo)) { return fuNorm; } } @@ -1058,6 +1060,9 @@ public class AdxpFlightServiceWebSocketClient implements WebSocketHandler { if (flightKey != null || bizRedisKey != null) { try { setValueOnFlightAndBizKeys(flightKey, bizRedisKey, "flightNumber", resolvedFlightNo); + if (normalizedBizKey != null) { + setValueOnFlightAndBizKeys(flightKey, bizRedisKey, "bizKey", normalizedBizKey); + } updateActiveBizKey(flightKey, normalizedBizKey, nowMs); // contactCross 仅对进港有效，避免进出港参数串用 if ("IN".equalsIgnoreCase(routeType) diff --git a/qaup-collision/src/main/java/com/qaup/collision/dataprocessing/service/DataProcessingService.java b/qaup-collision/src/main/java/com/qaup/collision/dataprocessing/service/DataProcessingService.java index f617855..bdc8f5b 100644 --- a/qaup-collision/src/main/java/com/qaup/collision/dataprocessing/service/DataProcessingService.java +++ b/qaup-collision/src/main/java/com/qaup/collision/dataprocessing/service/DataProcessingService.java @@ -297,13 +297,17 @@ public class DataProcessingService { finalHeading = 0.0; // 默认朝北 } + long payloadTimestamp = movingObject.getSourceTimestampMs() != null + ? movingObject.getSourceTimestampMs() + : currentTime; + PositionUpdatePayload payload = PositionUpdatePayload.builder() .objectId(movingObject.getObjectId()) .objectType(movingObject.getObjectType().name()) .position(positionPayload) .heading(finalHeading) .speed(new BigDecimal(finalSpeed).setScale(2, RoundingMode.HALF_UP).doubleValue()) - .timestamp(currentTime) + .timestamp(payloadTimestamp) .build(); // 发布WebSocket事件 diff --git a/qaup-collision/src/main/java/com/qaup/collision/websocket/broadcaster/WebSocketMessageBroadcaster.java b/qaup-collision/src/main/java/com/qaup/collision/websocket/broadcaster/WebSocketMessageBroadcaster.java index 80fc0fe..9aace8f 100644 --- a/qaup-collision/src/main/java/com/qaup/collision/websocket/broadcaster/WebSocketMessageBroadcaster.java +++ b/qaup-collision/src/main/java/com/qaup/collision/websocket/broadcaster/WebSocketMessageBroadcaster.java @@ -42,7 +42,6 @@ import java.util.concurrent.atomic.AtomicBoolean; @Component public class WebSocketMessageBroadcaster { - private static final int REACQUIRE_AFTER_REJECTS = 3; private static final double EARTH_RADIUS_METERS = 6371000.0; private static final double MIN_DT_SECONDS = 0.05; @@ -117,6 +116,9 @@ public class WebSocketMessageBroadcaster { @Value("${websocket.position.filter.state-ttl-ms:120000}") private long positionTrackStateTtlMs; + @Value("${websocket.position.filter.reacquire-gap-ms:8000}") + private long reacquireGapMs; + private record CoalescedPositionUpdate(PositionUpdatePayload payload, long timestamp) {} private static final class PositionTrackState { @@ -391,6 +393,7 @@ public class WebSocketMessageBroadcaster { } long payloadTimestamp = payload.getTimestamp() != null ? payload.getTimestamp() : System.currentTimeMillis(); + long filterTimestampMs = System.currentTimeMillis(); PositionTrackState state = positionTrackStates.computeIfAbsent(payload.getObjectId(), k -> new PositionTrackState()); double filteredLatitude = latitude; @@ -398,14 +401,14 @@ public class WebSocketMessageBroadcaster { synchronized (state) { if (!state.initialized) { - initializeTrackState(state, latitude, longitude, payloadTimestamp); + initializeTrackState(state, latitude, longitude, filterTimestampMs); } else { - if (payloadTimestamp <= state.lastPayloadTimestampMs) { + if (filterTimestampMs <= state.lastPayloadTimestampMs) { state.rejectedCount++; return null; } - long dtMs = payloadTimestamp - state.lastPayloadTimestampMs; + long dtMs = filterTimestampMs - state.lastPayloadTimestampMs; double dtSec = dtMs / 1000.0; double rawDistance = calculateDistanceMeters( state.lastRawLatitude, @@ -418,17 +421,27 @@ public class WebSocketMessageBroadcaster { double maxAllowedJump = resolveMaxAllowedJumpMeters(adaptive.maxSpeedMps(), adaptive.jumpMarginMeter(), dtSec); boolean isOutlierJump = rawDistance > maxAllowedJump; - if (isOutlierJump && state.rejectedCount < REACQUIRE_AFTER_REJECTS - 1) { - state.rejectedCount++; - log.debug("Drop outlier position update: objectId={}, jumpMeter={}, allowedMeter={}", - payload.getObjectId(), rawDistance, maxAllowedJump); - return null; - } if (isOutlierJump) { - state.rejectedCount = 0; - filteredLatitude = latitude; - filteredLongitude = longitude; + boolean longGapReacquire = dtMs >= Math.max(reacquireGapMs, 0L); + if (longGapReacquire) { + state.rejectedCount = 0; + filteredLatitude = latitude; + filteredLongitude = longitude; + } else { + state.rejectedCount++; + double[] clipped = clipPointByMaxDistance( + state.lastFilteredLatitude, + state.lastFilteredLongitude, + latitude, + longitude, + maxAllowedJump + ); + filteredLatitude = clipped[0]; + filteredLongitude = clipped[1]; + log.debug("Clip outlier position update: objectId={}, jumpMeter={}, allowedMeter={}, dtMs={}", + payload.getObjectId(), rawDistance, maxAllowedJump, dtMs); + } } else { state.rejectedCount = 0; if (rawDistance <= adaptive.jitterMeter()) { @@ -443,7 +456,7 @@ public class WebSocketMessageBroadcaster { state.lastRawLatitude = latitude; state.lastRawLongitude = longitude; - state.lastPayloadTimestampMs = payloadTimestamp; + state.lastPayloadTimestampMs = filterTimestampMs; state.lastFilteredLatitude = filteredLatitude; state.lastFilteredLongitude = filteredLongitude; state.lastAcceptedAtMs = System.currentTimeMillis(); @@ -556,4 +569,21 @@ public class WebSocketMessageBroadcaster { } return Math.min(value, max); } + + private double[] clipPointByMaxDistance( + double startLatitude, + double startLongitude, + double targetLatitude, + double targetLongitude, + double maxDistanceMeters) { + double distanceMeters = calculateDistanceMeters(startLatitude, startLongitude, targetLatitude, targetLongitude); + if (distanceMeters <= maxDistanceMeters || distanceMeters <= 0.0 || maxDistanceMeters <= 0.0) { + return new double[] {targetLatitude, targetLongitude}; + } + + double ratio = maxDistanceMeters / distanceMeters; + double clippedLatitude = startLatitude + (targetLatitude - startLatitude) * ratio; + double clippedLongitude = startLongitude + (targetLongitude - startLongitude) * ratio; + return new double[] {clippedLatitude, clippedLongitude}; + } } diff --git a/qaup-collision/src/main/java/com/qaup/collision/websocket/event/PositionUpdateEvent.java b/qaup-collision/src/main/java/com/qaup/collision/websocket/event/PositionUpdateEvent.java index 35963c5..481ab23 100644 --- a/qaup-collision/src/main/java/com/qaup/collision/websocket/event/PositionUpdateEvent.java +++ b/qaup-collision/src/main/java/com/qaup/collision/websocket/event/PositionUpdateEvent.java @@ -3,76 +3,59 @@ package com.qaup.collision.websocket.event; import com.qaup.collision.common.model.spatial.VehicleLocation; import com.qaup.collision.websocket.message.MessageTypeConstants; import com.qaup.collision.websocket.message.PositionUpdatePayload; - -import lombok.Getter; -import lombok.Data; -import lombok.Builder; -import lombok.NoArgsConstructor; import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.Getter; +import lombok.NoArgsConstructor; import java.util.UUID; /** - * 位置更新事件 - * 当车辆或航空器位置数据处理完成后发布此事件 + * Position update event. + * + * Outer event timestamp: + * - {@link #timestamp} is generated at publish time in epoch microseconds for envelope ordering. + * + * Payload timestamp: + * - {@link PositionUpdatePayload#getTimestamp()} is the sample time from data pipeline in epoch milliseconds. */ @Getter public class PositionUpdateEvent implements WebSocketEvent { - + private final String eventId; private final long timestamp; - private final Object payload; // 支持VehicleLocation和PositionUpdatePayload两种类型 - - // 支持VehicleLocation构造函数（用于无人车数据） + private final Object payload; + public PositionUpdateEvent(VehicleLocation vehicleLocation) { this.eventId = UUID.randomUUID().toString(); - this.timestamp = java.time.Instant.now().toEpochMilli() * 1000; // 微秒级绝对时间戳 + this.timestamp = java.time.Instant.now().toEpochMilli() * 1000; this.payload = vehicleLocation; } - - // 支持PositionUpdatePayload构造函数（用于航空器和机场车辆数据） + public PositionUpdateEvent(PositionUpdatePayload positionPayload) { this.eventId = UUID.randomUUID().toString(); - this.timestamp = java.time.Instant.now().toEpochMilli() * 1000; // 微秒级绝对时间戳 + this.timestamp = java.time.Instant.now().toEpochMilli() * 1000; this.payload = positionPayload; } - + @Override public String getEventType() { return MessageTypeConstants.POSITION_UPDATE; } - + @Override public Object getPayload() { return payload; } - - /** - * 获取车辆ID（索引性质） - * - * 按照API设计原则： - * - 对于无人车数据：返回数据库主键vehicleId - * - 对于航空器/机场车辆：返回null（因为它们没有数据库记录） - * - * @return 车辆ID（Long类型），不适用时返回null - */ + public Long getVehicleId() { if (payload instanceof VehicleLocation) { return ((VehicleLocation) payload).getVehicleId(); } - // 航空器和机场车辆没有数据库记录，返回null return null; } - - /** - * 获取业务标识符（详细信息） - * - * 按照API设计原则： - * - 对于无人车数据：返回车牌号 - * - 对于航空器/机场车辆：返回objectId（可能是车牌号、航班号等） - * - * @return 业务标识符（String类型） - */ + public String getLicensePlate() { if (payload instanceof VehicleLocation) { return ((VehicleLocation) payload).getLicensePlate(); @@ -81,12 +64,7 @@ public class PositionUpdateEvent implements WebSocketEvent { } return null; } - - /** - * 获取完整的标识符信息 - * - * @return 包含vehicleId和businessId的完整信息 - */ + public VehicleIdentifier getVehicleIdentifier() { if (payload instanceof VehicleLocation) { VehicleLocation location = (VehicleLocation) payload; @@ -98,17 +76,14 @@ public class PositionUpdateEvent implements WebSocketEvent { } else if (payload instanceof PositionUpdatePayload) { PositionUpdatePayload positionPayload = (PositionUpdatePayload) payload; return VehicleIdentifier.builder() - .vehicleId(null) // 航空器没有数据库记录 + .vehicleId(null) .businessId(positionPayload.getObjectId()) .vehicleType(positionPayload.getObjectType()) .build(); } return null; } - - /** - * 获取车辆类型 - */ + public String getVehicleType() { if (payload instanceof VehicleLocation) { return ((VehicleLocation) payload).getVehicleType().name(); @@ -117,17 +92,14 @@ public class PositionUpdateEvent implements WebSocketEvent { } return null; } - - /** - * 车辆标识符信息类 - */ + @Data @Builder @NoArgsConstructor @AllArgsConstructor public static class VehicleIdentifier { - private Long vehicleId; // 数据库主键（索引性质） - private String businessId; // 业务标识符（车牌号、航班号等） - private String vehicleType; // 车辆类型 + private Long vehicleId; + private String businessId; + private String vehicleType; } -} \ No newline at end of file +} diff --git a/qaup-collision/src/main/java/com/qaup/collision/websocket/message/PositionUpdatePayload.java b/qaup-collision/src/main/java/com/qaup/collision/websocket/message/PositionUpdatePayload.java index 75124a6..0f50c1b 100644 --- a/qaup-collision/src/main/java/com/qaup/collision/websocket/message/PositionUpdatePayload.java +++ b/qaup-collision/src/main/java/com/qaup/collision/websocket/message/PositionUpdatePayload.java @@ -1,76 +1,75 @@ package com.qaup.collision.websocket.message; import com.fasterxml.jackson.annotation.JsonProperty; -import lombok.Data; -import lombok.Builder; -import lombok.NoArgsConstructor; import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; /** - * 位置更新消息负载 - * 对应前端JSON格式: position_update + * Position update payload. */ @Data @Builder @NoArgsConstructor @AllArgsConstructor public class PositionUpdatePayload { - + /** - * 对象ID（车辆ID、航班号等） + * Object ID, such as vehicle ID or flight number. */ @JsonProperty("object_id") private String objectId; - + /** - * 对象类型（基于MovingObjectType枚举） + * Object type mapped from MovingObjectType. */ @JsonProperty("object_type") private String objectType; - + /** - * 位置信息 + * Current position. */ @JsonProperty("position") private Position position; - + /** - * 航向角（度） + * Heading in degrees. */ @JsonProperty("heading") private Double heading; - + /** - * 速度（米/秒） + * Speed value from backend (current behavior is km/h). */ @JsonProperty("speed") private Double speed; - + /** - * 时间戳（微秒级） + * Position sample timestamp in epoch milliseconds. */ @JsonProperty("timestamp") private Long timestamp; - + /** - * 位置坐标类 + * Coordinate model. */ @Data @Builder @NoArgsConstructor @AllArgsConstructor public static class Position { - + /** - * 纬度 + * Latitude. */ @JsonProperty("latitude") private Double latitude; - + /** - * 经度 + * Longitude. */ @JsonProperty("longitude") private Double longitude; } -} \ No newline at end of file +}