From 6aaa4bd87948dc09acbde1b647e62f1cc67a4bcb Mon Sep 17 00:00:00 2001
From: Junjie <fallin.jie@qq.com>
Date: 星期二, 31 三月 2026 19:51:21 +0800
Subject: [PATCH] #
---
src/main/java/com/zy/core/plugin/GslProcess.java | 97 +++++++++++++++++++++++++++++++++++++++---------
1 files changed, 79 insertions(+), 18 deletions(-)
diff --git a/src/main/java/com/zy/core/plugin/GslProcess.java b/src/main/java/com/zy/core/plugin/GslProcess.java
index 2f2f632..fe7e669 100644
--- a/src/main/java/com/zy/core/plugin/GslProcess.java
+++ b/src/main/java/com/zy/core/plugin/GslProcess.java
@@ -8,14 +8,13 @@
import com.zy.common.service.CommonService;
import com.zy.common.utils.RedisUtil;
import com.zy.core.News;
-import com.zy.core.cache.MessageQueue;
import com.zy.core.cache.SlaveConnection;
+import com.zy.core.dispatch.StationCommandDispatcher;
import com.zy.core.enums.RedisKeyType;
import com.zy.core.enums.SlaveType;
import com.zy.core.enums.StationCommandType;
import com.zy.core.enums.WrkIoType;
import com.zy.core.model.StationObjModel;
-import com.zy.core.model.Task;
import com.zy.core.model.command.StationCommand;
import com.zy.core.model.protocol.StationProtocol;
import com.zy.core.plugin.api.MainProcessPluginApi;
@@ -26,16 +25,26 @@
import com.zy.core.thread.StationThread;
import com.zy.core.utils.CrnOperateProcessUtils;
import com.zy.core.utils.StationOperateProcessUtils;
+import jakarta.annotation.PreDestroy;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
@Slf4j
@Component
public class GslProcess implements MainProcessPluginApi, StoreInTaskPolicy {
+ private static final long STATION_DISPATCH_INTERVAL_MS = 200L;
+ private static final long STATION_MAINTENANCE_INTERVAL_MS = 500L;
+ private static final long STATION_TASK_SLOW_LOG_THRESHOLD_MS = 1000L;
+ private static final long STATION_EXECUTOR_SHUTDOWN_TIMEOUT_SECONDS = 5L;
@Autowired
private CrnOperateProcessUtils crnOperateUtils;
@@ -49,6 +58,17 @@
private RedisUtil redisUtil;
@Autowired
private StoreInTaskGenerationService storeInTaskGenerationService;
+ @Autowired
+ private StationCommandDispatcher stationCommandDispatcher;
+
+ private final ExecutorService stationTaskExecutor = Executors.newSingleThreadExecutor(r -> {
+ Thread thread = new Thread(r);
+ thread.setName("StationTask");
+ thread.setDaemon(true);
+ return thread;
+ });
+ private final Map<String, AtomicBoolean> stationTaskRunningMap = new ConcurrentHashMap<>();
+ private final Map<String, Long> stationTaskLastSubmitTimeMap = new ConcurrentHashMap<>();
@Override
public void run() {
@@ -59,21 +79,15 @@
//鎵ц鍫嗗灈鏈轰换鍔�
crnOperateUtils.crnIoExecute();
- //鍫嗗灈鏈轰换鍔℃墽琛屽畬鎴�-鍏峰浠跨湡鑳藉姏
+ //鍫嗗灈鏈轰换鍔℃墽琛屽畬鎴�
crnOperateUtils.crnIoExecuteFinish();
- //鎵ц杈撻�佺珯鐐瑰叆搴撲换鍔�
- stationOperateProcessUtils.stationInExecute();
- //鎵ц杈撻�佺珯鐐瑰嚭搴撲换鍔�
- stationOperateProcessUtils.crnStationOutExecute();
- // 妫�娴嬪嚭搴撴帓搴�
- stationOperateProcessUtils.checkStationOutOrder();
- // 鐩戞帶缁曞湀绔欑偣
- stationOperateProcessUtils.watchCircleStation();
-
- //妫�娴嬭緭閫佺珯鐐规槸鍚﹁繍琛屽牭濉�
- stationOperateProcessUtils.checkStationRunBlock();
- //妫�娴嬭緭閫佺珯鐐逛换鍔″仠鐣欒秴鏃跺悗閲嶆柊璁＄畻璺緞
- stationOperateProcessUtils.checkStationIdleRecover();
+ //杈撻�佺珯鐐归�昏緫鍒囧埌鍚庡彴涓茶鎵ц锛岄伩鍏嶉樆濉炰富娴佺▼閲岀殑鍫嗗灈鏈哄彂浠诲姟
+ submitStationTask("stationInExecute", STATION_DISPATCH_INTERVAL_MS, stationOperateProcessUtils::stationInExecute);
+ submitStationTask("crnStationOutExecute", STATION_DISPATCH_INTERVAL_MS, stationOperateProcessUtils::crnStationOutExecute);
+ submitStationTask("checkStationOutOrder", STATION_MAINTENANCE_INTERVAL_MS, stationOperateProcessUtils::checkStationOutOrder);
+ submitStationTask("watchCircleStation", STATION_MAINTENANCE_INTERVAL_MS, stationOperateProcessUtils::watchCircleStation);
+ submitStationTask("checkStationRunBlock", STATION_MAINTENANCE_INTERVAL_MS, stationOperateProcessUtils::checkStationRunBlock);
+ submitStationTask("checkStationIdleRecover", STATION_MAINTENANCE_INTERVAL_MS, stationOperateProcessUtils::checkStationIdleRecover);
}
/**
@@ -115,7 +129,7 @@
News.taskInfo(stationProtocol.getTaskNo(), "{}宸ヤ綔,鑾峰彇杈撻�佺嚎鍛戒护澶辫触", stationProtocol.getTaskNo());
return false;
}
- MessageQueue.offer(SlaveType.Devp, context.getBasDevp().getDevpNo(), new Task(2, command));
+ stationCommandDispatcher.dispatch(context.getBasDevp().getDevpNo(), command, "gsl-process", "station-back");
News.taskInfo(stationProtocol.getTaskNo(), "{}鎵爜寮傚父锛屽凡閫�鍥炶嚦{}", backStation.getStationId());
redisUtil.set(RedisKeyType.GENERATE_STATION_BACK_LIMIT.key + stationProtocol.getStationId(), "lock", 10);
return true;
@@ -163,7 +177,7 @@
&& stationProtocol.isEnableIn()
) {
StationCommand command = stationThread.getCommand(StationCommandType.MOVE, commonService.getWorkNo(WrkIoType.ENABLE_IN.id), stationId, entity.getBarcodeStation().getStationId(), 0);
- MessageQueue.offer(SlaveType.Devp, basDevp.getDevpNo(), new Task(2, command));
+ stationCommandDispatcher.dispatch(basDevp.getDevpNo(), command, "gsl-process", "enable-in");
redisUtil.set(RedisKeyType.GENERATE_ENABLE_IN_STATION_DATA_LIMIT.key + stationId, "lock", 15);
News.info("{}绔欑偣鍚姩鍏ュ簱鎴愬姛锛屾暟鎹寘:{}", stationId, JSON.toJSONString(command));
}
@@ -171,4 +185,51 @@
}
}
+ private void submitStationTask(String taskName, long minIntervalMs, Runnable task) {
+ long now = System.currentTimeMillis();
+ Long lastSubmitTime = stationTaskLastSubmitTimeMap.get(taskName);
+ if (lastSubmitTime != null && now - lastSubmitTime < minIntervalMs) {
+ return;
+ }
+
+ AtomicBoolean running = stationTaskRunningMap.computeIfAbsent(taskName, key -> new AtomicBoolean(false));
+ if (!running.compareAndSet(false, true)) {
+ return;
+ }
+
+ stationTaskLastSubmitTimeMap.put(taskName, now);
+ try {
+ stationTaskExecutor.execute(() -> {
+ long startMs = System.currentTimeMillis();
+ try {
+ task.run();
+ } catch (Exception e) {
+ log.error("GslProcess station task {} execute error", taskName, e);
+ } finally {
+ long costMs = System.currentTimeMillis() - startMs;
+ if (costMs > STATION_TASK_SLOW_LOG_THRESHOLD_MS) {
+ log.warn("GslProcess station task {} executed slowly, cost={}ms", taskName, costMs);
+ }
+ running.set(false);
+ }
+ });
+ } catch (Exception e) {
+ running.set(false);
+ stationTaskLastSubmitTimeMap.remove(taskName);
+ log.error("GslProcess station task {} submit error", taskName, e);
+ }
+ }
+
+ @PreDestroy
+ public void shutDown() {
+ stationTaskExecutor.shutdownNow();
+ try {
+ if (!stationTaskExecutor.awaitTermination(STATION_EXECUTOR_SHUTDOWN_TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
+ log.warn("GslProcess station task executor did not terminate within {}s", STATION_EXECUTOR_SHUTDOWN_TIMEOUT_SECONDS);
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
}
--
Gitblit v1.9.1