#

Junjie

9 天以前 dc3f9cc91759823ce59486f19b138be4b296a0f1

 src/main/java/com/zy/common/utils/NavigateUtils.java

@@ -27,6 +27,7 @@
import com.zy.asrs.service.StationCycleCapacityService;
import com.zy.asrs.service.StationPathPolicyService;
import com.zy.core.News;
import org.springframework.scheduling.annotation.Scheduled;
import com.zy.core.model.StationObjModel;
import com.zy.core.model.command.StationCommand;
import com.zy.core.enums.StationCommandType;
@@ -55,7 +56,7 @@
public class NavigateUtils {

    private static final long STATION_PATH_SLOW_LOG_THRESHOLD_MS = 500L;
    private static final long STATION_PATH_RUNTIME_SNAPSHOT_TTL_MS = 200L;
    private static final long STATION_PATH_RUNTIME_SNAPSHOT_TTL_MS = 2000L;
    private static final double CONGESTION_BUSY_BASE = 1.0d;
    private static final double CONGESTION_ISSUED_RESERVE_BASE = 0.75d;
    private static final double CONGESTION_PENDING_QUEUE_BASE = 0.45d;
@@ -77,8 +78,23 @@
    @Autowired
    private StationTaskTraceRegistry stationTaskTraceRegistry;

    private final Object runtimeSnapshotLock = new Object();
    private volatile CachedStationPathRuntimeSnapshot cachedRuntimeSnapshot;

    private volatile CachedLoopMergeGuardContext cachedLoopMergeGuardContext;

    @Scheduled(fixedDelay = 1500, initialDelay = 3000)
    public void refreshStationPathCaches() {
        try {
            BaseRuntimeSnapshot snapshot = buildBaseRuntimeSnapshot(null);
            cachedRuntimeSnapshot = new CachedStationPathRuntimeSnapshot(System.currentTimeMillis(), snapshot);
        } catch (Exception ignore) {
        }
        try {
            LoopMergeGuardContext context = buildLoopMergeGuardContext();
            cachedLoopMergeGuardContext = new CachedLoopMergeGuardContext(System.currentTimeMillis(), context);
        } catch (Exception ignore) {
        }
    }

    public List<NavigateNode> calcOptimalPathByStationId(Integer startStationId,
                                                         Integer endStationId,
@@ -1651,15 +1667,25 @@
    }

    private StationPathRuntimeSnapshot loadStationPathRuntimeSnapshot(Integer currentTaskNo) {
        BaseRuntimeSnapshot baseRuntimeSnapshot = loadBaseRuntimeSnapshot();
        long totalStartNs = System.nanoTime();
        Map<String, Long> stepCostMap = new LinkedHashMap<>();

        long stepStartNs = System.nanoTime();
        BaseRuntimeSnapshot baseRuntimeSnapshot = loadBaseRuntimeSnapshot(stepCostMap);
        stepCostMap.put("loadBaseSnapshot", elapsedMillis(stepStartNs));
        if (baseRuntimeSnapshot == null) {
            logRuntimeSnapshotSlow(currentTaskNo, false, stepCostMap, totalStartNs);
            return StationPathRuntimeSnapshot.empty();
        }

        stepStartNs = System.nanoTime();
        StationTrafficSnapshot trafficSnapshot = buildStationTrafficSnapshot(
                baseRuntimeSnapshot.statusMap,
                currentTaskNo,
                baseRuntimeSnapshot.activeTraceList
        );
        stepCostMap.put("buildTrafficSnapshot", elapsedMillis(stepStartNs));
        logRuntimeSnapshotSlow(currentTaskNo, baseRuntimeSnapshot.cacheHit, stepCostMap, totalStartNs);
        return new StationPathRuntimeSnapshot(
                baseRuntimeSnapshot.statusMap,
                baseRuntimeSnapshot.stationLoopLoadMap,
@@ -1669,30 +1695,63 @@
        );
    }

    private BaseRuntimeSnapshot loadBaseRuntimeSnapshot() {
    private BaseRuntimeSnapshot loadBaseRuntimeSnapshot(Map<String, Long> stepCostMap) {
        long now = System.currentTimeMillis();
        CachedStationPathRuntimeSnapshot cachedSnapshot = cachedRuntimeSnapshot;
        if (cachedSnapshot != null && now - cachedSnapshot.cacheAtMs <= STATION_PATH_RUNTIME_SNAPSHOT_TTL_MS) {
            return cachedSnapshot.baseRuntimeSnapshot;
        }
        synchronized (runtimeSnapshotLock) {
            cachedSnapshot = cachedRuntimeSnapshot;
            if (cachedSnapshot != null && now - cachedSnapshot.cacheAtMs <= STATION_PATH_RUNTIME_SNAPSHOT_TTL_MS) {
                return cachedSnapshot.baseRuntimeSnapshot;
            if (stepCostMap != null) {
                stepCostMap.put("baseSnapshotCacheHit", 0L);
            }
            BaseRuntimeSnapshot baseRuntimeSnapshot = buildBaseRuntimeSnapshot();
            cachedRuntimeSnapshot = new CachedStationPathRuntimeSnapshot(now, baseRuntimeSnapshot);
            return baseRuntimeSnapshot;
            return cachedSnapshot.baseRuntimeSnapshot.toCacheHitSnapshot();
        }
        // 缓存过期：用过期缓存兜底（@Scheduled 定时任务会很快刷新），避免多线程阻塞
        if (cachedSnapshot != null) {
            if (stepCostMap != null) {
                stepCostMap.put("baseSnapshotStale", 0L);
            }
            return cachedSnapshot.baseRuntimeSnapshot.toCacheHitSnapshot();
        }
        // 极端情况：从未有过缓存（启动首次调用），同步构建
        long stepStartNs = System.nanoTime();
        BaseRuntimeSnapshot baseRuntimeSnapshot = buildBaseRuntimeSnapshot(stepCostMap);
        if (stepCostMap != null) {
            stepCostMap.put("buildBaseSnapshot", elapsedMillis(stepStartNs));
        }
        cachedRuntimeSnapshot = new CachedStationPathRuntimeSnapshot(System.currentTimeMillis(), baseRuntimeSnapshot);
        return baseRuntimeSnapshot;
    }

    private BaseRuntimeSnapshot buildBaseRuntimeSnapshot() {
    private BaseRuntimeSnapshot buildBaseRuntimeSnapshot(Map<String, Long> stepCostMap) {
        long stepStartNs = System.nanoTime();
        Map<Integer, StationProtocol> statusMap = Collections.unmodifiableMap(new LinkedHashMap<>(loadStationStatusMap()));
        if (stepCostMap != null) {
            stepCostMap.put("loadStatusMap", elapsedMillis(stepStartNs));
        }

        stepStartNs = System.nanoTime();
        Map<Integer, Double> stationLoopLoadMap = Collections.unmodifiableMap(new LinkedHashMap<>(loadStationLoopLoadMap()));
        if (stepCostMap != null) {
            stepCostMap.put("loadLoopLoadMap", elapsedMillis(stepStartNs));
        }

        stepStartNs = System.nanoTime();
        LoopMergeGuardContext loopMergeGuardContext = loadLoopMergeGuardContext();
        if (stepCostMap != null) {
            stepCostMap.put("loadLoopMergeGuard", elapsedMillis(stepStartNs));
        }

        stepStartNs = System.nanoTime();
        Set<Integer> outStationIdSet = Collections.unmodifiableSet(new LinkedHashSet<>(loadAllOutStationIdSet()));
        List<StationTaskTraceVo> activeTraceList = Collections.unmodifiableList(new ArrayList<>(loadPlanningActiveTraceList(statusMap)));
        return new BaseRuntimeSnapshot(statusMap, stationLoopLoadMap, loopMergeGuardContext, outStationIdSet, activeTraceList);
        if (stepCostMap != null) {
            stepCostMap.put("loadOutStationSet", elapsedMillis(stepStartNs));
        }

        stepStartNs = System.nanoTime();
        List<StationTaskTraceVo> activeTraceList = Collections.unmodifiableList(new ArrayList<>(loadPlanningActiveTraceList(statusMap, stepCostMap)));
        if (stepCostMap != null) {
            stepCostMap.put("loadActiveTraceList", elapsedMillis(stepStartNs));
        }
        return new BaseRuntimeSnapshot(false, statusMap, stationLoopLoadMap, loopMergeGuardContext, outStationIdSet, activeTraceList);
    }

    private Map<Integer, Double> loadStationLoopLoadMap() {
@@ -1722,6 +1781,22 @@
    }

    private LoopMergeGuardContext loadLoopMergeGuardContext() {
        long now = System.currentTimeMillis();
        CachedLoopMergeGuardContext cached = cachedLoopMergeGuardContext;
        if (cached != null && now - cached.cacheAtMs <= STATION_PATH_RUNTIME_SNAPSHOT_TTL_MS) {
            return new LoopMergeGuardContext(cached.context);
        }
        // 缓存过期：用过期缓存兜底（@Scheduled 定时任务会很快刷新）
        if (cached != null) {
            return new LoopMergeGuardContext(cached.context);
        }
        // 极端情况：从未有过缓存，同步构建
        LoopMergeGuardContext context = buildLoopMergeGuardContext();
        cachedLoopMergeGuardContext = new CachedLoopMergeGuardContext(System.currentTimeMillis(), context);
        return new LoopMergeGuardContext(context);
    }

    private LoopMergeGuardContext buildLoopMergeGuardContext() {
        LoopMergeGuardContext context = new LoopMergeGuardContext();
        try {
            if (stationCycleCapacityService == null) {
@@ -2094,10 +2169,19 @@
    }

    private List<StationTaskTraceVo> loadPlanningActiveTraceList(Map<Integer, StationProtocol> statusMap) {
        return loadPlanningActiveTraceList(statusMap, null);
    }

    private List<StationTaskTraceVo> loadPlanningActiveTraceList(Map<Integer, StationProtocol> statusMap,
                                                                   Map<String, Long> stepCostMap) {
        Map<Integer, StationTaskTraceVo> traceMap = new LinkedHashMap<>();
        if (stationTaskTraceRegistry != null) {
            try {
                List<StationTaskTraceVo> traceList = stationTaskTraceRegistry.listLatestTraces();
                long stepStartNs = System.nanoTime();
                List<StationTaskTraceVo> traceList = stationTaskTraceRegistry.listPlanningActiveTraceSnapshots();
                if (stepCostMap != null) {
                    stepCostMap.put("loadRegistryTraces", elapsedMillis(stepStartNs));
                }
                if (traceList != null) {
                    for (StationTaskTraceVo traceVo : traceList) {
                        if (!isPlanningActiveTrace(traceVo)) {
@@ -2111,7 +2195,11 @@
            } catch (Exception ignore) {
            }
        }
        Map<Integer, StationTaskTraceVo> fallbackTraceMap = loadFallbackActiveTraceMap(null, statusMap, traceMap.keySet());
        long stepStartNs = System.nanoTime();
        Map<Integer, StationTaskTraceVo> fallbackTraceMap = buildProtocolFallbackTraceMap(statusMap, traceMap.keySet());
        if (stepCostMap != null) {
            stepCostMap.put("loadFallbackTraces", elapsedMillis(stepStartNs));
        }
        if (!fallbackTraceMap.isEmpty()) {
            traceMap.putAll(fallbackTraceMap);
        }
@@ -2126,6 +2214,55 @@
        return StationTaskTraceRegistry.STATUS_WAITING.equals(status)
                || StationTaskTraceRegistry.STATUS_RUNNING.equals(status)
                || StationTaskTraceRegistry.STATUS_REROUTED.equals(status);
    }

    private Map<Integer, StationTaskTraceVo> buildProtocolFallbackTraceMap(Map<Integer, StationProtocol> statusMap,
                                                                            Set<Integer> existingTaskNoSet) {
        if (statusMap == null || statusMap.isEmpty()) {
            return Collections.emptyMap();
        }
        Map<Integer, StationTaskTraceVo> result = new LinkedHashMap<>();
        for (StationProtocol protocol : statusMap.values()) {
            if (protocol == null || protocol.getTaskNo() == null || protocol.getTaskNo() <= 0) {
                continue;
            }
            if (!Boolean.TRUE.equals(protocol.isLoading())) {
                continue;
            }
            if (existingTaskNoSet != null && existingTaskNoSet.contains(protocol.getTaskNo())) {
                continue;
            }
            Integer stationId = protocol.getStationId();
            Integer targetStaNo = protocol.getTargetStaNo();
            if (stationId == null || targetStaNo == null || stationId.equals(targetStaNo)) {
                continue;
            }
            List<Integer> path = new ArrayList<>();
            path.add(stationId);
            path.add(targetStaNo);
            List<Integer> pendingStationIds = new ArrayList<>();
            pendingStationIds.add(targetStaNo);
            StationTaskTraceVo vo = new StationTaskTraceVo();
            vo.setTaskNo(protocol.getTaskNo());
            vo.setThreadImpl("PROTOCOL_FALLBACK");
            vo.setStatus(StationTaskTraceRegistry.STATUS_RUNNING);
            vo.setTraceVersion(1);
            vo.setStartStationId(stationId);
            vo.setCurrentStationId(stationId);
            vo.setFinalTargetStationId(targetStaNo);
            vo.setFullPathStationIds(path);
            vo.setIssuedStationIds(path);
            vo.setPassedStationIds(Collections.emptyList());
            vo.setPendingStationIds(pendingStationIds);
            vo.setLatestIssuedSegmentPath(path);
            vo.setSegmentList(Collections.emptyList());
            vo.setIssuedSegmentCount(0);
            vo.setTotalSegmentCount(1);
            vo.setUpdatedAt(System.currentTimeMillis());
            vo.setEvents(Collections.emptyList());
            result.put(protocol.getTaskNo(), vo);
        }
        return result;
    }

    private Map<Integer, StationTaskTraceVo> loadFallbackActiveTraceMap(Integer currentTaskNo,
@@ -2156,7 +2293,7 @@
        }

        List<Integer> taskNoList = new ArrayList<>(activeTaskProtocolMap.keySet());
        int limit = Math.max(50, taskNoList.size() * 8);
        int limit = Math.max(taskNoList.size(), taskNoList.size() * 2);
        List<BasStationOpt> optList;
        try {
            optList = basStationOptService.list(new QueryWrapper<BasStationOpt>()
@@ -2719,6 +2856,21 @@
                totalCostMs);
    }

    private void logRuntimeSnapshotSlow(Integer currentTaskNo,
                                        boolean cacheHit,
                                        Map<String, Long> stepCostMap,
                                        long totalStartNs) {
        long totalCostMs = elapsedMillis(totalStartNs);
        if (totalCostMs < STATION_PATH_SLOW_LOG_THRESHOLD_MS) {
            return;
        }
        News.warn("站点路径运行时快照加载耗时较长，taskNo={}，cacheHit={}，stepCosts={}，totalCost={}ms",
                currentTaskNo,
                cacheHit,
                JSON.toJSONString(stepCostMap),
                totalCostMs);
    }

    private long elapsedMillis(long startNs) {
        long elapsedNs = System.nanoTime() - startNs;
        return elapsedNs <= 0L ? 0L : elapsedNs / 1_000_000L;
@@ -2921,22 +3073,34 @@
    }

    private static class BaseRuntimeSnapshot {
        private final boolean cacheHit;
        private final Map<Integer, StationProtocol> statusMap;
        private final Map<Integer, Double> stationLoopLoadMap;
        private final LoopMergeGuardContext loopMergeGuardContext;
        private final Set<Integer> outStationIdSet;
        private final List<StationTaskTraceVo> activeTraceList;

        private BaseRuntimeSnapshot(Map<Integer, StationProtocol> statusMap,
        private BaseRuntimeSnapshot(boolean cacheHit,
                                    Map<Integer, StationProtocol> statusMap,
                                    Map<Integer, Double> stationLoopLoadMap,
                                    LoopMergeGuardContext loopMergeGuardContext,
                                    Set<Integer> outStationIdSet,
                                    List<StationTaskTraceVo> activeTraceList) {
            this.cacheHit = cacheHit;
            this.statusMap = statusMap == null ? Collections.emptyMap() : statusMap;
            this.stationLoopLoadMap = stationLoopLoadMap == null ? Collections.emptyMap() : stationLoopLoadMap;
            this.loopMergeGuardContext = loopMergeGuardContext == null ? new LoopMergeGuardContext() : loopMergeGuardContext;
            this.outStationIdSet = outStationIdSet == null ? Collections.emptySet() : outStationIdSet;
            this.activeTraceList = activeTraceList == null ? Collections.emptyList() : activeTraceList;
        }

        private BaseRuntimeSnapshot toCacheHitSnapshot() {
            return new BaseRuntimeSnapshot(true,
                    statusMap,
                    stationLoopLoadMap,
                    loopMergeGuardContext,
                    outStationIdSet,
                    activeTraceList);
        }
    }

@@ -2947,6 +3111,16 @@
        private CachedStationPathRuntimeSnapshot(long cacheAtMs, BaseRuntimeSnapshot baseRuntimeSnapshot) {
            this.cacheAtMs = cacheAtMs;
            this.baseRuntimeSnapshot = baseRuntimeSnapshot;
        }
    }

    private static class CachedLoopMergeGuardContext {
        private final long cacheAtMs;
        private final LoopMergeGuardContext context;

        private CachedLoopMergeGuardContext(long cacheAtMs, LoopMergeGuardContext context) {
            this.cacheAtMs = cacheAtMs;
            this.context = context;
        }
    }

@@ -3003,6 +3177,16 @@
    private static class LoopMergeGuardContext {
        private final Set<Integer> loopStationIdSet = new HashSet<>();
        private final Map<Integer, Set<Integer>> loopNeighborMap = new HashMap<>();

        private LoopMergeGuardContext() {
        }

        private LoopMergeGuardContext(LoopMergeGuardContext source) {
            this.loopStationIdSet.addAll(source.loopStationIdSet);
            for (Map.Entry<Integer, Set<Integer>> entry : source.loopNeighborMap.entrySet()) {
                this.loopNeighborMap.put(entry.getKey(), new LinkedHashSet<>(entry.getValue()));
            }
        }
    }

    private static class LoopMergeEntry {