#

Junjie

21 小时以前 26784989e73fc36c6315e54939d1b13a50eb5020

 src/main/java/com/zy/core/utils/StationOperateProcessUtils.java

@@ -382,10 +382,11 @@

                boolean complete = false;
                Integer targetDeviceNo = null;
                StationThread stationThread = null;
                BasStation basStation = basStationService.getOne(new QueryWrapper<BasStation>().eq("station_id", targetStaNo));
                if (basStation != null) {
                    targetDeviceNo = basStation.getDeviceNo();
                    StationThread stationThread = (StationThread) SlaveConnection.get(SlaveType.Devp, basStation.getDeviceNo());
                    stationThread = (StationThread) SlaveConnection.get(SlaveType.Devp, basStation.getDeviceNo());
                    if (stationThread != null) {
                        Map<Integer, StationProtocol> statusMap = stationThread.getStatusMap();
                        StationProtocol stationProtocol = statusMap.get(basStation.getStationId());
@@ -396,11 +397,26 @@
                }

                if (complete) {
                    attemptClearTaskPath(stationThread, wrkNo);
                    completeStationRunTask(wrkMast, targetDeviceNo);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private void attemptClearTaskPath(StationThread stationThread, Integer taskNo) {
        if (stationThread == null || taskNo == null || taskNo <= 0) {
            return;
        }
        try {
            boolean cleared = stationThread.clearPath(taskNo);
            if (cleared) {
                News.info("输送站点任务运行完成后清理残留路径，工作号={}", taskNo);
            }
        } catch (Exception e) {
            News.error("输送站点任务运行完成后清理残留路径异常，工作号={}", taskNo, e);
        }
    }

@@ -804,17 +820,19 @@
        if (taskNo == null || taskNo <= 0 || stationId == null) {
            return RerouteExecutionResult.skip("invalid-station-task");
        }
        boolean runBlockReroute = context.sceneType() == RerouteSceneType.RUN_BLOCK_REROUTE;
        if (runBlockReroute) {
            // 站点进入堵塞后，设备侧可能已经把之前预下发的分段命令清掉了。
            // 先作废本地 session/segment 状态，再按新路线重发，避免被旧状态反向卡住。
            if (context.cancelSessionBeforeDispatch() && stationMoveCoordinator != null) {
                stationMoveCoordinator.cancelSession(taskNo);
            }
            if (context.resetSegmentCommandsBeforeDispatch()) {
                resetSegmentMoveCommandsBeforeReroute(taskNo);
            }
        if (stationMoveCoordinator != null) {
            return stationMoveCoordinator.withTaskDispatchLock(taskNo,
                    () -> executeReroutePlanWithTaskLock(context, plan, stationProtocol, taskNo, stationId));
        }
        return executeReroutePlanWithTaskLock(context, plan, stationProtocol, taskNo, stationId);
    }

    private RerouteExecutionResult executeReroutePlanWithTaskLock(RerouteContext context,
                                                                  RerouteCommandPlan plan,
                                                                  StationProtocol stationProtocol,
                                                                  Integer taskNo,
                                                                  Integer stationId) {
        boolean runBlockReroute = context.sceneType() == RerouteSceneType.RUN_BLOCK_REROUTE;
        if (context.checkRecentDispatch()
                && shouldSkipIdleRecoverForRecentDispatch(taskNo, stationId)) {
            return RerouteExecutionResult.skip("recent-dispatch");
@@ -846,6 +864,22 @@
        if (context.requireOutOrderDispatchLock()
                && !tryAcquireOutOrderDispatchLock(taskNo, stationId)) {
            return RerouteExecutionResult.skip("out-order-lock");
        }

        if (context.cancelSessionBeforeDispatch() && stationMoveCoordinator != null) {
            // 切路前先把旧 session 置为 CANCEL_PENDING，让已经排队中的旧分段线程在最终发送前停下。
            stationMoveCoordinator.markCancelPending(taskNo, "reroute_pending");
        }

        if (runBlockReroute) {
            // 站点进入堵塞后，设备侧可能已经把之前预下发的分段命令清掉了。
            // 先作废本地 session/segment 状态，再按新路线重发，避免被旧状态反向卡住。
            if (context.cancelSessionBeforeDispatch() && stationMoveCoordinator != null) {
                stationMoveCoordinator.cancelSession(taskNo);
            }
            if (context.resetSegmentCommandsBeforeDispatch()) {
                resetSegmentMoveCommandsBeforeReroute(taskNo);
            }
        }

        if (!runBlockReroute
@@ -1863,28 +1897,53 @@
    }

    private boolean shouldSkipIdleRecoverForRecentDispatch(Integer taskNo, Integer stationId) {
        if (stationMoveCoordinator == null || taskNo == null || taskNo <= 0 || stationId == null) {
        if (taskNo == null || taskNo <= 0 || stationId == null) {
            return false;
        }
        StationMoveSession session = stationMoveCoordinator.loadSession(taskNo);
        if (session == null || !session.isActive() || session.getLastIssuedAt() == null) {
            return false;
        }
        if (!Objects.equals(stationId, session.getCurrentStationId())
                && !Objects.equals(stationId, session.getDispatchStationId())) {
            return false;
        }
        long elapsedMs = System.currentTimeMillis() - session.getLastIssuedAt();
        long thresholdMs = STATION_IDLE_RECOVER_SECONDS * 1000L;
        if (elapsedMs >= thresholdMs) {
        StationMoveSession session = stationMoveCoordinator == null ? null : stationMoveCoordinator.loadSession(taskNo);
        if (session != null && session.isActive() && session.getLastIssuedAt() != null) {
            // 分段执行过程中，刚下发下一段命令时，session 的 currentStationId/dispatchStationId
            // 可能还没来得及和当前观察站点完全对齐；只要当前站点仍在这条活动路线里，
            // 就说明这次 recent dispatch 仍然和它相关，idle recover 不应在 10 秒窗口内再次介入。
            if (Objects.equals(stationId, session.getCurrentStationId())
                    || Objects.equals(stationId, session.getDispatchStationId())
                    || session.containsStation(stationId)) {
                long elapsedMs = System.currentTimeMillis() - session.getLastIssuedAt();
                if (elapsedMs < thresholdMs) {
                    saveStationTaskIdleTrack(new StationTaskIdleTrack(taskNo, stationId, System.currentTimeMillis()));
                    News.info("输送站点任务刚完成命令下发，已跳过停留重算。站点号={}，工作号={}，距上次下发={}ms，routeVersion={}",
                            stationId, taskNo, elapsedMs, session.getRouteVersion());
                    return true;
                }
            }
        }
        if (!hasRecentIssuedMoveCommand(taskNo, stationId, thresholdMs)) {
            return false;
        }
        saveStationTaskIdleTrack(new StationTaskIdleTrack(taskNo, stationId, System.currentTimeMillis()));
        News.info("输送站点任务刚完成命令下发，已跳过停留重算。站点号={}，工作号={}，距上次下发={}ms，routeVersion={}",
                stationId, taskNo, elapsedMs, session.getRouteVersion());
        News.info("输送站点任务刚完成命令下发，已跳过停留重算。站点号={}，工作号={}，距最近命令下发<{}ms，routeVersion={}",
                stationId, taskNo, thresholdMs, session == null ? null : session.getRouteVersion());
        return true;
    }

    private boolean hasRecentIssuedMoveCommand(Integer taskNo, Integer stationId, long thresholdMs) {
        if (taskNo == null || taskNo <= 0 || stationId == null || thresholdMs <= 0L || basStationOptService == null) {
            return false;
        }
        Date thresholdTime = new Date(System.currentTimeMillis() - thresholdMs);
        List<BasStationOpt> optList = basStationOptService.list(new QueryWrapper<BasStationOpt>()
                .select("id")
                .eq("task_no", taskNo)
                .eq("station_id", stationId)
                .eq("mode", String.valueOf(StationCommandType.MOVE))
                .eq("send", 1)
                .ge("send_time", thresholdTime)
                .orderByDesc("send_time")
                .last("limit 1"));
        return optList != null && !optList.isEmpty();
    }

    private void resetSegmentMoveCommandsBeforeReroute(Integer taskNo) {
        if (redisUtil == null || taskNo == null || taskNo <= 0) {
            return;