#执行优化

Junjie

8 小时以前 41aeff86351d1dd94fe2408175f96475f227c1b9

 src/main/java/com/zy/core/plugin/FakeProcess.java

@@ -6,6 +6,7 @@
import com.core.common.Cools;
import com.zy.asrs.domain.param.CreateInTaskParam;
import com.zy.asrs.domain.param.CreateOutTaskParam;
import com.zy.asrs.utils.Utils;
import com.zy.asrs.entity.*;
import com.zy.asrs.service.*;
import com.zy.common.entity.FindCrnNoResult;
@@ -27,7 +28,8 @@
import com.zy.core.plugin.store.StoreInTaskContext;
import com.zy.core.plugin.store.StoreInTaskGenerationService;
import com.zy.core.plugin.store.StoreInTaskPolicy;
import com.zy.core.task.MainProcessAsyncTaskScheduler;
import com.zy.core.task.MainProcessLane;
import com.zy.core.task.MainProcessTaskSubmitter;
import com.zy.core.thread.CrnThread;
import com.zy.core.thread.DualCrnThread;
import com.zy.core.thread.StationThread;
@@ -46,14 +48,8 @@
@Slf4j
@Component
public class FakeProcess implements MainProcessPluginApi, StoreInTaskPolicy {
    private static final String CRN_TASK_LANE = "fake-crn";
    private static final String STATION_TASK_LANE = "fake-station";
    private static final String DUAL_CRN_TASK_LANE = "fake-dual-crn";
    private static final String GENERATE_STORE_TASK_LANE_PREFIX = "fake-generate-store-";
    private static final String ASYNC_TASK_LANE_PREFIX = "fake-async-";
    private static final long MAIN_DISPATCH_INTERVAL_MS = 200L;
    private static final long ASYNC_DISPATCH_INTERVAL_MS = 50L;
    private static final long TASK_SLOW_LOG_THRESHOLD_MS = 1000L;

    private static final Map<Integer, Long> stationStayTimeMap = new ConcurrentHashMap<>();
    private static volatile String enableFake = "N";
@@ -90,34 +86,43 @@
    @Autowired
    private StationCommandDispatcher stationCommandDispatcher;
    @Autowired
    private MainProcessAsyncTaskScheduler mainProcessAsyncTaskScheduler;
    private MainProcessTaskSubmitter mainProcessTaskSubmitter;

    @Override
    public void run() {
        long startTime = System.currentTimeMillis();
        refreshFakeConfig();

        // 仿真相关任务按各自 lane 串行提交，互不阻塞
        submitAsyncTask("checkInStationHasTask", ASYNC_DISPATCH_INTERVAL_MS, this::checkInStationHasTask);
        submitAsyncTask("generateFakeInTask", ASYNC_DISPATCH_INTERVAL_MS, this::generateFakeInTask);
        submitAsyncTask("generateFakeOutTask", ASYNC_DISPATCH_INTERVAL_MS, this::generateFakeOutTask);
        submitAsyncTask("calcAllStationStayTime", ASYNC_DISPATCH_INTERVAL_MS, this::calcAllStationStayTime);
        submitAsyncTask("checkOutStationStayTimeOut", ASYNC_DISPATCH_INTERVAL_MS, this::checkOutStationStayTimeOut);
        submitAsyncTask("checkInStationCrnTake", ASYNC_DISPATCH_INTERVAL_MS, this::checkInStationCrnTake);
        submitStationTask("checkStationRunBlock", ASYNC_DISPATCH_INTERVAL_MS, stationOperateProcessUtils::checkStationRunBlock);
        // 仿真异步任务
        submitAsyncTasks(ASYNC_DISPATCH_INTERVAL_MS);
        // 仿真输送线堵塞检测
        stationOperateProcessUtils.submitCheckStationRunBlockTasks(MainProcessLane.FAKE_STATION_RUN_BLOCK, ASYNC_DISPATCH_INTERVAL_MS);

        // 请求生成入库任务，按站点拆成独立串行通道，避免单站阻塞整轮扫描
        // 请求生成入库任务
        generateStoreWrkFile();

        // 堆垛机、输送线、双工位堆垛机任务分别进入各自串行通道逐个执行
        submitCrnTask("crnIoExecute", MAIN_DISPATCH_INTERVAL_MS, crnOperateUtils::crnIoExecute);
        submitCrnTask("crnIoExecuteFinish", MAIN_DISPATCH_INTERVAL_MS, this::crnIoExecuteFinish);
        submitStationTask("stationInExecute", MAIN_DISPATCH_INTERVAL_MS, stationOperateProcessUtils::stationInExecute);
        submitStationTask("crnStationOutExecute", MAIN_DISPATCH_INTERVAL_MS, stationOperateProcessUtils::crnStationOutExecute);
        submitStationTask("checkStationOutOrder", MAIN_DISPATCH_INTERVAL_MS, stationOperateProcessUtils::checkStationOutOrder);
        submitStationTask("watchCircleStation", MAIN_DISPATCH_INTERVAL_MS, stationOperateProcessUtils::watchCircleStation);
        submitDualCrnTask("dualCrnIoExecute", MAIN_DISPATCH_INTERVAL_MS, dualCrnOperateProcessUtils::dualCrnIoExecute);
        submitDualCrnTask("dualCrnIoExecuteFinish", MAIN_DISPATCH_INTERVAL_MS, dualCrnOperateProcessUtils::dualCrnIoExecuteFinish);
        // 执行堆垛机任务
        crnOperateUtils.submitCrnIoTasks(MainProcessLane.FAKE_CRN_IO, MAIN_DISPATCH_INTERVAL_MS);
        // 堆垛机任务执行完成
        submitCrnIoExecuteFinishTasks(MAIN_DISPATCH_INTERVAL_MS);
        // 执行输送站点入库任务
        stationOperateProcessUtils.submitStationInTasks(MainProcessLane.FAKE_STATION_IN, MAIN_DISPATCH_INTERVAL_MS);
        // 检测入库任务是否已经到达目标站台
        stationOperateProcessUtils.submitInboundStationArrivalTasks(MAIN_DISPATCH_INTERVAL_MS);
        // 输送线执行堆垛机出库后的站台流转
        stationOperateProcessUtils.submitCrnStationOutTasks(MainProcessLane.FAKE_STATION_OUT, MAIN_DISPATCH_INTERVAL_MS);
        // 检测出库任务是否已经到达目标站台
        stationOperateProcessUtils.submitStationOutExecuteFinishTasks(MAIN_DISPATCH_INTERVAL_MS);
        // 检测站台运行完成后的任务转完成
        stationOperateProcessUtils.submitCheckTaskToCompleteTasks(MAIN_DISPATCH_INTERVAL_MS);
        // 检测并处理出库排序
        stationOperateProcessUtils.submitCheckStationOutOrderTasks(MainProcessLane.FAKE_STATION_OUT_ORDER, MAIN_DISPATCH_INTERVAL_MS);
        // 监控输送线绕圈站点
        stationOperateProcessUtils.submitWatchCircleStationTasks(MainProcessLane.FAKE_STATION_WATCH_CIRCLE, MAIN_DISPATCH_INTERVAL_MS);
        // 执行双工位堆垛机任务
        dualCrnOperateProcessUtils.submitDualCrnIoTasks(MainProcessLane.FAKE_DUAL_CRN_IO, MAIN_DISPATCH_INTERVAL_MS);
        // 双工位堆垛机任务执行完成
        dualCrnOperateProcessUtils.submitDualCrnIoExecuteFinishTasks(MainProcessLane.FAKE_DUAL_CRN_IO_FINISH, MAIN_DISPATCH_INTERVAL_MS);

        News.info("[WCS Debug] 主线程Run执行完成,耗时:{}ms", System.currentTimeMillis() - startTime);
    }
@@ -147,20 +152,40 @@
        }
    }

    private void submitCrnTask(String taskName, long minIntervalMs, Runnable task) {
        submitProcessTask(CRN_TASK_LANE, taskName, minIntervalMs, task);
    }

    private void submitStationTask(String taskName, long minIntervalMs, Runnable task) {
        submitProcessTask(STATION_TASK_LANE, taskName, minIntervalMs, task);
    }

    private void submitDualCrnTask(String taskName, long minIntervalMs, Runnable task) {
        submitProcessTask(DUAL_CRN_TASK_LANE, taskName, minIntervalMs, task);
    private void submitAsyncTasks(long minIntervalMs) {
        submitAsyncTask("checkInStationHasTask", minIntervalMs, this::checkInStationHasTask);
        submitAsyncTask("generateFakeInTask", minIntervalMs, this::generateFakeInTask);
        submitAsyncTask("generateFakeOutTask", minIntervalMs, this::generateFakeOutTask);
        submitAsyncTask("calcAllStationStayTime", minIntervalMs, this::calcAllStationStayTime);
        submitAsyncTask("checkOutStationStayTimeOut", minIntervalMs, this::checkOutStationStayTimeOut);
        submitAsyncTask("checkInStationCrnTake", minIntervalMs, this::checkInStationCrnTake);
    }

    private void submitAsyncTask(String taskName, long minIntervalMs, Runnable task) {
        submitProcessTask(ASYNC_TASK_LANE_PREFIX + taskName, taskName, minIntervalMs, task);
        mainProcessTaskSubmitter.submitKeyedSerialTask(
                MainProcessLane.FAKE_ASYNC,
                taskName,
                taskName,
                minIntervalMs,
                task
        );
    }

    private void submitCrnIoExecuteFinishTasks(long minIntervalMs) {
        List<BasCrnp> basCrnps = basCrnpService.list(new QueryWrapper<>());
        for (BasCrnp basCrnp : basCrnps) {
            Integer crnNo = basCrnp == null ? null : basCrnp.getCrnNo();
            if (crnNo == null) {
                continue;
            }
            mainProcessTaskSubmitter.submitKeyedSerialTask(
                    MainProcessLane.FAKE_CRN_IO_FINISH,
                    crnNo,
                    "crnIoExecuteFinish",
                    minIntervalMs,
                    () -> crnIoExecuteFinish(basCrnp)
            );
        }
    }

    private void submitGenerateStoreTasks() {
@@ -172,24 +197,16 @@
                if (stationId == null) {
                    continue;
                }
                submitGenerateStoreTask(stationId, MAIN_DISPATCH_INTERVAL_MS,
                        () -> storeInTaskGenerationService.generate(this, basDevp, stationObjModel));
                storeInTaskGenerationService.submitGenerateStoreTask(
                        this,
                        basDevp,
                        stationObjModel,
                        MainProcessLane.FAKE_GENERATE_STORE,
                        MAIN_DISPATCH_INTERVAL_MS,
                        () -> storeInTaskGenerationService.generate(this, basDevp, stationObjModel)
                );
            }
        }
    }

    private void submitGenerateStoreTask(Integer stationId, long minIntervalMs, Runnable task) {
        submitProcessTask(GENERATE_STORE_TASK_LANE_PREFIX + stationId, "generateStoreWrkFile", minIntervalMs, task);
    }

    private void submitProcessTask(String laneName, String taskName, long minIntervalMs, Runnable task) {
        mainProcessAsyncTaskScheduler.submit(
                laneName,
                taskName,
                minIntervalMs,
                TASK_SLOW_LOG_THRESHOLD_MS,
                task
        );
    }

    // 检测入库站是否有任务生成，并仿真生成模拟入库站点数据
@@ -236,6 +253,9 @@
                            commonService.getWorkNo(WrkIoType.FAKE_TASK_NO.id), stationId,
                            entity.getBarcodeStation().getStationId(), 0);
                    stationCommandDispatcher.dispatch(basDevp.getDevpNo(), command, "fake-process", "fake-enable-in");
                    if (entity.getBarcodeStation() != null && entity.getBarcodeStation().getStationId() != null) {
                        Utils.precomputeInTaskEnableRow(entity.getBarcodeStation().getStationId());
                    }
                    redisUtil.set(RedisKeyType.GENERATE_FAKE_IN_STATION_DATA_LIMIT.key + stationId, "lock", 5);
                }
            }
@@ -629,84 +649,90 @@
    public void crnIoExecuteFinish() {
        List<BasCrnp> basCrnps = basCrnpService.list(new QueryWrapper<>());
        for (BasCrnp basCrnp : basCrnps) {
            CrnThread crnThread = (CrnThread) SlaveConnection.get(SlaveType.Crn, basCrnp.getCrnNo());
            if (crnThread == null) {
                continue;
            }

            CrnProtocol crnProtocol = crnThread.getStatus();
            if (crnProtocol == null) {
                continue;
            }

            if (crnProtocol.getMode() == CrnModeType.AUTO.id
                    && crnProtocol.getTaskNo() > 0
                    && crnProtocol.getStatus() == CrnStatusType.WAITING.id) {
                Object lock = redisUtil.get(RedisKeyType.CRN_IO_EXECUTE_FINISH_LIMIT.key + basCrnp.getCrnNo());
                if (lock != null) {
                    continue;
                }

                // 获取待确认工作档
                WrkMast wrkMast = wrkMastService.selectByWorkNo(crnProtocol.getTaskNo());
                if (wrkMast == null) {
                    News.error("堆垛机处于等待确认且任务完成状态，但未找到工作档。堆垛机号={}，工作号={}", basCrnp.getCrnNo(), crnProtocol.getTaskNo());
                    continue;
                }

                Long updateWrkSts = null;
                if (wrkMast.getWrkSts() == WrkStsType.INBOUND_RUN.sts) {
                    updateWrkSts = WrkStsType.COMPLETE_INBOUND.sts;
                } else if (wrkMast.getWrkSts() == WrkStsType.OUTBOUND_RUN.sts) {
                    updateWrkSts = WrkStsType.OUTBOUND_RUN_COMPLETE.sts;

                    // 生成仿真站点数据
                    List<StationObjModel> outStationList = basCrnp.getOutStationList$();
                    if (outStationList.isEmpty()) {
                        News.info("堆垛机:{} 出库站点未设置", basCrnp.getCrnNo());
                        continue;
                    }

                    for (StationObjModel stationObjModel : outStationList) {
                        if (!stationObjModel.getStationId().equals(wrkMast.getSourceStaNo())) {
                            continue;
                        }

                        StationThread stationThread = (StationThread) SlaveConnection.get(SlaveType.Devp,
                                stationObjModel.getDeviceNo());
                        if (stationThread == null) {
                            continue;
                        }
                        // 生成仿真站点数据
                        StationCommand command = stationThread.getCommand(StationCommandType.WRITE_INFO, 9998,
                                wrkMast.getSourceStaNo(), 0, 0);
                        stationCommandDispatcher.dispatch(stationObjModel.getDeviceNo(), command, "fake-process", "crn-out-complete-write-info");
                        redisUtil.set(RedisKeyType.CRN_OUT_TASK_COMPLETE_STATION_INFO.key + wrkMast.getWrkNo(), JSON.toJSONString(stationObjModel, SerializerFeature.DisableCircularReferenceDetect), 60 * 60 * 24);
                    }
                } else if (wrkMast.getWrkSts() == WrkStsType.LOC_MOVE_RUN.sts) {
                    updateWrkSts = WrkStsType.COMPLETE_LOC_MOVE.sts;
                } else if (wrkMast.getWrkSts() == WrkStsType.CRN_MOVE_RUN.sts) {
                    updateWrkSts = WrkStsType.COMPLETE_CRN_MOVE.sts;
                } else {
                    News.error("堆垛机处于等待确认且任务完成状态，但工作状态异常。堆垛机号={}，工作号={}", basCrnp.getCrnNo(), crnProtocol.getTaskNo());
                    continue;
                }

                Date now = new Date();
                wrkMast.setWrkSts(updateWrkSts);
                wrkMast.setSystemMsg("");
                wrkMast.setIoTime(now);
                wrkMast.setModiTime(now);
                if (wrkMastService.updateById(wrkMast)) {
                    wrkAnalysisService.markCraneComplete(wrkMast, now, updateWrkSts);
                    CrnCommand resetCommand = crnThread.getResetCommand(crnProtocol.getTaskNo(), crnProtocol.getCrnNo());
                    MessageQueue.offer(SlaveType.Crn, crnProtocol.getCrnNo(), new Task(2, resetCommand));
                    News.info("堆垛机任务状态更新成功，堆垛机号={}，工作号={}", basCrnp.getCrnNo(), crnProtocol.getTaskNo());
                }

                redisUtil.set(RedisKeyType.CRN_IO_EXECUTE_FINISH_LIMIT.key + basCrnp.getCrnNo(), "lock", 10);
            }
            crnIoExecuteFinish(basCrnp);
        }
    }

    private void crnIoExecuteFinish(BasCrnp basCrnp) {
        CrnThread crnThread = (CrnThread) SlaveConnection.get(SlaveType.Crn, basCrnp.getCrnNo());
        if (crnThread == null) {
            return;
        }

        CrnProtocol crnProtocol = crnThread.getStatus();
        if (crnProtocol == null) {
            return;
        }

        if (crnProtocol.getMode() != CrnModeType.AUTO.id
                || crnProtocol.getTaskNo() <= 0
                || crnProtocol.getStatus() != CrnStatusType.WAITING.id) {
            return;
        }

        Object lock = redisUtil.get(RedisKeyType.CRN_IO_EXECUTE_FINISH_LIMIT.key + basCrnp.getCrnNo());
        if (lock != null) {
            return;
        }

        // 获取待确认工作档
        WrkMast wrkMast = wrkMastService.selectByWorkNo(crnProtocol.getTaskNo());
        if (wrkMast == null) {
            News.error("堆垛机处于等待确认且任务完成状态，但未找到工作档。堆垛机号={}，工作号={}", basCrnp.getCrnNo(), crnProtocol.getTaskNo());
            return;
        }

        Long updateWrkSts = null;
        if (wrkMast.getWrkSts() == WrkStsType.INBOUND_RUN.sts) {
            updateWrkSts = WrkStsType.COMPLETE_INBOUND.sts;
        } else if (wrkMast.getWrkSts() == WrkStsType.OUTBOUND_RUN.sts) {
            updateWrkSts = WrkStsType.OUTBOUND_RUN_COMPLETE.sts;

            // 生成仿真站点数据
            List<StationObjModel> outStationList = basCrnp.getOutStationList$();
            if (outStationList.isEmpty()) {
                News.info("堆垛机:{} 出库站点未设置", basCrnp.getCrnNo());
                return;
            }

            for (StationObjModel stationObjModel : outStationList) {
                if (!stationObjModel.getStationId().equals(wrkMast.getSourceStaNo())) {
                    continue;
                }

                StationThread stationThread = (StationThread) SlaveConnection.get(SlaveType.Devp,
                        stationObjModel.getDeviceNo());
                if (stationThread == null) {
                    continue;
                }
                // 生成仿真站点数据
                StationCommand command = stationThread.getCommand(StationCommandType.WRITE_INFO, 9998,
                        wrkMast.getSourceStaNo(), 0, 0);
                stationCommandDispatcher.dispatch(stationObjModel.getDeviceNo(), command, "fake-process", "crn-out-complete-write-info");
                redisUtil.set(RedisKeyType.CRN_OUT_TASK_COMPLETE_STATION_INFO.key + wrkMast.getWrkNo(), JSON.toJSONString(stationObjModel, SerializerFeature.DisableCircularReferenceDetect), 60 * 60 * 24);
            }
        } else if (wrkMast.getWrkSts() == WrkStsType.LOC_MOVE_RUN.sts) {
            updateWrkSts = WrkStsType.COMPLETE_LOC_MOVE.sts;
        } else if (wrkMast.getWrkSts() == WrkStsType.CRN_MOVE_RUN.sts) {
            updateWrkSts = WrkStsType.COMPLETE_CRN_MOVE.sts;
        } else {
            News.error("堆垛机处于等待确认且任务完成状态，但工作状态异常。堆垛机号={}，工作号={}", basCrnp.getCrnNo(), crnProtocol.getTaskNo());
            return;
        }

        Date now = new Date();
        wrkMast.setWrkSts(updateWrkSts);
        wrkMast.setSystemMsg("");
        wrkMast.setIoTime(now);
        wrkMast.setModiTime(now);
        if (wrkMastService.updateById(wrkMast)) {
            wrkAnalysisService.markCraneComplete(wrkMast, now, updateWrkSts);
            CrnCommand resetCommand = crnThread.getResetCommand(crnProtocol.getTaskNo(), crnProtocol.getCrnNo());
            MessageQueue.offer(SlaveType.Crn, crnProtocol.getCrnNo(), new Task(2, resetCommand));
            News.info("堆垛机任务状态更新成功，堆垛机号={}，工作号={}", basCrnp.getCrnNo(), crnProtocol.getTaskNo());
        }

        redisUtil.set(RedisKeyType.CRN_IO_EXECUTE_FINISH_LIMIT.key + basCrnp.getCrnNo(), "lock", 10);
    }

}