#

Junjie

9 天以前 dc3f9cc91759823ce59486f19b138be4b296a0f1

 src/main/java/com/zy/core/thread/impl/station/StationSegmentExecutor.java

@@ -6,6 +6,7 @@
import com.zy.asrs.domain.vo.StationTaskTraceSegmentVo;
import com.zy.asrs.entity.DeviceConfig;
import com.zy.common.utils.RedisUtil;
import lombok.extern.slf4j.Slf4j;
import com.zy.core.cache.SlaveConnection;
import com.zy.core.enums.RedisKeyType;
import com.zy.core.enums.SlaveType;
@@ -26,6 +27,7 @@
import java.util.Map;
import java.util.function.Function;

@Slf4j
public class StationSegmentExecutor {

    private static final String CFG_STATION_COMMAND_SEGMENT_ADVANCE_RATIO = "stationCommandSegmentAdvanceRatio";
@@ -239,15 +241,19 @@
                                         Integer currentStationId) {
        // 在下发新分段前检查路由版本是否仍然有效，避免在路由版本已更新的情况下下发旧版本命令
        if (!isRouteDispatchable(command == null ? null : command.getTaskNo(), command == null ? null : command.getRouteVersion())) {
            if (traceRegistry != null && command != null) {
                traceRegistry.markCancelled(command.getTaskNo(), traceVersion, currentStationId,
                        buildDetails("reason", "route_version_replaced_before_segment_send", "routeVersion", command.getRouteVersion()));
            // 首次校验失败可能是 Redis 写入延迟导致的，短暂等待后重试一次。
            sleepQuietly(50L);
            if (!isRouteDispatchable(command == null ? null : command.getTaskNo(), command == null ? null : command.getRouteVersion())) {
                if (traceRegistry != null && command != null) {
                    traceRegistry.markCancelled(command.getTaskNo(), traceVersion, currentStationId,
                            buildDetails("reason", "route_version_replaced_before_segment_send", "routeVersion", command.getRouteVersion()));
                }
                markCancelled(command == null ? null : command.getTaskNo(),
                        command == null ? null : command.getRouteVersion(),
                        currentStationId,
                        "route_version_replaced_before_segment_send");
                return false;
            }
            markCancelled(command == null ? null : command.getTaskNo(),
                    command == null ? null : command.getRouteVersion(),
                    currentStationId,
                    "route_version_replaced_before_segment_send");
            return false;
        }
        
        while (true) {
@@ -526,7 +532,15 @@
            return true;
        }
        StationMoveCoordinator moveCoordinator = loadMoveCoordinator();
        return moveCoordinator == null || moveCoordinator.canDispatchRoute(taskNo, routeVersion);
        if (moveCoordinator == null) {
            return true;
        }
        boolean dispatchable = moveCoordinator.canDispatchRoute(taskNo, routeVersion);
        if (!dispatchable) {
            log.warn("isRouteDispatchable rejected, taskNo={}, routeVersion={}, threadImpl={}",
                    taskNo, routeVersion, deviceConfig == null ? null : deviceConfig.getThreadImpl());
        }
        return dispatchable;
    }

    private StationMoveCoordinator loadMoveCoordinator() {