From 24681c81c09022f584a57006f2534b5f74723414 Mon Sep 17 00:00:00 2001
From: 云 <2163098428@qq.com>
Date: 星期二, 30 六月 2026 09:27:31 +0800
Subject: [PATCH] 初始化项目
---
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/common/manager/AbstractIotModbusPollScheduler.java | 137 +++++++++++++++++++++++++++++----------------
1 files changed, 88 insertions(+), 49 deletions(-)
diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/common/manager/AbstractIotModbusPollScheduler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/common/manager/AbstractIotModbusPollScheduler.java
index e62f85f..a41634b 100644
--- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/common/manager/AbstractIotModbusPollScheduler.java
+++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/common/manager/AbstractIotModbusPollScheduler.java
@@ -12,6 +12,7 @@
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
+import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertList;
import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertSet;
/**
@@ -19,6 +20,8 @@
* <p>
* 灏佽閫氱敤鐨勫畾鏃跺櫒绠＄悊銆乸er-device 璇锋眰闃熷垪闄愰�熼�昏緫銆�
* 瀛愮被鍙渶瀹炵幇 {@link #pollPoint(Long, Long)} 瀹氫箟鍏蜂綋鐨勮疆璇㈠姩浣溿��
+ * 濡傞渶灏嗗涓偣浣嶅悎骞朵负涓�涓疆璇换鍔★紝鍙鐩� {@link #buildPollTasks(IotModbusDeviceConfigRespDTO)}
+ * 鍜� {@link #pollTask(Long, String)}銆�
* <p>
*
* @author 鑺嬮亾婧愮爜
@@ -38,9 +41,9 @@
private static final int MAX_QUEUE_SIZE = 1000;
/**
- * 璁惧鐐逛綅鐨勫畾鏃跺櫒鏄犲皠锛歞eviceId -> (pointId -> PointTimerInfo)
+ * 璁惧杞浠诲姟鐨勫畾鏃跺櫒鏄犲皠锛歞eviceId -> (taskKey -> PollTimerInfo)
*/
- private final Map<Long, Map<Long, PointTimerInfo>> devicePointTimers = new ConcurrentHashMap<>();
+ private final Map<Long, Map<String, PollTimerInfo>> devicePollTimers = new ConcurrentHashMap<>();
/**
* per-device 璇锋眰闃熷垪锛歞eviceId -> 寰呮墽琛岃姹傞槦鍒�
@@ -60,11 +63,29 @@
}
/**
- * 鐐逛綅瀹氭椂鍣ㄤ俊鎭�
+ * 杞浠诲姟淇℃伅
*/
@Data
@AllArgsConstructor
- private static class PointTimerInfo {
+ protected static class PollTask {
+
+ /**
+ * 浠诲姟鏍囪瘑
+ */
+ private String key;
+ /**
+ * 杞闂撮殧锛堢敤浜庡垽鏂槸鍚﹂渶瑕佹洿鏂板畾鏃跺櫒锛�
+ */
+ private Integer pollInterval;
+
+ }
+
+ /**
+ * 杞瀹氭椂鍣ㄤ俊鎭�
+ */
+ @Data
+ @AllArgsConstructor
+ private static class PollTimerInfo {
/**
* Vert.x 瀹氭椂鍣� ID
@@ -82,75 +103,84 @@
/**
* 鏇存柊杞浠诲姟锛堝閲忔洿鏂帮級
*
- * 1. 銆愬垹闄ゃ�戠偣浣嶏細鍋滄瀵瑰簲鐨勮疆璇㈠畾鏃跺櫒
- * 2. 銆愭柊澧炪�戠偣浣嶏細鍒涘缓瀵瑰簲鐨勮疆璇㈠畾鏃跺櫒
- * 3. 銆愪慨鏀广�戠偣浣嶏細pollInterval 鍙樺寲锛岄噸寤哄搴旂殑杞瀹氭椂鍣�
- * 銆愪慨鏀广�戝叾浠栧睘鎬у彉鍖栵細涓嶉渶瑕侀噸寤哄畾鏃跺櫒锛坧ollPoint 杩愯鏃朵粠 configCache 鍙栨渶鏂� point锛�
+ * 1. 銆愬垹闄ゃ�戜换鍔★細鍋滄瀵瑰簲鐨勮疆璇㈠畾鏃跺櫒
+ * 2. 銆愭柊澧炪�戜换鍔★細鍒涘缓瀵瑰簲鐨勮疆璇㈠畾鏃跺櫒
+ * 3. 銆愪慨鏀广�戜换鍔★細pollInterval 鍙樺寲锛岄噸寤哄搴旂殑杞瀹氭椂鍣�
+ * 銆愪慨鏀广�戝叾浠栧睘鎬у彉鍖栵細涓嶉渶瑕侀噸寤哄畾鏃跺櫒锛坧ollTask 杩愯鏃朵粠 configCache 鍙栨渶鏂伴厤缃級
*/
public void updatePolling(IotModbusDeviceConfigRespDTO config) {
Long deviceId = config.getDeviceId();
- List<IotModbusPointRespDTO> newPoints = config.getPoints();
- Map<Long, PointTimerInfo> currentTimers = devicePointTimers
+ List<PollTask> newTasks = buildPollTasks(config);
+ Map<String, PollTimerInfo> currentTimers = devicePollTimers
.computeIfAbsent(deviceId, k -> new ConcurrentHashMap<>());
- // 1.1 璁＄畻鏂伴厤缃腑鐨勭偣浣� ID 闆嗗悎
- Set<Long> newPointIds = convertSet(newPoints, IotModbusPointRespDTO::getId);
- // 1.2 璁＄畻鍒犻櫎鐨勭偣浣� ID 闆嗗悎
- Set<Long> removedPointIds = new HashSet<>(currentTimers.keySet());
- removedPointIds.removeAll(newPointIds);
+ // 1.1 璁＄畻鏂伴厤缃腑鐨勪换鍔� Key 闆嗗悎
+ Set<String> newTaskKeys = convertSet(newTasks, PollTask::getKey);
+ // 1.2 璁＄畻鍒犻櫎鐨勪换鍔� Key 闆嗗悎
+ Set<String> removedTaskKeys = new HashSet<>(currentTimers.keySet());
+ removedTaskKeys.removeAll(newTaskKeys);
- // 2. 澶勭悊鍒犻櫎鐨勭偣浣嶏細鍋滄涓嶅啀瀛樺湪鐨勫畾鏃跺櫒
- for (Long pointId : removedPointIds) {
- PointTimerInfo timerInfo = currentTimers.remove(pointId);
+ // 2. 澶勭悊鍒犻櫎鐨勪换鍔★細鍋滄涓嶅啀瀛樺湪鐨勫畾鏃跺櫒
+ for (String taskKey : removedTaskKeys) {
+ PollTimerInfo timerInfo = currentTimers.remove(taskKey);
if (timerInfo != null) {
vertx.cancelTimer(timerInfo.getTimerId());
- log.debug("[updatePolling][璁惧 {} 鐐逛綅 {} 瀹氭椂鍣ㄥ凡鍒犻櫎]", deviceId, pointId);
+ log.debug("[updatePolling][璁惧 {} 杞浠诲姟 {} 瀹氭椂鍣ㄥ凡鍒犻櫎]", deviceId, taskKey);
}
}
- // 3. 澶勭悊鏂板鍜屼慨鏀圭殑鐐逛綅
- if (CollUtil.isEmpty(newPoints)) {
+ // 3. 澶勭悊鏂板鍜屼慨鏀圭殑浠诲姟
+ if (CollUtil.isEmpty(newTasks)) {
return;
}
- for (IotModbusPointRespDTO point : newPoints) {
- Long pointId = point.getId();
- Integer newPollInterval = point.getPollInterval();
- PointTimerInfo existingTimer = currentTimers.get(pointId);
- // 3.1 鏂板鐐逛綅锛氬垱寤哄畾鏃跺櫒
+ for (PollTask task : newTasks) {
+ String taskKey = task.getKey();
+ Integer newPollInterval = task.getPollInterval();
+ PollTimerInfo existingTimer = currentTimers.get(taskKey);
+ // 3.1 鏂板浠诲姟锛氬垱寤哄畾鏃跺櫒
if (existingTimer == null) {
- Long timerId = createPollTimer(deviceId, pointId, newPollInterval);
+ Long timerId = createPollTimer(deviceId, taskKey, newPollInterval);
if (timerId != null) {
- currentTimers.put(pointId, new PointTimerInfo(timerId, newPollInterval));
- log.debug("[updatePolling][璁惧 {} 鐐逛綅 {} 瀹氭椂鍣ㄥ凡鍒涘缓, interval={}ms]",
- deviceId, pointId, newPollInterval);
+ currentTimers.put(taskKey, new PollTimerInfo(timerId, newPollInterval));
+ log.debug("[updatePolling][璁惧 {} 杞浠诲姟 {} 瀹氭椂鍣ㄥ凡鍒涘缓, interval={}ms]",
+ deviceId, taskKey, newPollInterval);
}
} else if (!Objects.equals(existingTimer.getPollInterval(), newPollInterval)) {
// 3.2 pollInterval 鍙樺寲锛氶噸寤哄畾鏃跺櫒
vertx.cancelTimer(existingTimer.getTimerId());
- Long timerId = createPollTimer(deviceId, pointId, newPollInterval);
+ Long timerId = createPollTimer(deviceId, taskKey, newPollInterval);
if (timerId != null) {
- currentTimers.put(pointId, new PointTimerInfo(timerId, newPollInterval));
- log.debug("[updatePolling][璁惧 {} 鐐逛綅 {} 瀹氭椂鍣ㄥ凡鏇存柊, interval={}ms -> {}ms]",
- deviceId, pointId, existingTimer.getPollInterval(), newPollInterval);
+ currentTimers.put(taskKey, new PollTimerInfo(timerId, newPollInterval));
+ log.debug("[updatePolling][璁惧 {} 杞浠诲姟 {} 瀹氭椂鍣ㄥ凡鏇存柊, interval={}ms -> {}ms]",
+ deviceId, taskKey, existingTimer.getPollInterval(), newPollInterval);
} else {
- currentTimers.remove(pointId);
+ currentTimers.remove(taskKey);
}
}
- // 3.3 鍏朵粬灞炴�у彉鍖栵細鏃犻渶閲嶅缓瀹氭椂鍣紝鍥犱负 pollPoint() 杩愯鏃朵粠 configCache 鑾峰彇鏈�鏂� point锛岃嚜鍔ㄤ娇鐢ㄦ柊閰嶇疆
+ // 3.3 鍏朵粬灞炴�у彉鍖栵細鏃犻渶閲嶅缓瀹氭椂鍣紝鍥犱负 pollTask() 杩愯鏃朵粠 configCache 鑾峰彇鏈�鏂伴厤缃�
}
+ }
+
+ /**
+ * 鏋勫缓杞浠诲姟鍒楄〃
+ *
+ * 榛樿姣忎釜鐐逛綅涓�涓疆璇换鍔°�俆CP Client 绛夊崗璁彲瑕嗙洊璇ユ柟娉曪紝灏嗗涓偣浣嶅悎骞朵负涓�涓壒閲忚鍙栦换鍔°��
+ */
+ protected List<PollTask> buildPollTasks(IotModbusDeviceConfigRespDTO config) {
+ return convertList(config.getPoints(), point -> new PollTask(String.valueOf(point.getId()), point.getPollInterval()));
}
/**
* 鍒涘缓杞瀹氭椂鍣�
*/
- private Long createPollTimer(Long deviceId, Long pointId, Integer pollInterval) {
+ private Long createPollTimer(Long deviceId, String taskKey, Integer pollInterval) {
if (pollInterval == null || pollInterval <= 0) {
return null;
}
return vertx.setPeriodic(pollInterval, timerId -> {
try {
- submitPollRequest(deviceId, pointId);
+ submitPollRequest(deviceId, taskKey);
} catch (Exception e) {
- log.error("[createPollTimer][杞鐐逛綅澶辫触, deviceId={}, pointId={}]", deviceId, pointId, e);
+ log.error("[createPollTimer][杞浠诲姟澶辫触, deviceId={}, taskKey={}]", deviceId, taskKey, e);
}
});
}
@@ -160,7 +190,7 @@
/**
* 鎻愪氦杞璇锋眰鍒拌澶囪姹傞槦鍒楋紙淇濊瘉鍚岃澶囪姹傞棿闅旓級
*/
- private void submitPollRequest(Long deviceId, Long pointId) {
+ private void submitPollRequest(Long deviceId, String taskKey) {
// 1. 銆愰噸瑕併�戝皢璇锋眰娣诲姞鍒拌澶囩殑璇锋眰闃熷垪
Queue<Runnable> queue = deviceRequestQueues.computeIfAbsent(deviceId, k -> new ConcurrentLinkedQueue<>());
while (queue.size() >= MAX_QUEUE_SIZE) {
@@ -168,7 +198,7 @@
queue.poll();
log.warn("[submitPollRequest][璁惧 {} 璇锋眰闃熷垪宸叉弧({}), 涓㈠純鏈�鏃ц姹俔", deviceId, MAX_QUEUE_SIZE);
}
- queue.offer(() -> pollPoint(deviceId, pointId));
+ queue.offer(() -> pollTask(deviceId, taskKey));
// 2. 澶勭悊璁惧璇锋眰闃熷垪锛堝鏋滄病鏈夊欢杩� timer 鍦ㄧ瓑寰咃級
processDeviceQueue(deviceId);
@@ -239,6 +269,15 @@
// ========== 杞鎵ц ==========
/**
+ * 杞浠诲姟
+ *
+ * 榛樿灏嗕换鍔℃爣璇嗕綔涓虹偣浣� ID锛屾墽琛屽崟鐐硅疆璇€��
+ */
+ protected void pollTask(Long deviceId, String taskKey) {
+ pollPoint(deviceId, Long.valueOf(taskKey));
+ }
+
+ /**
* 杞鍗曚釜鐐逛綅锛堝瓙绫诲疄鐜板叿浣撶殑璇诲彇閫昏緫锛�
*
* @param deviceId 璁惧 ID
@@ -252,25 +291,25 @@
* 鍋滄璁惧鐨勮疆璇�
*/
public void stopPolling(Long deviceId) {
- Map<Long, PointTimerInfo> timers = devicePointTimers.remove(deviceId);
- if (CollUtil.isEmpty(timers)) {
- return;
- }
- for (PointTimerInfo timerInfo : timers.values()) {
- vertx.cancelTimer(timerInfo.getTimerId());
+ Map<String, PollTimerInfo> timers = devicePollTimers.remove(deviceId);
+ if (CollUtil.isNotEmpty(timers)) {
+ for (PollTimerInfo timerInfo : timers.values()) {
+ vertx.cancelTimer(timerInfo.getTimerId());
+ }
}
// 娓呯悊璇锋眰闃熷垪
deviceRequestQueues.remove(deviceId);
deviceLastRequestTime.remove(deviceId);
deviceDelayTimerActive.remove(deviceId);
- log.debug("[stopPolling][璁惧 {} 鍋滄浜� {} 涓疆璇㈠畾鏃跺櫒]", deviceId, timers.size());
+ log.debug("[stopPolling][璁惧 {} 鍋滄浜� {} 涓疆璇㈠畾鏃跺櫒]", deviceId,
+ CollUtil.isEmpty(timers) ? 0 : timers.size());
}
/**
* 鍋滄鎵�鏈夎疆璇�
*/
public void stopAll() {
- for (Long deviceId : new ArrayList<>(devicePointTimers.keySet())) {
+ for (Long deviceId : new ArrayList<>(devicePollTimers.keySet())) {
stopPolling(deviceId);
}
}
--
Gitblit v1.9.3