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 | 116 ++++++++++++++++++++++++++++++++++++++--------------------
1 files changed, 76 insertions(+), 40 deletions(-)
diff --git a/src/main/java/com/zy/core/plugin/GslProcess.java b/src/main/java/com/zy/core/plugin/GslProcess.java
index 3edeffe..fe7e669 100644
--- a/src/main/java/com/zy/core/plugin/GslProcess.java
+++ b/src/main/java/com/zy/core/plugin/GslProcess.java
@@ -22,21 +22,29 @@
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.properties.SystemProperties;
import com.zy.core.thread.StationThread;
import com.zy.core.utils.CrnOperateProcessUtils;
import com.zy.core.utils.StationOperateProcessUtils;
-import com.zy.system.entity.Config;
+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;
@@ -53,52 +61,33 @@
@Autowired
private StationCommandDispatcher stationCommandDispatcher;
- private Thread crnRunThread = null;
+ 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() {
- if (crnRunThread == null) {
- crnRunThread = new Thread(() -> {
- while (!Thread.currentThread().isInterrupted()) {
- try {
- //鎵ц鍫嗗灈鏈轰换鍔�
- crnOperateUtils.crnIoExecute();
- //鍫嗗灈鏈轰换鍔℃墽琛屽畬鎴�
- crnOperateUtils.crnIoExecuteFinish();
- // 闂撮殧
- Thread.sleep(50);
- } catch (InterruptedException ie) {
- Thread.currentThread().interrupt();
- crnRunThread = null;
- break;
- } catch (Exception e) {
- e.printStackTrace();
- }
- }
- });
- crnRunThread.setName("crnRunProcess");
- crnRunThread.setDaemon(true);
- crnRunThread.start();
- }
-
//妫�娴嬪叆搴撶珯鏄惁鏈変换鍔＄敓鎴愶紝骞跺惎鍔ㄥ叆搴�
checkInStationHasTask();
//璇锋眰鐢熸垚鍏ュ簱浠诲姟
generateStoreWrkFile();
- //鎵ц杈撻�佺珯鐐瑰叆搴撲换鍔�
- stationOperateProcessUtils.stationInExecute();
- //鎵ц杈撻�佺珯鐐瑰嚭搴撲换鍔�
- stationOperateProcessUtils.crnStationOutExecute();
- // 妫�娴嬪嚭搴撴帓搴�
- stationOperateProcessUtils.checkStationOutOrder();
- // 鐩戞帶缁曞湀绔欑偣
- stationOperateProcessUtils.watchCircleStation();
-
- //妫�娴嬭緭閫佺珯鐐规槸鍚﹁繍琛屽牭濉�
- stationOperateProcessUtils.checkStationRunBlock();
- //妫�娴嬭緭閫佺珯鐐逛换鍔″仠鐣欒秴鏃跺悗閲嶆柊璁＄畻璺緞
- stationOperateProcessUtils.checkStationIdleRecover();
+ //鎵ц鍫嗗灈鏈轰换鍔�
+ crnOperateUtils.crnIoExecute();
+ //鍫嗗灈鏈轰换鍔℃墽琛屽畬鎴�
+ crnOperateUtils.crnIoExecuteFinish();
+ //杈撻�佺珯鐐归�昏緫鍒囧埌鍚庡彴涓茶鎵ц锛岄伩鍏嶉樆濉炰富娴佺▼閲岀殑鍫嗗灈鏈哄彂浠诲姟
+ 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);
}
/**
@@ -196,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