fix(iot): 清理已删除 Modbus 设备的网关轮询

- 过滤 RPC Modbus 配置列表中的已删除设备 - Modbus TCP Server 刷新配置时清理失效设备资源 - 停止轮询任务并清理 Pending 请求、配置缓存和连接

fix(iot): 清理已删除 Modbus 设备的网关轮询
- 过滤 RPC Modbus 配置列表中的已删除设备 - Modbus TCP Server 刷新配置时清理失效设备资源 - 停止轮询任务并清理 Pending 请求、配置缓存和连接
YunaiV
1 parent 4ae3f6b2
Showing 4 changed files with 66 additions and 2 deletions
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/api/device/IoTDeviceApiImpl.java
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpserver/IotModbusTcpServerProtocol.java
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpserver/manager/IotModbusTcpServerConfigCacheService.java
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpserver/manager/IotModbusTcpServerConnectionManager.java
@@ -94,6 +94,9 @@ public class IoTDeviceApiImpl implements IotDeviceCommonApi {
         // 2. 组装返回结果
         Set<Long> deviceIds = convertSet(configList, IotDeviceModbusConfigDO::getDeviceId);
         Map<Long, IotDeviceDO> deviceMap = deviceService.getDeviceMap(deviceIds);
+        if (CollUtil.isEmpty(deviceMap)) {
+            return success(new ArrayList<>());
+        }
         Map<Long, List<IotDeviceModbusPointDO>> pointMap = modbusPointService.getEnabledDeviceModbusPointMapByDeviceIds(deviceIds);
         Map<Long, IotProductDO> productMap = productService.getProductMap(convertSet(deviceMap.values(), IotDeviceDO::getProductId));
         List<IotModbusDeviceConfigRespDTO> result = new ArrayList<>(configList.size());
@@ -139,4 +142,4 @@ public class IoTDeviceApiImpl implements IotDeviceCommonApi {
         return success(deviceService.registerSubDevices(reqDTO));
     }
  
-}
 \ No newline at end of file
+}
@@ -326,9 +326,27 @@ public class IotModbusTcpServerProtocol implements IotProtocol {
                     log.error("[refreshConfig][处理设备配置失败, deviceId={}]", config.getDeviceId(), e);
                 }
             }
+
+            // 3. 清理本轮不再返回配置的已连接设备，避免继续轮询已删除设备的旧点位
+            Set<Long> missingDeviceIds = configCacheService.cleanupMissingConfigs(connectedDeviceIds, configs);
+            for (Long deviceId : missingDeviceIds) {
+                cleanupMissingDevice(deviceId);
+            }
         } catch (Exception e) {
             log.error("[refreshConfig][刷新配置失败]", e);
         }
     }
  
+    private void cleanupMissingDevice(Long deviceId) {
+        try {
+            pollScheduler.stopPolling(deviceId);
+            pendingRequestManager.removeDevice(deviceId);
+            configCacheService.removeConfig(deviceId);
+            connectionManager.closeConnection(deviceId);
+            log.info("[cleanupMissingDevice][设备 {} 配置已失效，已停止轮询并清理连接]", deviceId);
+        } catch (Exception e) {
+            log.error("[cleanupMissingDevice][清理设备失败, deviceId={}]", deviceId, e);
+        }
+    }
+
 }
@@ -12,11 +12,14 @@ import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
  
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
  
+import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertSet;
+
 /**
  * IoT Modbus TCP Server 配置缓存：认证时按需加载，断连时清理，定时刷新已连接设备
  *
@@ -97,6 +100,27 @@ public class IotModbusTcpServerConfigCacheService {
     }
  
     /**
+     * 清理本轮刷新后不再有效的设备配置
+     *
+     * @param refreshedDeviceIds 本轮参与刷新的设备编号
+     * @param currentConfigs     本轮远端返回的有效配置
+     * @return 本轮已不再有效的设备编号
+     */
+    public Set<Long> cleanupMissingConfigs(Set<Long> refreshedDeviceIds,
+                                           List<IotModbusDeviceConfigRespDTO> currentConfigs) {
+        if (CollUtil.isEmpty(refreshedDeviceIds)) {
+            return Collections.emptySet();
+        }
+        Set<Long> currentDeviceIds = convertSet(currentConfigs, IotModbusDeviceConfigRespDTO::getDeviceId);
+        Set<Long> missingDeviceIds = new HashSet<>(refreshedDeviceIds);
+        missingDeviceIds.removeAll(currentDeviceIds);
+        for (Long deviceId : missingDeviceIds) {
+            configCache.remove(deviceId);
+        }
+        return missingDeviceIds;
+    }
+
+    /**
      * 获取设备配置
      */
     public IotModbusDeviceConfigRespDTO getConfig(Long deviceId) {
@@ -9,6 +9,7 @@ import lombok.experimental.Accessors;
 import lombok.extern.slf4j.Slf4j;
  
 import java.util.ArrayList;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -109,7 +110,7 @@ public class IotModbusTcpServerConnectionManager {
      * 获取所有已连接设备的 ID 集合
      */
     public Set<Long> getConnectedDeviceIds() {
-        return deviceSocketMap.keySet();
+        return new HashSet<>(deviceSocketMap.keySet());
     }
  
     /**
@@ -131,6 +132,24 @@ public class IotModbusTcpServerConnectionManager {
     }
  
     /**
+     * 关闭指定设备连接，并先移除映射，避免 closeHandler 再按正常断连发送下线消息
+     */
+    public void closeConnection(Long deviceId) {
+        NetSocket socket = deviceSocketMap.remove(deviceId);
+        if (socket == null) {
+            return;
+        }
+        connectionMap.remove(socket);
+        try {
+            socket.close();
+            log.info("[closeConnection][设备 {} 连接已关闭]", deviceId);
+        } catch (Exception e) {
+            log.warn("[closeConnection][关闭设备连接失败, deviceId={}, remoteAddress={}]",
+                    deviceId, socket.remoteAddress(), e);
+        }
+    }
+
+    /**
      * 发送数据到设备
      *
      * @return 发送结果 Future