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/tcpclient/manager/IotModbusTcpClientPollScheduler.java | 219 ++++++++++++++++++++++++++++++++++++++++++++++++++++++
1 files changed, 219 insertions(+), 0 deletions(-)
diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpclient/manager/IotModbusTcpClientPollScheduler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpclient/manager/IotModbusTcpClientPollScheduler.java
index 946937d..fba2f75 100644
--- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpclient/manager/IotModbusTcpClientPollScheduler.java
+++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpclient/manager/IotModbusTcpClientPollScheduler.java
@@ -9,7 +9,15 @@
import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.utils.IotModbusTcpClientUtils;
import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpclient.handler.upstream.IotModbusTcpClientUpstreamHandler;
import io.vertx.core.Vertx;
+import lombok.AllArgsConstructor;
+import lombok.Data;
import lombok.extern.slf4j.Slf4j;
+
+import java.util.*;
+
+import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertList;
+import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertMultiMap;
+import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.filterList;
/**
* IoT Modbus TCP Client 杞璋冨害鍣細绠＄悊鐐逛綅鐨勮疆璇㈠畾鏃跺櫒锛岃皟搴﹁鍙栦换鍔″苟涓婃姤缁撴灉
@@ -34,6 +42,47 @@
}
// ========== 杞鎵ц ==========
+
+ @Override
+ protected List<PollTask> buildPollTasks(IotModbusDeviceConfigRespDTO config) {
+ return convertList(buildReadSegments(config), segment -> new PollTask(segment.getKey(), segment.getPollInterval()));
+ }
+
+ /**
+ * 杞璇诲彇娈�
+ */
+ @Override
+ protected void pollTask(Long deviceId, String taskKey) {
+ // 1.1 浠� configCache 鑾峰彇鏈�鏂伴厤缃�
+ IotModbusDeviceConfigRespDTO config = configCacheService.getConfig(deviceId);
+ if (config == null || CollUtil.isEmpty(config.getPoints())) {
+ log.warn("[pollTask][璁惧 {} 娌℃湁閰嶇疆]", deviceId);
+ return;
+ }
+ // 1.2 鏌ユ壘璇诲彇娈点�傞厤缃彉鍖栧悗锛屽鏋滃綋鍓� taskKey 宸蹭笉瀛樺湪锛岀洿鎺ヨ烦杩囩瓑寰呬笅涓�杞� updatePolling 娓呯悊 timer
+ ReadSegment segment = findReadSegment(config, taskKey);
+ if (segment == null) {
+ log.debug("[pollTask][璁惧 {} 璇诲彇娈� {} 鏈壘鍒帮紝璺宠繃闄堟棫杞浠诲姟]", deviceId, taskKey);
+ return;
+ }
+
+ // 2.1 鑾峰彇杩炴帴
+ IotModbusTcpClientConnectionManager.ModbusConnection connection = connectionManager.getConnection(deviceId);
+ if (connection == null) {
+ log.warn("[pollTask][璁惧 {} 娌℃湁杩炴帴]", deviceId);
+ return;
+ }
+ // 2.2 鑾峰彇 slave ID
+ Integer slaveId = connectionManager.getSlaveId(deviceId);
+ Assert.notNull(slaveId, "璁惧 {} 娌℃湁閰嶇疆 slaveId", deviceId);
+
+ // 3. 鎵ц Modbus 鎵归噺璇诲彇
+ IotModbusTcpClientUtils.read(connection, slaveId, segment.getFunctionCode(),
+ segment.getStartAddress(), segment.getRegisterCount())
+ .onSuccess(rawValues -> handleSegmentReadResult(config, segment, rawValues))
+ .onFailure(e -> log.error("[pollTask][璇诲彇鐐逛綅娈靛け璐�, deviceId={}, segment={}]",
+ deviceId, segment.getKey(), e));
+ }
/**
* 杞鍗曚釜鐐逛綅
@@ -70,4 +119,174 @@
deviceId, point.getIdentifier(), e));
}
+ private void handleSegmentReadResult(IotModbusDeviceConfigRespDTO config,
+ ReadSegment segment,
+ int[] rawValues) {
+ for (IotModbusPointRespDTO point : segment.getPoints()) {
+ // 鎵归噺璇诲彇杩斿洖鐨勬槸鏁翠釜杩炵画鍦板潃娈碉紝闇�瑕佹寜鐐逛綅鍦板潃鍒囩墖鍚庡啀澶嶇敤鍗曠偣涓婃姤閫昏緫
+ int[] pointRawValues = extractPointRawValues(rawValues, segment, point);
+ if (pointRawValues == null) {
+ log.warn("[handleSegmentReadResult][璇诲彇娈电粨鏋滈暱搴︿笉瓒�, deviceId={}, segment={}, identifier={}]",
+ config.getDeviceId(), segment.getKey(), point.getIdentifier());
+ continue;
+ }
+ upstreamHandler.handleReadResult(config, point, pointRawValues);
+ }
+ }
+
+ private ReadSegment findReadSegment(IotModbusDeviceConfigRespDTO config, String taskKey) {
+ return CollUtil.findOne(buildReadSegments(config), segment -> segment.getKey().equals(taskKey));
+ }
+
+ /**
+ * 鏋勫缓鎵归噺璇诲彇娈�
+ *
+ * <p>鍙悎骞跺姛鑳界爜銆佽疆璇㈤棿闅旂浉鍚岋紝涓斿湴鍧�杩炵画鎴栭噸鍙犵殑鐐逛綅锛涗笉璺ㄥ姛鑳界爜銆佷笉璺ㄨ疆璇㈤棿闅旓紝閬垮厤鏀瑰彉鍘熸湁杞璇箟銆�
+ * 鍚屾椂鎸� Modbus 鍗忚闄愬埗鎺у埗鍗曟璇诲彇闀垮害锛岃秴杩囬檺鍒舵椂鎷嗘垚澶氫釜璇诲彇娈点��
+ */
+ static List<ReadSegment> buildReadSegments(IotModbusDeviceConfigRespDTO config) {
+ if (config == null) {
+ return Collections.emptyList();
+ }
+
+ // 1. 鎸夊姛鑳界爜鍜岃疆璇㈤棿闅斿垎缁勶細涓よ�呬换涓�涓嶅悓锛岄兘涓嶈兘鍏辩敤鍚屼竴娆� Modbus 璇昏姹�
+ List<IotModbusPointRespDTO> validPoints = filterList(config.getPoints(), IotModbusTcpClientPollScheduler::isValidReadPoint);
+ if (CollUtil.isEmpty(validPoints)) {
+ return Collections.emptyList();
+ }
+ Map<SegmentGroupKey, List<IotModbusPointRespDTO>> pointsByGroup = convertMultiMap(validPoints,
+ point -> new SegmentGroupKey(point.getFunctionCode(), point.getPollInterval()));
+
+ // 2. 缁勫唴鎸夊湴鍧�鎺掑簭鍚庯紝鍚堝苟杩炵画鎴栭噸鍙犲尯闂达紝鐢熸垚瀹為檯杞鐨勮鍙栨
+ List<ReadSegment> segments = new ArrayList<>();
+ for (Map.Entry<SegmentGroupKey, List<IotModbusPointRespDTO>> entry : pointsByGroup.entrySet()) {
+ List<IotModbusPointRespDTO> points = entry.getValue();
+ points.sort(Comparator.comparing(IotModbusPointRespDTO::getRegisterAddress)
+ .thenComparing(IotModbusPointRespDTO::getRegisterCount)
+ .thenComparing(IotModbusPointRespDTO::getId));
+ buildReadSegments(entry.getKey(), points, segments);
+ }
+ // 3. 鍥哄畾鎺掑簭锛屼繚璇佺敓鎴愮殑 taskKey 绋冲畾锛屼究浜� updatePolling 鍋氬閲忔洿鏂�
+ segments.sort(Comparator.comparing(ReadSegment::getFunctionCode)
+ .thenComparing(ReadSegment::getPollInterval)
+ .thenComparing(ReadSegment::getStartAddress));
+ return segments;
+ }
+
+ private static void buildReadSegments(SegmentGroupKey groupKey,
+ List<IotModbusPointRespDTO> points,
+ List<ReadSegment> segments) {
+ ReadSegment current = null;
+ int maxRegisterCount = getMaxRegisterCount(groupKey.getFunctionCode());
+ // points 宸叉寜 registerAddress 鎺掑簭锛屽洜姝ゅ彲浠ョ嚎鎬у悎骞惰繛缁�/閲嶅彔鍦板潃娈�
+ for (IotModbusPointRespDTO point : points) {
+ int pointStartAddress = point.getRegisterAddress();
+ int pointEndAddress = pointStartAddress + point.getRegisterCount();
+ // 1. 褰撳墠鐐逛綅鏃犳硶鍚堝苟鏃讹紝鏂板缓涓�涓鍙栨
+ if (current == null || !canMerge(current, pointStartAddress, pointEndAddress, maxRegisterCount)) {
+ current = new ReadSegment(groupKey.getFunctionCode(), groupKey.getPollInterval(),
+ pointStartAddress, point.getRegisterCount(), new ArrayList<>());
+ segments.add(current);
+ } else {
+ // 2. 褰撳墠鐐逛綅鍙悎骞舵椂锛屾墿灞曡鍙栨瑕嗙洊鑼冨洿
+ current.setRegisterCount(Math.max(current.getEndAddress(), pointEndAddress) - current.getStartAddress());
+ }
+ // 3. 璁板綍璇诲彇娈靛寘鍚殑鐐逛綅锛岃鍙栨垚鍔熷悗鎸夌偣浣嶉�愪釜鍒囩墖涓婃姤
+ current.getPoints().add(point);
+ }
+ }
+
+ /**
+ * 鍒ゆ柇鐐逛綅鏄惁鍙互鍚堝苟鍒板綋鍓嶈鍙栨
+ *
+ * <p>浠呭悎骞惰繛缁垨閲嶅彔鍖洪棿锛屼笉鍚堝苟瀛樺湪鍦板潃绌烘礊鐨勫尯闂达紝閬垮厤棰濆璇诲彇鏃犲叧瀵勫瓨鍣ㄣ��
+ */
+ private static boolean canMerge(ReadSegment segment, int pointStartAddress, int pointEndAddress, int maxRegisterCount) {
+ if (pointStartAddress > segment.getEndAddress()) {
+ return false;
+ }
+ int mergedRegisterCount = Math.max(segment.getEndAddress(), pointEndAddress) - segment.getStartAddress();
+ return mergedRegisterCount <= maxRegisterCount;
+ }
+
+ /**
+ * 浠庢壒閲忚鍙栫粨鏋滀腑鎻愬彇鍗曚釜鐐逛綅鐨勫師濮嬪��
+ *
+ * <p>渚嬪璇诲彇娈典粠鍦板潃 10 寮�濮嬶紝鐐逛綅鍦板潃涓� 12銆佹暟閲忎负 2锛屽垯鍙� rawValues[2..4)銆�
+ */
+ static int[] extractPointRawValues(int[] rawValues, ReadSegment segment, IotModbusPointRespDTO point) {
+ if (rawValues == null) {
+ return null;
+ }
+ // 1. 璁＄畻鐐逛綅鍦ㄦ壒閲忚鍙栫粨鏋滀腑鐨勭浉瀵瑰亸绉�
+ int offset = point.getRegisterAddress() - segment.getStartAddress();
+ int end = offset + point.getRegisterCount();
+ // 2. 闃插尽寮傚父鍝嶅簲闀垮害锛岄伩鍏嶈秺鐣屽奖鍝嶅悓涓�璇诲彇娈靛唴鍏跺畠鐐逛綅
+ if (offset < 0 || end > rawValues.length) {
+ return null;
+ }
+ // 3. 杩斿洖鍗曚釜鐐逛綅闇�瑕佺殑鍘熷瀵勫瓨鍣ㄥ��
+ return Arrays.copyOfRange(rawValues, offset, end);
+ }
+
+ private static boolean isValidReadPoint(IotModbusPointRespDTO point) {
+ return point != null
+ && point.getId() != null
+ && point.getFunctionCode() != null
+ && point.getRegisterAddress() != null
+ && point.getRegisterCount() != null
+ && point.getRegisterCount() > 0
+ && point.getPollInterval() != null
+ && point.getPollInterval() > 0;
+ }
+
+ @SuppressWarnings("EnhancedSwitchMigration")
+ private static int getMaxRegisterCount(Integer functionCode) {
+ switch (functionCode) {
+ case IotModbusCommonUtils.FC_READ_COILS:
+ case IotModbusCommonUtils.FC_READ_DISCRETE_INPUTS:
+ return 2000;
+ case IotModbusCommonUtils.FC_READ_HOLDING_REGISTERS:
+ case IotModbusCommonUtils.FC_READ_INPUT_REGISTERS:
+ return 125;
+ default:
+ return 0;
+ }
+ }
+
+ /**
+ * 璇诲彇娈靛垎缁� Key
+ */
+ @Data
+ @AllArgsConstructor
+ static class SegmentGroupKey {
+
+ private Integer functionCode;
+ private Integer pollInterval;
+
+ }
+
+ /**
+ * 涓�娆� Modbus 鎵归噺璇昏姹傚搴旂殑杩炵画鍦板潃娈�
+ */
+ @Data
+ @AllArgsConstructor
+ static class ReadSegment {
+
+ private Integer functionCode;
+ private Integer pollInterval;
+ private Integer startAddress;
+ private Integer registerCount;
+ private List<IotModbusPointRespDTO> points;
+
+ String getKey() {
+ return functionCode + ":" + pollInterval + ":" + startAddress + ":" + registerCount;
+ }
+
+ int getEndAddress() {
+ return startAddress + registerCount;
+ }
+
+ }
+
}
--
Gitblit v1.9.3