修复gis数据获取流程，去掉轨迹概念
This commit is contained in:
parent
ff9143acc7
commit
ee275f3868
@ -125,7 +125,6 @@ public class DataCollectorService {
|
||||
|
||||
// 用于缓存所有活跃的MovingObject的最新状态
|
||||
private final Map<String, MovingObject> activeMovingObjectsCache = new ConcurrentHashMap<>();
|
||||
private final Map<String, String> selectedTrackByFlightNo = new ConcurrentHashMap<>();
|
||||
|
||||
// 记录每个对象最近一次更新的时间戳，用于淘汰不活跃对象（防止内存长期增长）
|
||||
private final Map<String, Long> activeMovingObjectsLastUpdatedAtMs = new ConcurrentHashMap<>();
|
||||
@ -421,38 +420,28 @@ public class DataCollectorService {
|
||||
* 定时采集航空器数据
|
||||
* - 不进行数据持久化
|
||||
*/
|
||||
private List<Aircraft> deduplicateAircraftByLatestTimestamp(List<Aircraft> aircrafts) {
|
||||
private List<Aircraft> deduplicateAircraftByLatestFlightPoint(List<Aircraft> aircrafts) {
|
||||
if (aircrafts == null || aircrafts.isEmpty()) {
|
||||
return Collections.emptyList();
|
||||
}
|
||||
|
||||
Map<String, Map<String, Aircraft>> latestByFlightAndTrack = new HashMap<>();
|
||||
Map<String, Aircraft> 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<String, Aircraft> 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<Aircraft> selected = new java.util.ArrayList<>(latestByFlightAndTrack.size());
|
||||
for (Map.Entry<String, Map<String, Aircraft>> 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<String, Aircraft> 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<Aircraft> deduplicatedAircrafts = deduplicateAircraftByLatestTimestamp(newAircrafts);
|
||||
List<Aircraft> 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;
|
||||
|
||||
@ -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);
|
||||
}
|
||||
}
|
||||
|
||||
Loading…
Reference in New Issue
Block a user