#

Junjie

9 天以前 dc3f9cc91759823ce59486f19b138be4b296a0f1

 src/main/java/com/zy/core/move/StationMoveCoordinator.java

@@ -5,6 +5,7 @@
import com.zy.common.utils.RedisUtil;
import com.zy.core.enums.RedisKeyType;
import com.zy.core.model.command.StationCommand;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@@ -12,15 +13,22 @@
import java.security.MessageDigest;
import java.util.ArrayList;
import java.util.Collections;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantLock;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.function.Supplier;

@Slf4j
@Component
public class StationMoveCoordinator {

    private static final int SESSION_EXPIRE_SECONDS = 60 * 60 * 24;
    private static final long TASK_DISPATCH_LOCK_SLOW_THRESHOLD_MS = 50L;
    private static final long RECORD_DISPATCH_SLOW_THRESHOLD_MS = 50L;
    private final Map<Integer, ReentrantLock> taskDispatchLocks = new ConcurrentHashMap<>();

    @Autowired
    private StationMoveSessionRegistry sessionRegistry;
@@ -36,6 +44,10 @@

    public boolean isActiveRoute(Integer taskNo, Integer routeVersion) {
        return sessionRegistry != null && sessionRegistry.isActiveRoute(taskNo, routeVersion);
    }

    public boolean canDispatchRoute(Integer taskNo, Integer routeVersion) {
        return sessionRegistry != null && sessionRegistry.canDispatchRoute(taskNo, routeVersion);
    }

    public void markSegmentIssued(Integer taskNo, Integer routeVersion) {
@@ -71,9 +83,33 @@
        if (session == null || !session.isActive()) {
            return;
        }
        log.info("markCancelPending, taskNo={}, routeVersion={}, cancelReason={}", taskNo, session.getRouteVersion(), cancelReason);
        session.setStatus(StationMoveSession.STATUS_CANCEL_PENDING);
        session.setCancelReason(cancelReason);
        saveSession(session);
    }

    public <T> T withTaskDispatchLock(Integer taskNo, Supplier<T> supplier) {
        if (supplier == null) {
            return null;
        }
        if (taskNo == null || taskNo <= 0) {
            return supplier.get();
        }
        ReentrantLock lock = taskDispatchLocks.computeIfAbsent(taskNo, key -> new ReentrantLock());
        long waitStartMs = System.currentTimeMillis();
        lock.lock();
        long lockWaitMs = System.currentTimeMillis() - waitStartMs;
        long holdStartMs = System.currentTimeMillis();
        try {
            return supplier.get();
        } finally {
            long holdMs = System.currentTimeMillis() - holdStartMs;
            if (lockWaitMs > TASK_DISPATCH_LOCK_SLOW_THRESHOLD_MS || holdMs > TASK_DISPATCH_LOCK_SLOW_THRESHOLD_MS) {
                log.warn("taskDispatchLock slow, taskNo={}, lockWaitMs={}ms, lockHoldMs={}ms", taskNo, lockWaitMs, holdMs);
            }
            lock.unlock();
        }
    }

    public boolean shouldSuppressDispatch(Integer taskNo, Integer currentStationId, StationCommand candidateCommand) {
@@ -87,14 +123,17 @@
        }

        String candidateSignature = buildPathSignature(candidateCommand);
        // 同 task、同当前位置、同路径签名的命令视为重复派发，直接压制。
        if (!isBlank(candidateSignature) && Objects.equals(candidateSignature, session.getPathSignature())) {
            return true;
        }

        // 到达下一决策站后允许重新决策，不继续受旧 session 的中间路径保护。
        if (Objects.equals(currentStationId, session.getNextDecisionStationId())) {
            return false;
        }

        // 还处在旧路线覆盖范围内时，其他触发源不应再插入一条新命令。
        return session.containsStation(currentStationId);
    }

@@ -107,6 +146,7 @@
            return null;
        }

        long startMs = System.currentTimeMillis();
        StationMoveSession current = loadSession(taskNo);
        long now = System.currentTimeMillis();
        String pathSignature = buildPathSignature(command);
@@ -117,6 +157,7 @@
                && Objects.equals(current.getNextDecisionStationId(), command.getTargetStaNo())
                && Objects.equals(current.getPathSignature(), pathSignature);

        // 同一触发站、同一目标、同一路径签名时复用当前 session，只刷新下发时间，不新开 routeVersion。
        StationMoveSession session = reuseCurrent ? current : new StationMoveSession();
        if (!reuseCurrent) {
            session.setRouteVersion(current == null || current.getRouteVersion() == null
@@ -145,6 +186,16 @@

        command.setRouteVersion(session.getRouteVersion());
        saveSession(session);
        long recordDispatchCostMs = System.currentTimeMillis() - startMs;
        log.info("recordDispatch done, taskNo={}, routeVersion={}, reuse={}, prevRouteVersion={}, dispatchStationId={}, triggerName={}, recordDispatchCostMs={}ms",
                taskNo, session.getRouteVersion(), reuseCurrent,
                current == null ? null : current.getRouteVersion(),
                dispatchStationId, triggerName, recordDispatchCostMs);
        if (recordDispatchCostMs > RECORD_DISPATCH_SLOW_THRESHOLD_MS) {
            log.warn("recordDispatch slow, taskNo={}, dispatchStationId={}, triggerName={}, recordDispatchCostMs={}ms, pathSize={}",
                    taskNo, dispatchStationId, triggerName, recordDispatchCostMs,
                    fullPathStationIds == null ? 0 : fullPathStationIds.size());
        }

        if (circleRoute) {
            saveLegacyCircleCommand(taskNo, command);
@@ -161,6 +212,7 @@
            clearLegacyCircleCommand(taskNo);
            return null;
        }
        log.info("cancelSession, taskNo={}, routeVersion={}, cancelReason=reroute_cancelled", taskNo, session.getRouteVersion());
        session.setStatus(StationMoveSession.STATUS_CANCELLED);
        session.setCancelReason("reroute_cancelled");
        saveSession(session);
@@ -212,8 +264,11 @@
        }
        StationMoveSession session = sessionRegistry.load(taskNo);
        if (session == null || !Objects.equals(session.getRouteVersion(), routeVersion)) {
            log.warn("updateTerminal skipped: session mismatch, taskNo={}, cmdRouteVersion={}, sessionRouteVersion={}, targetStatus={}, cancelReason={}",
                    taskNo, routeVersion, session == null ? null : session.getRouteVersion(), status, cancelReason);
            return;
        }
        log.info("updateTerminal, taskNo={}, routeVersion={}, status={}, cancelReason={}", taskNo, routeVersion, status, cancelReason);
        session.setCurrentStationId(currentStationId);
        session.setStatus(status);
        session.setCancelReason(cancelReason);