BJ_ZTEITS-Developers / parking-cloud-yudao

Commit c1101dc4b9089d6d13ea2ef128e4137193a80294

Authored by YunaiV
1 parent 7b6adb41

feat(iot): 优化 Modbus TCP Client 连续点位批量读取 

- 泛化 Modbus 轮询任务为 taskKey，保持默认单点轮询兼容
- 支持 TCP Client 按功能码、轮询间隔和连续地址聚合读取段
- 新增区间读取方法，批量读取后按点位切片上报
- 按 Modbus 单次读取上限拆分读取段
- 补充连续、重叠、拆段和线圈切片单测

https://github.com/YunaiV/ruoyi-vue-pro/issues/1147
Inline Side-by-side
Showing 4 changed files with 487 additions and 55 deletions
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/common/manager/AbstractIotModbusPollScheduler.java
  View file @c1101dc
@@ -12,6 +12,7 @@ import java.util.*; @@ -12,6 +12,7 @@ import java.util.*;
12 import java.util.concurrent.ConcurrentHashMap; 12 import java.util.concurrent.ConcurrentHashMap;
13 import java.util.concurrent.ConcurrentLinkedQueue; 13 import java.util.concurrent.ConcurrentLinkedQueue;
14 14
  15 +import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertList;
15 import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertSet; 16 import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertSet;
16 17
17 /** 18 /**
@@ -19,6 +20,8 @@ import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils. @@ -19,6 +20,8 @@ import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.
19 * <p> 20 * <p>
20 * 封装通用的定时器管理、per-device 请求队列限速逻辑。 21 * 封装通用的定时器管理、per-device 请求队列限速逻辑。
21 * 子类只需实现 {@link #pollPoint(Long, Long)} 定义具体的轮询动作。 22 * 子类只需实现 {@link #pollPoint(Long, Long)} 定义具体的轮询动作。
  23 + * 如需将多个点位合并为一个轮询任务，可覆盖 {@link #buildPollTasks(IotModbusDeviceConfigRespDTO)}
  24 + * 和 {@link #pollTask(Long, String)}。
22 * <p> 25 * <p>
23 * 26 *
24 * @author 芋道源码 27 * @author 芋道源码
@@ -38,9 +41,9 @@ public abstract class AbstractIotModbusPollScheduler { @@ -38,9 +41,9 @@ public abstract class AbstractIotModbusPollScheduler {
38 private static final int MAX_QUEUE_SIZE = 1000; 41 private static final int MAX_QUEUE_SIZE = 1000;
39 42
40 /** 43 /**
41 - * 设备点位的定时器映射：deviceId -> (pointId -> PointTimerInfo) 44 + * 设备轮询任务的定时器映射：deviceId -> (taskKey -> PollTimerInfo)
42 */ 45 */
43 - private final Map<Long, Map<Long, PointTimerInfo>> devicePointTimers = new ConcurrentHashMap<>(); 46 + private final Map<Long, Map<String, PollTimerInfo>> devicePollTimers = new ConcurrentHashMap<>();
44 47
45 /** 48 /**
46 * per-device 请求队列：deviceId -> 待执行请求队列 49 * per-device 请求队列：deviceId -> 待执行请求队列
@@ -60,11 +63,29 @@ public abstract class AbstractIotModbusPollScheduler { @@ -60,11 +63,29 @@ public abstract class AbstractIotModbusPollScheduler {
60 } 63 }
61 64
62 /** 65 /**
63 - * 点位定时器信息 66 + * 轮询任务信息
64 */ 67 */
65 @Data 68 @Data
66 @AllArgsConstructor 69 @AllArgsConstructor
67 - private static class PointTimerInfo { 70 + protected static class PollTask {
  71 +
  72 + /**
  73 + * 任务标识
  74 + */
  75 + private String key;
  76 + /**
  77 + * 轮询间隔（用于判断是否需要更新定时器）
  78 + */
  79 + private Integer pollInterval;
  80 +
  81 + }
  82 +
  83 + /**
  84 + * 轮询定时器信息
  85 + */
  86 + @Data
  87 + @AllArgsConstructor
  88 + private static class PollTimerInfo {
68 89
69 /** 90 /**
70 * Vert.x 定时器 ID 91 * Vert.x 定时器 ID
@@ -82,75 +103,84 @@ public abstract class AbstractIotModbusPollScheduler { @@ -82,75 +103,84 @@ public abstract class AbstractIotModbusPollScheduler {
82 /** 103 /**
83 * 更新轮询任务（增量更新） 104 * 更新轮询任务（增量更新）
84 * 105 *
85 - * 1. 【删除】点位：停止对应的轮询定时器  
86 - * 2. 【新增】点位：创建对应的轮询定时器  
87 - * 3. 【修改】点位：pollInterval 变化，重建对应的轮询定时器  
88 - * 【修改】其他属性变化：不需要重建定时器（pollPoint 运行时从 configCache 取最新 point） 106 + * 1. 【删除】任务：停止对应的轮询定时器
  107 + * 2. 【新增】任务：创建对应的轮询定时器
  108 + * 3. 【修改】任务：pollInterval 变化，重建对应的轮询定时器
  109 + * 【修改】其他属性变化：不需要重建定时器（pollTask 运行时从 configCache 取最新配置）
89 */ 110 */
90 public void updatePolling(IotModbusDeviceConfigRespDTO config) { 111 public void updatePolling(IotModbusDeviceConfigRespDTO config) {
91 Long deviceId = config.getDeviceId(); 112 Long deviceId = config.getDeviceId();
92 - List<IotModbusPointRespDTO> newPoints = config.getPoints();  
93 - Map<Long, PointTimerInfo> currentTimers = devicePointTimers 113 + List<PollTask> newTasks = buildPollTasks(config);
  114 + Map<String, PollTimerInfo> currentTimers = devicePollTimers
94 .computeIfAbsent(deviceId, k -> new ConcurrentHashMap<>()); 115 .computeIfAbsent(deviceId, k -> new ConcurrentHashMap<>());
95 - // 1.1 计算新配置中的点位 ID 集合  
96 - Set<Long> newPointIds = convertSet(newPoints, IotModbusPointRespDTO::getId);  
97 - // 1.2 计算删除的点位 ID 集合  
98 - Set<Long> removedPointIds = new HashSet<>(currentTimers.keySet());  
99 - removedPointIds.removeAll(newPointIds);  
100 -  
101 - // 2. 处理删除的点位：停止不再存在的定时器  
102 - for (Long pointId : removedPointIds) {  
103 - PointTimerInfo timerInfo = currentTimers.remove(pointId); 116 + // 1.1 计算新配置中的任务 Key 集合
  117 + Set<String> newTaskKeys = convertSet(newTasks, PollTask::getKey);
  118 + // 1.2 计算删除的任务 Key 集合
  119 + Set<String> removedTaskKeys = new HashSet<>(currentTimers.keySet());
  120 + removedTaskKeys.removeAll(newTaskKeys);
  121 +
  122 + // 2. 处理删除的任务：停止不再存在的定时器
  123 + for (String taskKey : removedTaskKeys) {
  124 + PollTimerInfo timerInfo = currentTimers.remove(taskKey);
104 if (timerInfo != null) { 125 if (timerInfo != null) {
105 vertx.cancelTimer(timerInfo.getTimerId()); 126 vertx.cancelTimer(timerInfo.getTimerId());
106 - log.debug("[updatePolling][设备 {} 点位 {} 定时器已删除]", deviceId, pointId); 127 + log.debug("[updatePolling][设备 {} 轮询任务 {} 定时器已删除]", deviceId, taskKey);
107 } 128 }
108 } 129 }
109 130
110 - // 3. 处理新增和修改的点位  
111 - if (CollUtil.isEmpty(newPoints)) { 131 + // 3. 处理新增和修改的任务
  132 + if (CollUtil.isEmpty(newTasks)) {
112 return; 133 return;
113 } 134 }
114 - for (IotModbusPointRespDTO point : newPoints) {  
115 - Long pointId = point.getId();  
116 - Integer newPollInterval = point.getPollInterval();  
117 - PointTimerInfo existingTimer = currentTimers.get(pointId);  
118 - // 3.1 新增点位：创建定时器 135 + for (PollTask task : newTasks) {
  136 + String taskKey = task.getKey();
  137 + Integer newPollInterval = task.getPollInterval();
  138 + PollTimerInfo existingTimer = currentTimers.get(taskKey);
  139 + // 3.1 新增任务：创建定时器
119 if (existingTimer == null) { 140 if (existingTimer == null) {
120 - Long timerId = createPollTimer(deviceId, pointId, newPollInterval); 141 + Long timerId = createPollTimer(deviceId, taskKey, newPollInterval);
121 if (timerId != null) { 142 if (timerId != null) {
122 - currentTimers.put(pointId, new PointTimerInfo(timerId, newPollInterval));  
123 - log.debug("[updatePolling][设备 {} 点位 {} 定时器已创建, interval={}ms]",  
124 - deviceId, pointId, newPollInterval); 143 + currentTimers.put(taskKey, new PollTimerInfo(timerId, newPollInterval));
  144 + log.debug("[updatePolling][设备 {} 轮询任务 {} 定时器已创建, interval={}ms]",
  145 + deviceId, taskKey, newPollInterval);
125 } 146 }
126 } else if (!Objects.equals(existingTimer.getPollInterval(), newPollInterval)) { 147 } else if (!Objects.equals(existingTimer.getPollInterval(), newPollInterval)) {
127 // 3.2 pollInterval 变化：重建定时器 148 // 3.2 pollInterval 变化：重建定时器
128 vertx.cancelTimer(existingTimer.getTimerId()); 149 vertx.cancelTimer(existingTimer.getTimerId());
129 - Long timerId = createPollTimer(deviceId, pointId, newPollInterval); 150 + Long timerId = createPollTimer(deviceId, taskKey, newPollInterval);
130 if (timerId != null) { 151 if (timerId != null) {
131 - currentTimers.put(pointId, new PointTimerInfo(timerId, newPollInterval));  
132 - log.debug("[updatePolling][设备 {} 点位 {} 定时器已更新, interval={}ms -> {}ms]",  
133 - deviceId, pointId, existingTimer.getPollInterval(), newPollInterval); 152 + currentTimers.put(taskKey, new PollTimerInfo(timerId, newPollInterval));
  153 + log.debug("[updatePolling][设备 {} 轮询任务 {} 定时器已更新, interval={}ms -> {}ms]",
  154 + deviceId, taskKey, existingTimer.getPollInterval(), newPollInterval);
134 } else { 155 } else {
135 - currentTimers.remove(pointId); 156 + currentTimers.remove(taskKey);
136 } 157 }
137 } 158 }
138 - // 3.3 其他属性变化：无需重建定时器，因为 pollPoint() 运行时从 configCache 获取最新 point，自动使用新配置 159 + // 3.3 其他属性变化：无需重建定时器，因为 pollTask() 运行时从 configCache 获取最新配置
139 } 160 }
140 } 161 }
141 162
142 /** 163 /**
  164 + * 构建轮询任务列表
  165 + *
  166 + * 默认每个点位一个轮询任务。TCP Client 等协议可覆盖该方法，将多个点位合并为一个批量读取任务。
  167 + */
  168 + protected List<PollTask> buildPollTasks(IotModbusDeviceConfigRespDTO config) {
  169 + return convertList(config.getPoints(), point -> new PollTask(String.valueOf(point.getId()), point.getPollInterval()));
  170 + }
  171 +
  172 + /**
143 * 创建轮询定时器 173 * 创建轮询定时器
144 */ 174 */
145 - private Long createPollTimer(Long deviceId, Long pointId, Integer pollInterval) { 175 + private Long createPollTimer(Long deviceId, String taskKey, Integer pollInterval) {
146 if (pollInterval == null || pollInterval <= 0) { 176 if (pollInterval == null || pollInterval <= 0) {
147 return null; 177 return null;
148 } 178 }
149 return vertx.setPeriodic(pollInterval, timerId -> { 179 return vertx.setPeriodic(pollInterval, timerId -> {
150 try { 180 try {
151 - submitPollRequest(deviceId, pointId); 181 + submitPollRequest(deviceId, taskKey);
152 } catch (Exception e) { 182 } catch (Exception e) {
153 - log.error("[createPollTimer][轮询点位失败, deviceId={}, pointId={}]", deviceId, pointId, e); 183 + log.error("[createPollTimer][轮询任务失败, deviceId={}, taskKey={}]", deviceId, taskKey, e);
154 } 184 }
155 }); 185 });
156 } 186 }
@@ -160,7 +190,7 @@ public abstract class AbstractIotModbusPollScheduler { @@ -160,7 +190,7 @@ public abstract class AbstractIotModbusPollScheduler {
160 /** 190 /**
161 * 提交轮询请求到设备请求队列（保证同设备请求间隔） 191 * 提交轮询请求到设备请求队列（保证同设备请求间隔）
162 */ 192 */
163 - private void submitPollRequest(Long deviceId, Long pointId) { 193 + private void submitPollRequest(Long deviceId, String taskKey) {
164 // 1. 【重要】将请求添加到设备的请求队列 194 // 1. 【重要】将请求添加到设备的请求队列
165 Queue<Runnable> queue = deviceRequestQueues.computeIfAbsent(deviceId, k -> new ConcurrentLinkedQueue<>()); 195 Queue<Runnable> queue = deviceRequestQueues.computeIfAbsent(deviceId, k -> new ConcurrentLinkedQueue<>());
166 while (queue.size() >= MAX_QUEUE_SIZE) { 196 while (queue.size() >= MAX_QUEUE_SIZE) {
@@ -168,7 +198,7 @@ public abstract class AbstractIotModbusPollScheduler { @@ -168,7 +198,7 @@ public abstract class AbstractIotModbusPollScheduler {
168 queue.poll(); 198 queue.poll();
169 log.warn("[submitPollRequest][设备 {} 请求队列已满({}), 丢弃最旧请求]", deviceId, MAX_QUEUE_SIZE); 199 log.warn("[submitPollRequest][设备 {} 请求队列已满({}), 丢弃最旧请求]", deviceId, MAX_QUEUE_SIZE);
170 } 200 }
171 - queue.offer(() -> pollPoint(deviceId, pointId)); 201 + queue.offer(() -> pollTask(deviceId, taskKey));
172 202
173 // 2. 处理设备请求队列（如果没有延迟 timer 在等待） 203 // 2. 处理设备请求队列（如果没有延迟 timer 在等待）
174 processDeviceQueue(deviceId); 204 processDeviceQueue(deviceId);
@@ -239,6 +269,15 @@ public abstract class AbstractIotModbusPollScheduler { @@ -239,6 +269,15 @@ public abstract class AbstractIotModbusPollScheduler {
239 // ========== 轮询执行 ========== 269 // ========== 轮询执行 ==========
240 270
241 /** 271 /**
  272 + * 轮询任务
  273 + *
  274 + * 默认将任务标识作为点位 ID，执行单点轮询。
  275 + */
  276 + protected void pollTask(Long deviceId, String taskKey) {
  277 + pollPoint(deviceId, Long.valueOf(taskKey));
  278 + }
  279 +
  280 + /**
242 * 轮询单个点位（子类实现具体的读取逻辑） 281 * 轮询单个点位（子类实现具体的读取逻辑）
243 * 282 *
244 * @param deviceId 设备 ID 283 * @param deviceId 设备 ID
@@ -252,25 +291,25 @@ public abstract class AbstractIotModbusPollScheduler { @@ -252,25 +291,25 @@ public abstract class AbstractIotModbusPollScheduler {
252 * 停止设备的轮询 291 * 停止设备的轮询
253 */ 292 */
254 public void stopPolling(Long deviceId) { 293 public void stopPolling(Long deviceId) {
255 - Map<Long, PointTimerInfo> timers = devicePointTimers.remove(deviceId);  
256 - if (CollUtil.isEmpty(timers)) {  
257 - return;  
258 - }  
259 - for (PointTimerInfo timerInfo : timers.values()) {  
260 - vertx.cancelTimer(timerInfo.getTimerId()); 294 + Map<String, PollTimerInfo> timers = devicePollTimers.remove(deviceId);
  295 + if (CollUtil.isNotEmpty(timers)) {
  296 + for (PollTimerInfo timerInfo : timers.values()) {
  297 + vertx.cancelTimer(timerInfo.getTimerId());
  298 + }
261 } 299 }
262 // 清理请求队列 300 // 清理请求队列
263 deviceRequestQueues.remove(deviceId); 301 deviceRequestQueues.remove(deviceId);
264 deviceLastRequestTime.remove(deviceId); 302 deviceLastRequestTime.remove(deviceId);
265 deviceDelayTimerActive.remove(deviceId); 303 deviceDelayTimerActive.remove(deviceId);
266 - log.debug("[stopPolling][设备 {} 停止了 {} 个轮询定时器]", deviceId, timers.size()); 304 + log.debug("[stopPolling][设备 {} 停止了 {} 个轮询定时器]", deviceId,
  305 + CollUtil.isEmpty(timers) ? 0 : timers.size());
267 } 306 }
268 307
269 /** 308 /**
270 * 停止所有轮询 309 * 停止所有轮询
271 */ 310 */
272 public void stopAll() { 311 public void stopAll() {
273 - for (Long deviceId : new ArrayList<>(devicePointTimers.keySet())) { 312 + for (Long deviceId : new ArrayList<>(devicePollTimers.keySet())) {
274 stopPolling(deviceId); 313 stopPolling(deviceId);
275 } 314 }
276 } 315 }
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/common/utils/IotModbusTcpClientUtils.java
  View file @c1101dc
@@ -39,11 +39,38 @@ public class IotModbusTcpClientUtils { @@ -39,11 +39,38 @@ public class IotModbusTcpClientUtils {
39 public static Future<int[]> read(IotModbusTcpClientConnectionManager.ModbusConnection connection, 39 public static Future<int[]> read(IotModbusTcpClientConnectionManager.ModbusConnection connection,
40 Integer slaveId, 40 Integer slaveId,
41 IotModbusPointRespDTO point) { 41 IotModbusPointRespDTO point) {
  42 + return read(connection, slaveId, point.getFunctionCode(),
  43 + point.getRegisterAddress(), point.getRegisterCount(), point.getIdentifier());
  44 + }
  45 +
  46 + /**
  47 + * 读取 Modbus 数据
  48 + *
  49 + * @param connection Modbus 连接
  50 + * @param slaveId 从站地址
  51 + * @param functionCode 功能码
  52 + * @param registerAddress 寄存器起始地址
  53 + * @param registerCount 寄存器数量
  54 + * @return 原始值（int 数组）
  55 + */
  56 + public static Future<int[]> read(IotModbusTcpClientConnectionManager.ModbusConnection connection,
  57 + Integer slaveId,
  58 + Integer functionCode,
  59 + Integer registerAddress,
  60 + Integer registerCount) {
  61 + return read(connection, slaveId, functionCode, registerAddress, registerCount, null);
  62 + }
  63 +
  64 + private static Future<int[]> read(IotModbusTcpClientConnectionManager.ModbusConnection connection,
  65 + Integer slaveId,
  66 + Integer functionCode,
  67 + Integer registerAddress,
  68 + Integer registerCount,
  69 + String identifier) {
42 return connection.executeBlocking(tcpConnection -> { 70 return connection.executeBlocking(tcpConnection -> {
43 try { 71 try {
44 // 1. 创建请求 72 // 1. 创建请求
45 - ModbusRequest request = createReadRequest(point.getFunctionCode(),  
46 - point.getRegisterAddress(), point.getRegisterCount()); 73 + ModbusRequest request = createReadRequest(functionCode, registerAddress, registerCount);
47 request.setUnitID(slaveId); 74 request.setUnitID(slaveId);
48 75
49 // 2. 执行事务（请求） 76 // 2. 执行事务（请求）
@@ -53,10 +80,10 @@ public class IotModbusTcpClientUtils { @@ -53,10 +80,10 @@ public class IotModbusTcpClientUtils {
53 80
54 // 3. 解析响应 81 // 3. 解析响应
55 ModbusResponse response = transaction.getResponse(); 82 ModbusResponse response = transaction.getResponse();
56 - return extractValues(response, point.getFunctionCode()); 83 + return extractValues(response, functionCode);
57 } catch (Exception e) { 84 } catch (Exception e) {
58 - throw new RuntimeException(String.format("Modbus 读取失败 [slaveId=%d, identifier=%s, address=%d]",  
59 - slaveId, point.getIdentifier(), point.getRegisterAddress()), e); 85 + throw new RuntimeException(String.format("Modbus 读取失败 [slaveId=%d, identifier=%s, functionCode=%d, address=%d, count=%d]",
  86 + slaveId, identifier, functionCode, registerAddress, registerCount), e);
60 } 87 }
61 }); 88 });
62 } 89 }
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpclient/manager/IotModbusTcpClientPollScheduler.java
  View file @c1101dc
@@ -9,8 +9,16 @@ import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.utils.IotModbu @@ -9,8 +9,16 @@ import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.utils.IotModbu
9 import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.utils.IotModbusTcpClientUtils; 9 import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.utils.IotModbusTcpClientUtils;
10 import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpclient.handler.upstream.IotModbusTcpClientUpstreamHandler; 10 import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpclient.handler.upstream.IotModbusTcpClientUpstreamHandler;
11 import io.vertx.core.Vertx; 11 import io.vertx.core.Vertx;
  12 +import lombok.AllArgsConstructor;
  13 +import lombok.Data;
12 import lombok.extern.slf4j.Slf4j; 14 import lombok.extern.slf4j.Slf4j;
13 15
  16 +import java.util.*;
  17 +
  18 +import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertList;
  19 +import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertMultiMap;
  20 +import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.filterList;
  21 +
14 /** 22 /**
15 * IoT Modbus TCP Client 轮询调度器：管理点位的轮询定时器，调度读取任务并上报结果 23 * IoT Modbus TCP Client 轮询调度器：管理点位的轮询定时器，调度读取任务并上报结果
16 * 24 *
@@ -35,6 +43,47 @@ public class IotModbusTcpClientPollScheduler extends AbstractIotModbusPollSchedu @@ -35,6 +43,47 @@ public class IotModbusTcpClientPollScheduler extends AbstractIotModbusPollSchedu
35 43
36 // ========== 轮询执行 ========== 44 // ========== 轮询执行 ==========
37 45
  46 + @Override
  47 + protected List<PollTask> buildPollTasks(IotModbusDeviceConfigRespDTO config) {
  48 + return convertList(buildReadSegments(config), segment -> new PollTask(segment.getKey(), segment.getPollInterval()));
  49 + }
  50 +
  51 + /**
  52 + * 轮询读取段
  53 + */
  54 + @Override
  55 + protected void pollTask(Long deviceId, String taskKey) {
  56 + // 1.1 从 configCache 获取最新配置
  57 + IotModbusDeviceConfigRespDTO config = configCacheService.getConfig(deviceId);
  58 + if (config == null || CollUtil.isEmpty(config.getPoints())) {
  59 + log.warn("[pollTask][设备 {} 没有配置]", deviceId);
  60 + return;
  61 + }
  62 + // 1.2 查找读取段。配置变化后，如果当前 taskKey 已不存在，直接跳过等待下一轮 updatePolling 清理 timer
  63 + ReadSegment segment = findReadSegment(config, taskKey);
  64 + if (segment == null) {
  65 + log.debug("[pollTask][设备 {} 读取段 {} 未找到，跳过陈旧轮询任务]", deviceId, taskKey);
  66 + return;
  67 + }
  68 +
  69 + // 2.1 获取连接
  70 + IotModbusTcpClientConnectionManager.ModbusConnection connection = connectionManager.getConnection(deviceId);
  71 + if (connection == null) {
  72 + log.warn("[pollTask][设备 {} 没有连接]", deviceId);
  73 + return;
  74 + }
  75 + // 2.2 获取 slave ID
  76 + Integer slaveId = connectionManager.getSlaveId(deviceId);
  77 + Assert.notNull(slaveId, "设备 {} 没有配置 slaveId", deviceId);
  78 +
  79 + // 3. 执行 Modbus 批量读取
  80 + IotModbusTcpClientUtils.read(connection, slaveId, segment.getFunctionCode(),
  81 + segment.getStartAddress(), segment.getRegisterCount())
  82 + .onSuccess(rawValues -> handleSegmentReadResult(config, segment, rawValues))
  83 + .onFailure(e -> log.error("[pollTask][读取点位段失败, deviceId={}, segment={}]",
  84 + deviceId, segment.getKey(), e));
  85 + }
  86 +
38 /** 87 /**
39 * 轮询单个点位 88 * 轮询单个点位
40 */ 89 */
@@ -70,4 +119,174 @@ public class IotModbusTcpClientPollScheduler extends AbstractIotModbusPollSchedu @@ -70,4 +119,174 @@ public class IotModbusTcpClientPollScheduler extends AbstractIotModbusPollSchedu
70 deviceId, point.getIdentifier(), e)); 119 deviceId, point.getIdentifier(), e));
71 } 120 }
72 121
  122 + private void handleSegmentReadResult(IotModbusDeviceConfigRespDTO config,
  123 + ReadSegment segment,
  124 + int[] rawValues) {
  125 + for (IotModbusPointRespDTO point : segment.getPoints()) {
  126 + // 批量读取返回的是整个连续地址段，需要按点位地址切片后再复用单点上报逻辑
  127 + int[] pointRawValues = extractPointRawValues(rawValues, segment, point);
  128 + if (pointRawValues == null) {
  129 + log.warn("[handleSegmentReadResult][读取段结果长度不足, deviceId={}, segment={}, identifier={}]",
  130 + config.getDeviceId(), segment.getKey(), point.getIdentifier());
  131 + continue;
  132 + }
  133 + upstreamHandler.handleReadResult(config, point, pointRawValues);
  134 + }
  135 + }
  136 +
  137 + private ReadSegment findReadSegment(IotModbusDeviceConfigRespDTO config, String taskKey) {
  138 + return CollUtil.findOne(buildReadSegments(config), segment -> segment.getKey().equals(taskKey));
  139 + }
  140 +
  141 + /**
  142 + * 构建批量读取段
  143 + *
  144 + * <p>只合并功能码、轮询间隔相同，且地址连续或重叠的点位；不跨功能码、不跨轮询间隔，避免改变原有轮询语义。
  145 + * 同时按 Modbus 协议限制控制单次读取长度，超过限制时拆成多个读取段。
  146 + */
  147 + static List<ReadSegment> buildReadSegments(IotModbusDeviceConfigRespDTO config) {
  148 + if (config == null) {
  149 + return Collections.emptyList();
  150 + }
  151 +
  152 + // 1. 按功能码和轮询间隔分组：两者任一不同，都不能共用同一次 Modbus 读请求
  153 + List<IotModbusPointRespDTO> validPoints = filterList(config.getPoints(), IotModbusTcpClientPollScheduler::isValidReadPoint);
  154 + if (CollUtil.isEmpty(validPoints)) {
  155 + return Collections.emptyList();
  156 + }
  157 + Map<SegmentGroupKey, List<IotModbusPointRespDTO>> pointsByGroup = convertMultiMap(validPoints,
  158 + point -> new SegmentGroupKey(point.getFunctionCode(), point.getPollInterval()));
  159 +
  160 + // 2. 组内按地址排序后，合并连续或重叠区间，生成实际轮询的读取段
  161 + List<ReadSegment> segments = new ArrayList<>();
  162 + for (Map.Entry<SegmentGroupKey, List<IotModbusPointRespDTO>> entry : pointsByGroup.entrySet()) {
  163 + List<IotModbusPointRespDTO> points = entry.getValue();
  164 + points.sort(Comparator.comparing(IotModbusPointRespDTO::getRegisterAddress)
  165 + .thenComparing(IotModbusPointRespDTO::getRegisterCount)
  166 + .thenComparing(IotModbusPointRespDTO::getId));
  167 + buildReadSegments(entry.getKey(), points, segments);
  168 + }
  169 + // 3. 固定排序，保证生成的 taskKey 稳定，便于 updatePolling 做增量更新
  170 + segments.sort(Comparator.comparing(ReadSegment::getFunctionCode)
  171 + .thenComparing(ReadSegment::getPollInterval)
  172 + .thenComparing(ReadSegment::getStartAddress));
  173 + return segments;
  174 + }
  175 +
  176 + private static void buildReadSegments(SegmentGroupKey groupKey,
  177 + List<IotModbusPointRespDTO> points,
  178 + List<ReadSegment> segments) {
  179 + ReadSegment current = null;
  180 + int maxRegisterCount = getMaxRegisterCount(groupKey.getFunctionCode());
  181 + // points 已按 registerAddress 排序，因此可以线性合并连续/重叠地址段
  182 + for (IotModbusPointRespDTO point : points) {
  183 + int pointStartAddress = point.getRegisterAddress();
  184 + int pointEndAddress = pointStartAddress + point.getRegisterCount();
  185 + // 1. 当前点位无法合并时，新建一个读取段
  186 + if (current == null || !canMerge(current, pointStartAddress, pointEndAddress, maxRegisterCount)) {
  187 + current = new ReadSegment(groupKey.getFunctionCode(), groupKey.getPollInterval(),
  188 + pointStartAddress, point.getRegisterCount(), new ArrayList<>());
  189 + segments.add(current);
  190 + } else {
  191 + // 2. 当前点位可合并时，扩展读取段覆盖范围
  192 + current.setRegisterCount(Math.max(current.getEndAddress(), pointEndAddress) - current.getStartAddress());
  193 + }
  194 + // 3. 记录读取段包含的点位，读取成功后按点位逐个切片上报
  195 + current.getPoints().add(point);
  196 + }
  197 + }
  198 +
  199 + /**
  200 + * 判断点位是否可以合并到当前读取段
  201 + *
  202 + * <p>仅合并连续或重叠区间，不合并存在地址空洞的区间，避免额外读取无关寄存器。
  203 + */
  204 + private static boolean canMerge(ReadSegment segment, int pointStartAddress, int pointEndAddress, int maxRegisterCount) {
  205 + if (pointStartAddress > segment.getEndAddress()) {
  206 + return false;
  207 + }
  208 + int mergedRegisterCount = Math.max(segment.getEndAddress(), pointEndAddress) - segment.getStartAddress();
  209 + return mergedRegisterCount <= maxRegisterCount;
  210 + }
  211 +
  212 + /**
  213 + * 从批量读取结果中提取单个点位的原始值
  214 + *
  215 + * <p>例如读取段从地址 10 开始，点位地址为 12、数量为 2，则取 rawValues[2..4)。
  216 + */
  217 + static int[] extractPointRawValues(int[] rawValues, ReadSegment segment, IotModbusPointRespDTO point) {
  218 + if (rawValues == null) {
  219 + return null;
  220 + }
  221 + // 1. 计算点位在批量读取结果中的相对偏移
  222 + int offset = point.getRegisterAddress() - segment.getStartAddress();
  223 + int end = offset + point.getRegisterCount();
  224 + // 2. 防御异常响应长度，避免越界影响同一读取段内其它点位
  225 + if (offset < 0 || end > rawValues.length) {
  226 + return null;
  227 + }
  228 + // 3. 返回单个点位需要的原始寄存器值
  229 + return Arrays.copyOfRange(rawValues, offset, end);
  230 + }
  231 +
  232 + private static boolean isValidReadPoint(IotModbusPointRespDTO point) {
  233 + return point != null
  234 + && point.getId() != null
  235 + && point.getFunctionCode() != null
  236 + && point.getRegisterAddress() != null
  237 + && point.getRegisterCount() != null
  238 + && point.getRegisterCount() > 0
  239 + && point.getPollInterval() != null
  240 + && point.getPollInterval() > 0;
  241 + }
  242 +
  243 + @SuppressWarnings("EnhancedSwitchMigration")
  244 + private static int getMaxRegisterCount(Integer functionCode) {
  245 + switch (functionCode) {
  246 + case IotModbusCommonUtils.FC_READ_COILS:
  247 + case IotModbusCommonUtils.FC_READ_DISCRETE_INPUTS:
  248 + return 2000;
  249 + case IotModbusCommonUtils.FC_READ_HOLDING_REGISTERS:
  250 + case IotModbusCommonUtils.FC_READ_INPUT_REGISTERS:
  251 + return 125;
  252 + default:
  253 + return 0;
  254 + }
  255 + }
  256 +
  257 + /**
  258 + * 读取段分组 Key
  259 + */
  260 + @Data
  261 + @AllArgsConstructor
  262 + static class SegmentGroupKey {
  263 +
  264 + private Integer functionCode;
  265 + private Integer pollInterval;
  266 +
  267 + }
  268 +
  269 + /**
  270 + * 一次 Modbus 批量读请求对应的连续地址段
  271 + */
  272 + @Data
  273 + @AllArgsConstructor
  274 + static class ReadSegment {
  275 +
  276 + private Integer functionCode;
  277 + private Integer pollInterval;
  278 + private Integer startAddress;
  279 + private Integer registerCount;
  280 + private List<IotModbusPointRespDTO> points;
  281 +
  282 + String getKey() {
  283 + return functionCode + ":" + pollInterval + ":" + startAddress + ":" + registerCount;
  284 + }
  285 +
  286 + int getEndAddress() {
  287 + return startAddress + registerCount;
  288 + }
  289 +
  290 + }
  291 +
73 } 292 }
yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpclient/manager/IotModbusTcpClientPollSchedulerTest.java 0 → 100644
  View file @c1101dc
  1 +package cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpclient.manager;
  2 +
  3 +import cn.iocoder.yudao.module.iot.core.biz.dto.IotModbusDeviceConfigRespDTO;
  4 +import cn.iocoder.yudao.module.iot.core.biz.dto.IotModbusPointRespDTO;
  5 +import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpclient.manager.IotModbusTcpClientPollScheduler.ReadSegment;
  6 +import org.junit.jupiter.api.Test;
  7 +
  8 +import java.util.Arrays;
  9 +import java.util.Collections;
  10 +import java.util.List;
  11 +
  12 +import static cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.utils.IotModbusCommonUtils.FC_READ_COILS;
  13 +import static cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.utils.IotModbusCommonUtils.FC_READ_HOLDING_REGISTERS;
  14 +import static cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.utils.IotModbusCommonUtils.FC_READ_INPUT_REGISTERS;
  15 +import static org.junit.jupiter.api.Assertions.*;
  16 +
  17 +/**
  18 + * {@link IotModbusTcpClientPollScheduler} 的单元测试
  19 + *
  20 + * @author 芋道源码
  21 + */
  22 +public class IotModbusTcpClientPollSchedulerTest {
  23 +
  24 + @Test
  25 + public void testBuildReadSegments_mergeContinuousPoints() {
  26 + // 准备参数
  27 + IotModbusPointRespDTO point01 = randomPoint(1L, FC_READ_HOLDING_REGISTERS, 0, 1, 1000);
  28 + IotModbusPointRespDTO point02 = randomPoint(2L, FC_READ_HOLDING_REGISTERS, 1, 2, 1000);
  29 + IotModbusPointRespDTO point03 = randomPoint(3L, FC_READ_HOLDING_REGISTERS, 4, 1, 1000);
  30 + IotModbusDeviceConfigRespDTO config = randomConfig(point01, point02, point03);
  31 +
  32 + // 调用
  33 + List<ReadSegment> segments = IotModbusTcpClientPollScheduler.buildReadSegments(config);
  34 +
  35 + // 断言
  36 + assertEquals(2, segments.size());
  37 + assertEquals(0, segments.get(0).getStartAddress());
  38 + assertEquals(3, segments.get(0).getRegisterCount());
  39 + assertIterableEquals(Arrays.asList(point01, point02), segments.get(0).getPoints());
  40 + assertEquals(4, segments.get(1).getStartAddress());
  41 + assertEquals(1, segments.get(1).getRegisterCount());
  42 + assertIterableEquals(Collections.singletonList(point03), segments.get(1).getPoints());
  43 + }
  44 +
  45 + @Test
  46 + public void testBuildReadSegments_mergeOverlappingPoints() {
  47 + // 准备参数
  48 + IotModbusPointRespDTO point01 = randomPoint(1L, FC_READ_HOLDING_REGISTERS, 0, 2, 1000);
  49 + IotModbusPointRespDTO point02 = randomPoint(2L, FC_READ_HOLDING_REGISTERS, 1, 1, 1000);
  50 + IotModbusDeviceConfigRespDTO config = randomConfig(point01, point02);
  51 +
  52 + // 调用
  53 + List<ReadSegment> segments = IotModbusTcpClientPollScheduler.buildReadSegments(config);
  54 +
  55 + // 断言
  56 + assertEquals(1, segments.size());
  57 + assertEquals(0, segments.get(0).getStartAddress());
  58 + assertEquals(2, segments.get(0).getRegisterCount());
  59 + assertIterableEquals(Arrays.asList(point01, point02), segments.get(0).getPoints());
  60 + }
  61 +
  62 + @Test
  63 + public void testBuildReadSegments_notMergeDifferentFunctionCodeOrPollInterval() {
  64 + // 准备参数
  65 + IotModbusPointRespDTO point01 = randomPoint(1L, FC_READ_HOLDING_REGISTERS, 0, 1, 1000);
  66 + IotModbusPointRespDTO point02 = randomPoint(2L, FC_READ_INPUT_REGISTERS, 1, 1, 1000);
  67 + IotModbusPointRespDTO point03 = randomPoint(3L, FC_READ_HOLDING_REGISTERS, 1, 1, 2000);
  68 + IotModbusDeviceConfigRespDTO config = randomConfig(point01, point02, point03);
  69 +
  70 + // 调用
  71 + List<ReadSegment> segments = IotModbusTcpClientPollScheduler.buildReadSegments(config);
  72 +
  73 + // 断言
  74 + assertEquals(3, segments.size());
  75 + assertEquals(FC_READ_HOLDING_REGISTERS, segments.get(0).getFunctionCode());
  76 + assertEquals(1000, segments.get(0).getPollInterval());
  77 + assertEquals(FC_READ_HOLDING_REGISTERS, segments.get(1).getFunctionCode());
  78 + assertEquals(2000, segments.get(1).getPollInterval());
  79 + assertEquals(FC_READ_INPUT_REGISTERS, segments.get(2).getFunctionCode());
  80 + assertEquals(1000, segments.get(2).getPollInterval());
  81 + }
  82 +
  83 + @Test
  84 + public void testBuildReadSegments_splitWhenExceedsMaxRegisterCount() {
  85 + // 准备参数
  86 + IotModbusPointRespDTO point01 = randomPoint(1L, FC_READ_HOLDING_REGISTERS, 0, 100, 1000);
  87 + IotModbusPointRespDTO point02 = randomPoint(2L, FC_READ_HOLDING_REGISTERS, 100, 30, 1000);
  88 + IotModbusDeviceConfigRespDTO config = randomConfig(point01, point02);
  89 +
  90 + // 调用
  91 + List<ReadSegment> segments = IotModbusTcpClientPollScheduler.buildReadSegments(config);
  92 +
  93 + // 断言
  94 + assertEquals(2, segments.size());
  95 + assertEquals(0, segments.get(0).getStartAddress());
  96 + assertEquals(100, segments.get(0).getRegisterCount());
  97 + assertEquals(100, segments.get(1).getStartAddress());
  98 + assertEquals(30, segments.get(1).getRegisterCount());
  99 + }
  100 +
  101 + @Test
  102 + public void testExtractPointRawValues() {
  103 + // 准备参数
  104 + IotModbusPointRespDTO point = randomPoint(1L, FC_READ_HOLDING_REGISTERS, 12, 2, 1000);
  105 + ReadSegment segment = new ReadSegment(FC_READ_HOLDING_REGISTERS, 1000, 10, 4, Collections.singletonList(point));
  106 + int[] rawValues = new int[]{100, 200, 300, 400};
  107 +
  108 + // 调用
  109 + int[] pointRawValues = IotModbusTcpClientPollScheduler.extractPointRawValues(rawValues, segment, point);
  110 +
  111 + // 断言
  112 + assertArrayEquals(new int[]{300, 400}, pointRawValues);
  113 + }
  114 +
  115 + @Test
  116 + public void testExtractPointRawValuesForCoils() {
  117 + // 准备参数
  118 + IotModbusPointRespDTO point = randomPoint(1L, FC_READ_COILS, 3, 2, 1000);
  119 + ReadSegment segment = new ReadSegment(FC_READ_COILS, 1000, 0, 5, Collections.singletonList(point));
  120 + int[] rawValues = new int[]{1, 0, 1, 1, 0, 0, 0, 0}; // 线圈响应可能按字节补齐
  121 +
  122 + // 调用
  123 + int[] pointRawValues = IotModbusTcpClientPollScheduler.extractPointRawValues(rawValues, segment, point);
  124 +
  125 + // 断言
  126 + assertArrayEquals(new int[]{1, 0}, pointRawValues);
  127 + }
  128 +
  129 + private static IotModbusDeviceConfigRespDTO randomConfig(IotModbusPointRespDTO... points) {
  130 + IotModbusDeviceConfigRespDTO config = new IotModbusDeviceConfigRespDTO();
  131 + config.setDeviceId(1L);
  132 + config.setPoints(Arrays.asList(points));
  133 + return config;
  134 + }
  135 +
  136 + private static IotModbusPointRespDTO randomPoint(Long id, Integer functionCode, Integer registerAddress,
  137 + Integer registerCount, Integer pollInterval) {
  138 + IotModbusPointRespDTO point = new IotModbusPointRespDTO();
  139 + point.setId(id);
  140 + point.setFunctionCode(functionCode);
  141 + point.setRegisterAddress(registerAddress);
  142 + point.setRegisterCount(registerCount);
  143 + point.setPollInterval(pollInterval);
  144 + return point;
  145 + }
  146 +
  147 +}