From ee275f38685dcfcf253b0bda3da6e6a2f2a4b037 Mon Sep 17 00:00:00 2001 From: sladro Date: Wed, 11 Feb 2026 14:55:04 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8Dgis=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E8=8E=B7=E5=8F=96=E6=B5=81=E7=A8=8B=EF=BC=8C=E5=8E=BB=E6=8E=89?= =?UTF-8?q?=E8=BD=A8=E8=BF=B9=E6=A6=82=E5=BF=B5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../service/DataCollectorService.java | 96 ++---------- .../WebSocketMessageBroadcaster.java | 148 ++++++++++++++---- 2 files changed, 130 insertions(+), 114 deletions(-) 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 11942cf..0e74c9f 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 @@ -125,7 +125,6 @@ public class DataCollectorService { // 用于缓存所有活跃的MovingObject的最新状态 private final Map activeMovingObjectsCache = new ConcurrentHashMap<>(); - private final Map selectedTrackByFlightNo = new ConcurrentHashMap<>(); // 记录每个对象最近一次更新的时间戳，用于淘汰不活跃对象（防止内存长期增长） private final Map activeMovingObjectsLastUpdatedAtMs = new ConcurrentHashMap<>(); @@ -421,38 +420,28 @@ public class DataCollectorService { * 定时采集航空器数据 * - 不进行数据持久化 */ - private List deduplicateAircraftByLatestTimestamp(List aircrafts) { + private List deduplicateAircraftByLatestFlightPoint(List aircrafts) { if (aircrafts == null || aircrafts.isEmpty()) { return Collections.emptyList(); } - Map> latestByFlightAndTrack = new HashMap<>(); + Map latestByFlight = new HashMap<>(); for (Aircraft aircraft : aircrafts) { if (aircraft == null || aircraft.getObjectId() == null || aircraft.getObjectId().isBlank()) { continue; } String flightNo = aircraft.getObjectId(); - String trackKey = normalizeTrackKey(aircraft.getTrackNumber()); - Map byTrack = latestByFlightAndTrack.computeIfAbsent(flightNo, k -> new HashMap<>()); - - Aircraft existing = byTrack.get(trackKey); - if (existing == null || isCandidateNewer(existing, aircraft)) { - byTrack.put(trackKey, aircraft); + Aircraft existing = latestByFlight.get(flightNo); + if (existing == null || isCandidateNewerBySourceTimestamp(existing, aircraft)) { + latestByFlight.put(flightNo, aircraft); } } - List selected = new java.util.ArrayList<>(latestByFlightAndTrack.size()); - for (Map.Entry> entry : latestByFlightAndTrack.entrySet()) { - Aircraft selectedAircraft = selectAircraftTrackCandidate(entry.getKey(), entry.getValue()); - if (selectedAircraft != null) { - selected.add(selectedAircraft); - } - } - return selected; + return new java.util.ArrayList<>(latestByFlight.values()); } - private boolean isCandidateNewer(Aircraft existing, Aircraft candidate) { + private boolean isCandidateNewerBySourceTimestamp(Aircraft existing, Aircraft candidate) { Long existingTs = existing.getSourceTimestampMs(); Long candidateTs = candidate.getSourceTimestampMs(); @@ -468,72 +457,6 @@ public class DataCollectorService { return candidateTs >= existingTs; } - private Aircraft selectAircraftTrackCandidate(String flightNo, Map byTrack) { - if (byTrack == null || byTrack.isEmpty()) { - return null; - } - if (byTrack.size() == 1) { - Aircraft only = byTrack.values().iterator().next(); - selectedTrackByFlightNo.put(flightNo, normalizeTrackKey(only.getTrackNumber())); - return only; - } - - String selectedTrack = selectedTrackByFlightNo.get(flightNo); - if (selectedTrack != null && byTrack.containsKey(selectedTrack)) { - return byTrack.get(selectedTrack); - } - - MovingObject previous = activeMovingObjectsCache.get(flightNo); - Aircraft best = null; - double bestDistance = Double.MAX_VALUE; - if (previous != null && previous.getCurrentPosition() != null) { - for (Aircraft candidate : byTrack.values()) { - if (candidate == null || candidate.getCurrentPosition() == null) { - continue; - } - double d = calculateDistanceMeters(previous.getCurrentPosition(), candidate.getCurrentPosition()); - if (d < bestDistance) { - best = candidate; - bestDistance = d; - } - } - } - - if (best == null) { - for (Aircraft candidate : byTrack.values()) { - if (best == null || isCandidateNewer(best, candidate)) { - best = candidate; - } - } - } - - if (best != null) { - selectedTrackByFlightNo.put(flightNo, normalizeTrackKey(best.getTrackNumber())); - } - return best; - } - - private String normalizeTrackKey(String trackNumber) { - if (trackNumber == null || trackNumber.isBlank()) { - return "_DEFAULT_TRACK_"; - } - return trackNumber.trim(); - } - - private double calculateDistanceMeters(Point from, Point to) { - double lat1 = Math.toRadians(from.getY()); - double lon1 = Math.toRadians(from.getX()); - double lat2 = Math.toRadians(to.getY()); - double lon2 = Math.toRadians(to.getX()); - - double dLat = lat2 - lat1; - double dLon = lon2 - lon1; - double a = Math.sin(dLat / 2.0) * Math.sin(dLat / 2.0) - + Math.cos(lat1) * Math.cos(lat2) * Math.sin(dLon / 2.0) * Math.sin(dLon / 2.0); - double c = 2.0 * Math.atan2(Math.sqrt(a), Math.sqrt(1.0 - a)); - return 6371000.0 * c; - } - // Fixed-delay scheduling prevents overlap without dropping cycles. @Scheduled(fixedDelayString = "${data.collector.interval}") public void collectAircraftData() { @@ -551,9 +474,9 @@ public class DataCollectorService { log.info("采集到 {} 条航空器数据，用于实时处理", newAircrafts.size()); - List deduplicatedAircrafts = deduplicateAircraftByLatestTimestamp(newAircrafts); + List deduplicatedAircrafts = deduplicateAircraftByLatestFlightPoint(newAircrafts); if (deduplicatedAircrafts.size() != newAircrafts.size()) { - log.info("Aircraft batch deduplicated by latest source time: raw={}, deduped={}", + log.info("Aircraft batch merged by flightNo latest source point: raw={}, merged={}", newAircrafts.size(), deduplicatedAircrafts.size()); } @@ -1325,7 +1248,6 @@ public class DataCollectorService { Long last = entry.getValue(); if (last == null || last < cutoff) { activeMovingObjectsCache.remove(entry.getKey()); - selectedTrackByFlightNo.remove(entry.getKey()); return true; } return false; 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 1f1c593..e27b284 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 @@ -45,6 +45,7 @@ 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; + private static final int AIRCRAFT_REACQUIRE_CONFIRM_POINTS = 2; private final MessageCacheService messageCacheService; private final CollisionWebSocketHandler collisionWebSocketHandler; @@ -117,6 +118,18 @@ public class WebSocketMessageBroadcaster { @Value("${websocket.position.filter.state-ttl-ms:120000}") private long positionTrackStateTtlMs; + @Value("${websocket.position.filter.aircraft.reacquire-near-factor:0.6}") + private double aircraftReacquireNearFactor; + + @Value("${websocket.position.filter.aircraft.max-step-factor:1.8}") + private double aircraftMaxStepFactor; + + @Value("${websocket.position.filter.max-effective-dt-ms:5000}") + private long maxEffectiveDtMs; + + @Value("${websocket.position.filter.aircraft.speed-cap-multiplier:1.3}") + private double aircraftSpeedCapMultiplier; + private record CoalescedPositionUpdate(PositionUpdatePayload payload, long timestamp) {} private static final class PositionTrackState { @@ -128,6 +141,10 @@ public class WebSocketMessageBroadcaster { private double lastFilteredLatitude; private double lastFilteredLongitude; private int rejectedCount; + private boolean aircraftPendingReacquire; + private int aircraftReacquireCount; + private double aircraftPendingLatitude; + private double aircraftPendingLongitude; } private enum MotionProfile { @@ -430,8 +447,12 @@ public class WebSocketMessageBroadcaster { } } - long dtMs = filterTimestampMs - state.lastPayloadTimestampMs; - double dtSec = dtMs / 1000.0; + long sourceDtMs = filterTimestampMs - state.lastPayloadTimestampMs; + long arrivalDtMs = System.currentTimeMillis() - state.lastAcceptedAtMs; + long boundedArrivalDtMs = clampLong(arrivalDtMs, 0L, Math.max(maxEffectiveDtMs, 50L)); + long boundedSourceDtMs = clampLong(sourceDtMs, 0L, Math.max(maxEffectiveDtMs, 50L)); + long effectiveDtMs = Math.max(boundedSourceDtMs, boundedArrivalDtMs); + double dtSec = Math.max(effectiveDtMs / 1000.0, MIN_DT_SECONDS); double rawDistance = calculateDistanceMeters( state.lastRawLatitude, state.lastRawLongitude, @@ -441,29 +462,83 @@ public class WebSocketMessageBroadcaster { double rawSpeedMps = rawDistance / Math.max(dtSec, MIN_DT_SECONDS); AdaptiveFilterParams adaptive = resolveAdaptiveFilterParams(payload.getObjectType(), rawSpeedMps); boolean aircraft = isAircraftType(payload.getObjectType()); - - double maxAllowedJump = resolveMaxAllowedJumpMeters(adaptive.maxSpeedMps(), adaptive.jumpMarginMeter(), dtSec); + double effectiveSpeedCapMps = resolveEffectiveSpeedCapMps(payload, adaptive.maxSpeedMps(), aircraft); + double maxAllowedJump = resolveMaxAllowedJumpMeters(effectiveSpeedCapMps, adaptive.jumpMarginMeter(), dtSec); boolean isOutlierJump = rawDistance > maxAllowedJump; - if (isOutlierJump && !aircraft && state.rejectedCount < REACQUIRE_AFTER_REJECTS - 1) { - state.rejectedCount++; - logDroppedPosition( - "outlier_jump_rejected", - payload, - rawLatitude, - rawLongitude, - String.format("jumpMeter=%.3f,allowedMeter=%.3f,rejectedCount=%d", rawDistance, maxAllowedJump, state.rejectedCount) - ); - log.debug("Drop outlier position update: objectId={}, jumpMeter={}, allowedMeter={}", - payload.getObjectId(), rawDistance, maxAllowedJump); - return null; + if (isOutlierJump) { + if (aircraft) { + double nearThreshold = Math.max(maxAllowedJump * clamp(aircraftReacquireNearFactor, 0.2, 2.0), adaptive.jitterMeter()); + if (state.aircraftPendingReacquire) { + double pendingDistance = calculateDistanceMeters( + state.aircraftPendingLatitude, + state.aircraftPendingLongitude, + latitude, + longitude + ); + if (pendingDistance <= nearThreshold) { + state.aircraftReacquireCount++; + } else { + state.aircraftReacquireCount = 1; + state.aircraftPendingLatitude = latitude; + state.aircraftPendingLongitude = longitude; + } + } else { + state.aircraftPendingReacquire = true; + state.aircraftReacquireCount = 1; + state.aircraftPendingLatitude = latitude; + state.aircraftPendingLongitude = longitude; + } + + if (state.aircraftReacquireCount < AIRCRAFT_REACQUIRE_CONFIRM_POINTS) { + logDroppedPosition( + "aircraft_outlier_hold", + payload, + rawLatitude, + rawLongitude, + String.format("jumpMeter=%.3f,allowedMeter=%.3f,confirm=%d/%d", + rawDistance, maxAllowedJump, state.aircraftReacquireCount, AIRCRAFT_REACQUIRE_CONFIRM_POINTS) + ); + filteredLatitude = state.lastFilteredLatitude; + filteredLongitude = state.lastFilteredLongitude; + } else { + state.aircraftPendingReacquire = false; + state.aircraftReacquireCount = 0; + + double transitionMaxStep = Math.max(maxAllowedJump * clamp(aircraftMaxStepFactor, 1.0, 4.0), adaptive.jitterMeter() * 2.0); + if (rawDistance > transitionMaxStep) { + double ratio = transitionMaxStep / rawDistance; + filteredLatitude = state.lastFilteredLatitude + ratio * (latitude - state.lastFilteredLatitude); + filteredLongitude = state.lastFilteredLongitude + ratio * (longitude - state.lastFilteredLongitude); + log.warn("[TMP-POS-FIX] reason=aircraft_jump_step_limit objectId={} rawLatitude={} rawLongitude={} jumpMeter={} allowedStepMeter={}", + safeObjectId(payload), rawLatitude, rawLongitude, rawDistance, transitionMaxStep); + } else { + filteredLatitude = latitude; + filteredLongitude = longitude; + } + } + } else if (state.rejectedCount < REACQUIRE_AFTER_REJECTS - 1) { + state.rejectedCount++; + logDroppedPosition( + "outlier_jump_rejected", + payload, + rawLatitude, + rawLongitude, + String.format("jumpMeter=%.3f,allowedMeter=%.3f,rejectedCount=%d", rawDistance, maxAllowedJump, state.rejectedCount) + ); + log.debug("Drop outlier position update: objectId={}, jumpMeter={}, allowedMeter={}", + payload.getObjectId(), rawDistance, maxAllowedJump); + return null; + } else { + state.rejectedCount = 0; + filteredLatitude = latitude; + filteredLongitude = longitude; + } } - if (isOutlierJump) { - state.rejectedCount = 0; - filteredLatitude = latitude; - filteredLongitude = longitude; - } else { + if (!isOutlierJump) { state.rejectedCount = 0; + state.aircraftPendingReacquire = false; + state.aircraftReacquireCount = 0; if (aircraft) { if (rawDistance <= adaptive.jitterMeter()) { filteredLatitude = state.lastFilteredLatitude; @@ -485,12 +560,6 @@ public class WebSocketMessageBroadcaster { } } - if (isOutlierJump && aircraft) { - // Aircraft points should not be dropped to avoid visual path gaps. - filteredLatitude = latitude; - filteredLongitude = longitude; - } - state.lastRawLatitude = latitude; state.lastRawLongitude = longitude; state.lastPayloadTimestampMs = filterTimestampMs; @@ -536,6 +605,10 @@ public class WebSocketMessageBroadcaster { state.lastFilteredLatitude = latitude; state.lastFilteredLongitude = longitude; state.rejectedCount = 0; + state.aircraftPendingReacquire = false; + state.aircraftReacquireCount = 0; + state.aircraftPendingLatitude = latitude; + state.aircraftPendingLongitude = longitude; } private boolean isValidCoordinate(double latitude, double longitude) { @@ -641,4 +714,25 @@ public class WebSocketMessageBroadcaster { } return Math.min(value, max); } + + private long clampLong(long value, long min, long max) { + if (value < min) { + return min; + } + return Math.min(value, max); + } + + private double resolveEffectiveSpeedCapMps(PositionUpdatePayload payload, double adaptiveSpeedCapMps, boolean aircraft) { + double base = Math.max(adaptiveSpeedCapMps, 0.0); + if (!aircraft || payload == null || payload.getSpeed() == null) { + return base; + } + double speedKph = payload.getSpeed(); + if (Double.isNaN(speedKph) || Double.isInfinite(speedKph) || speedKph <= 0.0) { + return base; + } + double speedMps = speedKph / 3.6; + double multiplier = clamp(aircraftSpeedCapMultiplier, 1.0, 3.0); + return Math.max(base, speedMps * multiplier); + } }