package com.zy.asrs.task; import com.baomidou.mybatisplus.mapper.EntityWrapper; import com.core.common.Cools; import com.core.common.R; import com.zy.api.controller.params.WorkTaskParams; import com.zy.api.service.WcsApiService; import com.zy.asrs.entity.WrkMast; import com.zy.asrs.service.WrkMastService; import com.zy.asrs.task.core.ReturnT; import com.zy.asrs.task.handler.WorkMastHandler; import com.zy.asrs.task.support.OutboundBatchSeqReleaseGuard; import com.zy.asrs.task.support.WorkPublishLockKeys; import com.zy.asrs.utils.Utils; import com.zy.common.utils.RedisUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; import java.math.BigInteger; import java.util.ArrayList; import java.util.Arrays; import java.util.LinkedHashMap; import java.util.Date; import java.util.Map; import java.util.List; import java.util.Objects; /** * Created by vincent on 2020/7/7 */ @Component public class WorkMastScheduler { private static final Logger log = LoggerFactory.getLogger(WorkMastScheduler.class); private static final int MAX_PUBLISH_TASKS_ONCE = 20; private static final long OUTBOUND_USER_NO_LOCK_SECONDS = 60L; @Autowired private WcsApiService wcsApiService; @Autowired private WrkMastService wrkMastService; @Autowired private WorkMastHandler workMastHandler; @Autowired private OutboundBatchSeqReleaseGuard outboundBatchSeqReleaseGuard; @Autowired private RedisUtil redisUtil; @Scheduled(cron = "0/3 * * * * ? ") private void execute(){ List wrkMasts = wrkMastService.selectToBeCompleteData(); if (wrkMasts.isEmpty()) { return; } for (WrkMast wrkMast : wrkMasts) { ReturnT returnT = workMastHandler.start(wrkMast); if (!returnT.isSuccess()) { wrkMast.setUpdMk("X"); wrkMast.setErrorMemo(returnT.getMsg()); wrkMast.setErrorTime(new Date()); if (!wrkMastService.updateById(wrkMast)) { log.error("工作档[workNo={}]标记待处理失败", wrkMast.getWrkNo()); } } } } /** * 任务自动下发。 *

* 调度器只负责从工作档中挑出“当前允许下发”的任务,并将其转换成 WCS 接口需要的报文结构; * 出库任务按 userNo -> batchSeq 分层汇总后串行下发,确保同一 userNo 下前一个 batchSeq 完成后再发下一个。 *

* 当前批量下发的归并维度是: * 1. WCS接口路径(入库/出库/移库不能混发); * 2. 出库任务按 work_mast.user_no -> work_mast.batch_seq 分层汇总,并按 batchSeq 自然升序下发。 * * @author Ryan * @date 2026/1/10 14:42 */ @Scheduled(cron = "0/10 * * * * ? ") private synchronized void autoPubTasks() { // 仅处理待下发/已生成下发号的工作档。 List wrkMasts = wrkMastService.selectList(new EntityWrapper().in("wrk_sts", Arrays.asList(1L, 11L)) .orderBy("user_no", true) .orderBy("batch_seq", true) .orderBy("plt_type", true)); if (wrkMasts.isEmpty()) { return; } List paramsList = new ArrayList<>(); Map>> outboundTasksByUserNo = new LinkedHashMap<>(); for (WrkMast wrkMast : wrkMasts) { // 出库类任务(ioType > 100)默认需要 ERP 确认;未确认的任务在这里直接跳过。 if (wrkMast.getIoType() > 100 && !"Y".equalsIgnoreCase(wrkMast.getPdcType())) { continue; } if (isOutboundPublishTask(wrkMast)) { if (Cools.isEmpty(wrkMast.getBatchSeq())) { log.warn("出库进仓编号(batchSeq)为空,跳过下发, wrkNo={}, userNo={}", wrkMast.getWrkNo(), wrkMast.getUserNo()); continue; } WorkTaskParams params = buildWorkTaskParams(wrkMast); String userNo = normalizeGroupKey(wrkMast.getUserNo()); String batchSeq = normalizeGroupKey(wrkMast.getBatchSeq()); outboundTasksByUserNo .computeIfAbsent(userNo, key -> new LinkedHashMap<>()) .computeIfAbsent(batchSeq, key -> new ArrayList<>()) .add(params); } else { paramsList.add(buildWorkTaskParams(wrkMast)); } } if (publishTaskChunks(paramsList)) { return; } if (outboundTasksByUserNo.isEmpty()) { return; } for (Map.Entry>> userEntry : outboundTasksByUserNo.entrySet()) { String userNo = userEntry.getKey(); List batchSeqs = new ArrayList<>(userEntry.getValue().keySet()); batchSeqs.sort(this::compareBatchSeqNatural); for (String batchSeq : batchSeqs) { String blockMsg = outboundBatchSeqReleaseGuard.validateReady(userNo, batchSeq); if (!Cools.isEmpty(blockMsg)) { log.info(blockMsg); break; } List batchParams = userEntry.getValue().get(batchSeq); if (batchParams == null || batchParams.isEmpty()) { continue; } if (publishOutboundTaskChunks(userNo, batchSeq, batchParams)) { return; } } } } private WorkTaskParams buildWorkTaskParams(WrkMast wrkMast) { // WMS 库位编码转换成 WCS 可识别的库位编码。 String wcsSourceLocNo = Cools.isEmpty(wrkMast.getSourceLocNo()) ? "" : Utils.WMSLocToWCSLoc(wrkMast.getSourceLocNo()); String wcsLocNo = Cools.isEmpty(wrkMast.getLocNo()) ? "" : Utils.WMSLocToWCSLoc(wrkMast.getLocNo()); WorkTaskParams params = new WorkTaskParams(); // 101: 出库。有序任务才向 WCS 传 batch/batchSeq;seq=0 表示无序,不传这两个字段。 if (wrkMast.getIoType() == 101) { params.setType("out") .setTaskNo(wrkMast.getWrkNo() + "") .setLocNo(wcsSourceLocNo) .setStaNo(String.valueOf(wrkMast.getStaNo())) .setTaskPri(wrkMast.getIoPri().intValue()) .setBarcode(wrkMast.getBarcode()); if (wrkMast.getPltType() != null && wrkMast.getPltType() > 0) { params.setBatch(wrkMast.getBatchSeq()) .setBatchSeq(wrkMast.getPltType()); } // 2: 入库。入库接口使用 sourceStaNo + 目标库位。 } else if (wrkMast.getIoType() == 2 && !Cools.isEmpty(wrkMast.getSourceStaNo())) { params.setType("in") .setTaskNo(wrkMast.getWrkNo() + "") .setSourceStaNo(String.valueOf(wrkMast.getSourceStaNo())) .setLocNo(wcsLocNo) .setTaskPri(wrkMast.getIoPri().intValue()) .setBarcode(wrkMast.getBarcode()); // 其余走移库接口,源库位和目标库位都需要带给 WCS。 } else { params.setType("move") .setTaskNo(wrkMast.getWrkNo() + "") .setSourceLocNo(wcsSourceLocNo) .setLocNo(wcsLocNo) .setBarcode(wrkMast.getBarcode()); } return params; } private boolean isOutboundPublishTask(WrkMast wrkMast) { return wrkMast != null && Objects.equals(wrkMast.getIoType(), 101); } private int compareBatchSeqNatural(String left, String right) { String safeLeft = Cools.isEmpty(left) ? "" : left; String safeRight = Cools.isEmpty(right) ? "" : right; int leftIndex = 0; int rightIndex = 0; while (leftIndex < safeLeft.length() && rightIndex < safeRight.length()) { char leftChar = safeLeft.charAt(leftIndex); char rightChar = safeRight.charAt(rightIndex); if (Character.isDigit(leftChar) && Character.isDigit(rightChar)) { int leftStart = leftIndex; int rightStart = rightIndex; while (leftIndex < safeLeft.length() && Character.isDigit(safeLeft.charAt(leftIndex))) { leftIndex++; } while (rightIndex < safeRight.length() && Character.isDigit(safeRight.charAt(rightIndex))) { rightIndex++; } String leftNumber = safeLeft.substring(leftStart, leftIndex); String rightNumber = safeRight.substring(rightStart, rightIndex); int compare = new BigInteger(leftNumber).compareTo(new BigInteger(rightNumber)); if (compare != 0) { return compare; } compare = Integer.compare(leftNumber.length(), rightNumber.length()); if (compare != 0) { return compare; } continue; } int compare = Character.compare(leftChar, rightChar); if (compare != 0) { return compare; } leftIndex++; rightIndex++; } return Integer.compare(safeLeft.length(), safeRight.length()); } private String normalizeGroupKey(String value) { return Cools.isEmpty(value) ? "" : value; } private boolean publishTaskChunks(List paramsList) { if (paramsList == null || paramsList.isEmpty()) { return false; } for (int start = 0; start < paramsList.size(); start += MAX_PUBLISH_TASKS_ONCE) { int end = Math.min(start + MAX_PUBLISH_TASKS_ONCE, paramsList.size()); List chunk = paramsList.subList(start, end); R r = wcsApiService.pubWrksToWcs(chunk); if (isWcsSuccess(r)) { return true; } log.warn("批量下发任务到WCS失败, start={}, size={}, result={}", start, chunk.size(), r); } return false; } private boolean publishOutboundTaskChunks(String userNo, String batchSeq, List batchParams) { if (batchParams == null || batchParams.isEmpty()) { return false; } for (int start = 0; start < batchParams.size(); start += MAX_PUBLISH_TASKS_ONCE) { int end = Math.min(start + MAX_PUBLISH_TASKS_ONCE, batchParams.size()); List chunk = batchParams.subList(start, end); String lockKey = WorkPublishLockKeys.outboundUserNoLock(userNo); String lockValue = String.valueOf(System.currentTimeMillis()); if (!redisUtil.setIfAbsent(lockKey, lockValue, OUTBOUND_USER_NO_LOCK_SECONDS)) { log.info("出库任务正在下发,跳过本轮, userNo={}, batchSeq={}, lockKey={}", userNo, batchSeq, lockKey); return false; } try { R r = wcsApiService.pubWrksToWcs(chunk); if (isWcsSuccess(r)) { return true; } log.warn("批量下发出库任务到WCS失败, userNo={}, batchSeq={}, start={}, size={}, result={}", userNo, batchSeq, start, chunk.size(), r); } finally { Object currentLockValue = redisUtil.get(lockKey); if (Objects.equals(currentLockValue, lockValue)) { redisUtil.del(lockKey); } } } return false; } private boolean isWcsSuccess(R r) { return r != null && Objects.equals(r.get("code"), 200); } }