From 3bc1bfe1d738549c40f07438372c4842505d863f Mon Sep 17 00:00:00 2001 From: YunaiV Date: Sat, 17 Jan 2026 17:48:29 +0800 Subject: [PATCH] feat：【iot】modbus-tcp 协议接入 40%：1）优化包结构；2）增加 ModbusTcpSlaveSimulatorTest 模拟从站设备 --- yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/api/device/IoTDeviceApiImpl.java | 76 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/config/IotGatewayConfiguration.java | 7 +++++++ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusDataConverter.java | 285 --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpClient.java | 198 ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpConfigCacheService.java | 99 --------------------------------------------------------------------------------------------------- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpConnectionManager.java | 223 ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpDownstreamHandler.java | 117 --------------------------------------------------------------------------------------------------------------------- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpDownstreamSubscriber.java | 1 + yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpPollScheduler.java | 178 ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpUpstreamHandler.java | 62 -------------------------------------------------------------- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpUpstreamProtocol.java | 4 ++++ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/client/IotModbusTcpClient.java | 201 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/codec/IotModbusDataConverter.java | 287 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/manager/IotModbusTcpConfigCacheService.java | 99 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/manager/IotModbusTcpConnectionManager.java | 225 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/manager/IotModbusTcpPollScheduler.java | 180 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/package-info.java | 9 +++++++++ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/router/IotModbusTcpDownstreamHandler.java | 122 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/router/IotModbusTcpUpstreamHandler.java | 63 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/ModbusTcpSlaveSimulatorTest.java | 96 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 20 files changed, 1370 insertions(+), 1162 deletions(-) delete mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusDataConverter.java delete mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpClient.java delete mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpConfigCacheService.java delete mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpConnectionManager.java delete mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpDownstreamHandler.java delete mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpPollScheduler.java delete mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpUpstreamHandler.java create mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/client/IotModbusTcpClient.java create mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/codec/IotModbusDataConverter.java create mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/manager/IotModbusTcpConfigCacheService.java create mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/manager/IotModbusTcpConnectionManager.java create mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/manager/IotModbusTcpPollScheduler.java create mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/package-info.java create mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/router/IotModbusTcpDownstreamHandler.java create mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/router/IotModbusTcpUpstreamHandler.java create mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/ModbusTcpSlaveSimulatorTest.java diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/api/device/IoTDeviceApiImpl.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/api/device/IoTDeviceApiImpl.java index 4a18802..a1aca2a 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/api/device/IoTDeviceApiImpl.java +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/api/device/IoTDeviceApiImpl.java @@ -22,6 +22,7 @@ import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RestController; +import java.math.BigDecimal; import java.util.ArrayList; import java.util.List; import java.util.Map; @@ -74,6 +75,11 @@ public class IoTDeviceApiImpl implements IotDeviceCommonApi { @PostMapping(RpcConstants.RPC_API_PREFIX + "/iot/modbus/enabled-configs") @PermitAll public CommonResult> getEnabledModbusDeviceConfigs() { + // TODO @芋艿：临时 mock 数据，用于测试 ModbusTcpSlaveSimulator + if (true) { + return success(buildMockModbusConfigs()); + } + // 1. 获取所有启用的 Modbus 连接配置 List configList = modbusConfigService.getEnabledDeviceModbusConfigList(); if (CollUtil.isEmpty(configList)) { @@ -105,4 +111,74 @@ public class IoTDeviceApiImpl implements IotDeviceCommonApi { return success(result); } + /** + * 构建 Mock Modbus 配置，对接 ModbusTcpSlaveSimulator + * + * 设备：productKey=4aymZgOTOOCrDKRT, deviceName=small + * 物模型字段：width, height, oneThree + */ + private List buildMockModbusConfigs() { + List configs = new ArrayList<>(); + + // 设备配置 + IotModbusDeviceConfigRespDTO config = new IotModbusDeviceConfigRespDTO(); + config.setDeviceId(1L); + config.setProductKey("4aymZgOTOOCrDKRT"); + config.setDeviceName("small"); + config.setIp("127.0.0.1"); + config.setPort(5020); // 对应 ModbusTcpSlaveSimulator 的端口 + config.setSlaveId(1); + config.setTimeout(3000); + config.setRetryInterval(5000); + + // 点位配置（对应物模型字段：width, height, oneThree） + List points = new ArrayList<>(); + + // 点位 1：width - 读取保持寄存器地址 0（功能码 03） + IotModbusPointRespDTO point1 = new IotModbusPointRespDTO(); + point1.setId(1L); + point1.setIdentifier("width"); + point1.setName("宽度"); + point1.setFunctionCode(3); // 读保持寄存器 + point1.setRegisterAddress(0); + point1.setRegisterCount(1); + point1.setByteOrder("AB"); + point1.setRawDataType("INT16"); + point1.setScale(BigDecimal.ONE); + point1.setPollInterval(3000); // 3 秒轮询 + points.add(point1); + + // 点位 2：height - 读取保持寄存器地址 1（功能码 03） + IotModbusPointRespDTO point2 = new IotModbusPointRespDTO(); + point2.setId(2L); + point2.setIdentifier("height"); + point2.setName("高度"); + point2.setFunctionCode(3); // 读保持寄存器 + point2.setRegisterAddress(1); + point2.setRegisterCount(1); + point2.setByteOrder("AB"); + point2.setRawDataType("INT16"); + point2.setScale(BigDecimal.ONE); + point2.setPollInterval(3000); // 3 秒轮询 + points.add(point2); + + // 点位 3：oneThree - 读取保持寄存器地址 2（功能码 03） + IotModbusPointRespDTO point3 = new IotModbusPointRespDTO(); + point3.setId(3L); + point3.setIdentifier("oneThree"); + point3.setName("一三"); + point3.setFunctionCode(3); // 读保持寄存器 + point3.setRegisterAddress(2); + point3.setRegisterCount(1); + point3.setByteOrder("AB"); + point3.setRawDataType("INT16"); + point3.setScale(BigDecimal.ONE); + point3.setPollInterval(3000); // 3 秒轮询 + points.add(point3); + + config.setPoints(points); + configs.add(config); + return configs; + } + } \ No newline at end of file diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/config/IotGatewayConfiguration.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/config/IotGatewayConfiguration.java index 6641052..04a71e4 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/config/IotGatewayConfiguration.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/config/IotGatewayConfiguration.java @@ -7,6 +7,13 @@ import cn.iocoder.yudao.module.iot.gateway.protocol.emqx.IotEmqxUpstreamProtocol import cn.iocoder.yudao.module.iot.gateway.protocol.http.IotHttpDownstreamSubscriber; import cn.iocoder.yudao.module.iot.gateway.protocol.http.IotHttpUpstreamProtocol; import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.*; +import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.client.IotModbusTcpClient; +import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.codec.IotModbusDataConverter; +import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.manager.IotModbusTcpConfigCacheService; +import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.manager.IotModbusTcpConnectionManager; +import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.manager.IotModbusTcpPollScheduler; +import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.router.IotModbusTcpDownstreamHandler; +import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.router.IotModbusTcpUpstreamHandler; import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttDownstreamSubscriber; import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttUpstreamProtocol; import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.manager.IotMqttConnectionManager; diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusDataConverter.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusDataConverter.java deleted file mode 100644 index 22e06cf..0000000 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusDataConverter.java +++ /dev/null @@ -1,285 +0,0 @@ -package cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp; - -import cn.hutool.core.util.ArrayUtil; -import cn.hutool.core.util.ObjectUtil; -import cn.iocoder.yudao.framework.common.util.object.ObjectUtils; -import cn.iocoder.yudao.module.iot.core.biz.dto.IotModbusPointRespDTO; -import cn.iocoder.yudao.module.iot.core.enums.IotModbusByteOrderEnum; -import cn.iocoder.yudao.module.iot.core.enums.IotModbusRawDataTypeEnum; -import lombok.extern.slf4j.Slf4j; - -import java.math.BigDecimal; -import java.math.RoundingMode; -import java.nio.ByteBuffer; -import java.nio.ByteOrder; - -/** - * IoT Modbus 数据转换器，负责： - * 1. 将 Modbus 原始寄存器值转换为物模型属性值 - * 2. 将物模型属性值转换为 Modbus 原始寄存器值 - * - * @author 芋道源码 - */ -@Slf4j -public class IotModbusDataConverter { - - /** - * 将原始值转换为物模型属性值 - * - * @param rawValues 原始值数组（寄存器值或线圈值） - * @param point 点位配置 - * @return 转换后的属性值 - */ - public Object convertToPropertyValue(int[] rawValues, IotModbusPointRespDTO point) { - if (ArrayUtil.isEmpty(rawValues)) { - return null; - } - String rawDataType = point.getRawDataType(); - String byteOrder = point.getByteOrder(); - BigDecimal scale = ObjectUtil.defaultIfNull(point.getScale(), BigDecimal.ONE); - - // 1. 根据原始数据类型解析原始数值 - Number rawNumber = parseRawValue(rawValues, rawDataType, byteOrder); - if (rawNumber == null) { - return null; - } - - // 2. 应用缩放因子：实际值 = 原始值 × scale - BigDecimal actualValue = new BigDecimal(rawNumber.toString()).multiply(scale); - - // 3. 根据数据类型返回合适的 Java 类型 - return formatValue(actualValue, rawDataType); - } - - /** - * 将物模型属性值转换为原始寄存器值 - * - * @param propertyValue 属性值 - * @param point 点位配置 - * @return 原始值数组 - */ - public int[] convertToRawValues(Object propertyValue, IotModbusPointRespDTO point) { - if (propertyValue == null) { - return new int[0]; - } - String rawDataType = point.getRawDataType(); - String byteOrder = point.getByteOrder(); - BigDecimal scale = ObjectUtil.defaultIfNull(point.getScale(), BigDecimal.ONE); - int registerCount = ObjectUtil.defaultIfNull(point.getRegisterCount(), 1); - - // 1. 转换为 BigDecimal - BigDecimal actualValue = new BigDecimal(propertyValue.toString()); - - // 2. 应用缩放因子：原始值 = 实际值 ÷ scale - BigDecimal rawValue = actualValue.divide(scale, 0, RoundingMode.HALF_UP); - - // 3. 根据原始数据类型编码为寄存器值 - return encodeToRegisters(rawValue, rawDataType, byteOrder, registerCount); - } - - /** - * 解析原始值 - */ - @SuppressWarnings("EnhancedSwitchMigration") - private Number parseRawValue(int[] rawValues, String rawDataType, String byteOrder) { - IotModbusRawDataTypeEnum dataTypeEnum = IotModbusRawDataTypeEnum.getByType(rawDataType); - if (dataTypeEnum == null) { - log.warn("[parseRawValue][不支持的数据类型: {}]", rawDataType); - return rawValues[0]; - } - switch (dataTypeEnum) { - case BOOLEAN: - return rawValues[0] != 0 ? 1 : 0; - case INT16: - return (short) rawValues[0]; - case UINT16: - return rawValues[0] & 0xFFFF; - case INT32: - return parseInt32(rawValues, byteOrder); - case UINT32: - return parseUint32(rawValues, byteOrder); - case FLOAT: - return parseFloat(rawValues, byteOrder); - case DOUBLE: - return parseDouble(rawValues, byteOrder); - default: - log.warn("[parseRawValue][不支持的数据类型: {}]", rawDataType); - return rawValues[0]; - } - } - - private int parseInt32(int[] rawValues, String byteOrder) { - if (rawValues.length < 2) { - return rawValues[0]; - } - byte[] bytes = reorderBytes(registersToBytes(rawValues, 2), byteOrder); - return ByteBuffer.wrap(bytes).order(ByteOrder.BIG_ENDIAN).getInt(); - } - - private long parseUint32(int[] rawValues, String byteOrder) { - if (rawValues.length < 2) { - return rawValues[0] & 0xFFFFFFFFL; - } - byte[] bytes = reorderBytes(registersToBytes(rawValues, 2), byteOrder); - return ByteBuffer.wrap(bytes).order(ByteOrder.BIG_ENDIAN).getInt() & 0xFFFFFFFFL; - } - - private float parseFloat(int[] rawValues, String byteOrder) { - if (rawValues.length < 2) { - return (float) rawValues[0]; - } - byte[] bytes = reorderBytes(registersToBytes(rawValues, 2), byteOrder); - return ByteBuffer.wrap(bytes).order(ByteOrder.BIG_ENDIAN).getFloat(); - } - - private double parseDouble(int[] rawValues, String byteOrder) { - if (rawValues.length < 4) { - return rawValues[0]; - } - byte[] bytes = reorderBytes(registersToBytes(rawValues, 4), byteOrder); - return ByteBuffer.wrap(bytes).order(ByteOrder.BIG_ENDIAN).getDouble(); - } - - /** - * 将寄存器值转换为字节数组 - */ - private byte[] registersToBytes(int[] registers, int count) { - byte[] bytes = new byte[count * 2]; - for (int i = 0; i < Math.min(registers.length, count); i++) { - bytes[i * 2] = (byte) ((registers[i] >> 8) & 0xFF); - bytes[i * 2 + 1] = (byte) (registers[i] & 0xFF); - } - return bytes; - } - - /** - * 根据字节序重排字节 - */ - @SuppressWarnings("EnhancedSwitchMigration") - private byte[] reorderBytes(byte[] bytes, String byteOrder) { - IotModbusByteOrderEnum byteOrderEnum = IotModbusByteOrderEnum.getByOrder(byteOrder); - // null 或者大端序，不需要调整 - if (ObjectUtils.equalsAny(byteOrderEnum, null, IotModbusByteOrderEnum.ABCD, IotModbusByteOrderEnum.AB)) { - return bytes; - } - - // 其他字节序调整 - byte[] result = new byte[bytes.length]; - switch (byteOrderEnum) { - case BA: // 小端序（16 位） - if (bytes.length >= 2) { - result[0] = bytes[1]; - result[1] = bytes[0]; - } - break; - case CDAB: // 大端字交换（32 位） - if (bytes.length >= 4) { - result[0] = bytes[2]; - result[1] = bytes[3]; - result[2] = bytes[0]; - result[3] = bytes[1]; - } - break; - case DCBA: // 小端序（32 位） - if (bytes.length >= 4) { - result[0] = bytes[3]; - result[1] = bytes[2]; - result[2] = bytes[1]; - result[3] = bytes[0]; - } - break; - case BADC: // 小端字交换（32 位） - if (bytes.length >= 4) { - result[0] = bytes[1]; - result[1] = bytes[0]; - result[2] = bytes[3]; - result[3] = bytes[2]; - } - break; - default: - return bytes; - } - return result; - } - - /** - * 编码为寄存器值 - */ - @SuppressWarnings("EnhancedSwitchMigration") - private int[] encodeToRegisters(BigDecimal rawValue, String rawDataType, String byteOrder, int registerCount) { - IotModbusRawDataTypeEnum dataTypeEnum = IotModbusRawDataTypeEnum.getByType(rawDataType); - if (dataTypeEnum == null) { - return new int[]{rawValue.intValue()}; - } - switch (dataTypeEnum) { - case BOOLEAN: - return new int[]{rawValue.intValue() != 0 ? 1 : 0}; - case INT16: - case UINT16: - return new int[]{rawValue.intValue() & 0xFFFF}; - case INT32: - case UINT32: - return encodeInt32(rawValue.intValue(), byteOrder); - case FLOAT: - return encodeFloat(rawValue.floatValue(), byteOrder); - case DOUBLE: - return encodeDouble(rawValue.doubleValue(), byteOrder); - default: - return new int[]{rawValue.intValue()}; - } - } - - private int[] encodeInt32(int value, String byteOrder) { - byte[] bytes = ByteBuffer.allocate(4).order(ByteOrder.BIG_ENDIAN).putInt(value).array(); - bytes = reorderBytes(bytes, byteOrder); - return bytesToRegisters(bytes); - } - - private int[] encodeFloat(float value, String byteOrder) { - byte[] bytes = ByteBuffer.allocate(4).order(ByteOrder.BIG_ENDIAN).putFloat(value).array(); - bytes = reorderBytes(bytes, byteOrder); - return bytesToRegisters(bytes); - } - - private int[] encodeDouble(double value, String byteOrder) { - byte[] bytes = ByteBuffer.allocate(8).order(ByteOrder.BIG_ENDIAN).putDouble(value).array(); - bytes = reorderBytes(bytes, byteOrder); - return bytesToRegisters(bytes); - } - - private int[] bytesToRegisters(byte[] bytes) { - int[] registers = new int[bytes.length / 2]; - for (int i = 0; i < registers.length; i++) { - registers[i] = ((bytes[i * 2] & 0xFF) << 8) | (bytes[i * 2 + 1] & 0xFF); - } - return registers; - } - - /** - * 格式化返回值 - */ - @SuppressWarnings("EnhancedSwitchMigration") - private Object formatValue(BigDecimal value, String rawDataType) { - IotModbusRawDataTypeEnum dataTypeEnum = IotModbusRawDataTypeEnum.getByType(rawDataType); - if (dataTypeEnum == null) { - return value; - } - switch (dataTypeEnum) { - case BOOLEAN: - return value.intValue() != 0; - case INT16: - case INT32: - return value.intValue(); - case UINT16: - case UINT32: - return value.longValue(); - case FLOAT: - return value.floatValue(); - case DOUBLE: - return value.doubleValue(); - default: - return value; - } - } - -} diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpClient.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpClient.java deleted file mode 100644 index e42002f..0000000 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpClient.java +++ /dev/null @@ -1,198 +0,0 @@ -package cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp; - -import cn.iocoder.yudao.module.iot.core.biz.dto.IotModbusPointRespDTO; -import cn.iocoder.yudao.module.iot.core.enums.IotModbusFunctionCodeEnum; -import com.ghgande.j2mod.modbus.io.ModbusTCPTransaction; -import com.ghgande.j2mod.modbus.msg.*; -import com.ghgande.j2mod.modbus.procimg.InputRegister; -import com.ghgande.j2mod.modbus.procimg.Register; -import com.ghgande.j2mod.modbus.procimg.SimpleRegister; -import com.ghgande.j2mod.modbus.util.BitVector; -import io.vertx.core.Future; -import lombok.extern.slf4j.Slf4j; - -/** - * IoT Modbus TCP 客户端，负责： - * 1. 封装 Modbus 读/写操作 - * 2. 根据功能码，执行对应的 Modbus 请求 - * - * @author 芋道源码 - */ -@Slf4j -public class IotModbusTcpClient { - - /** - * 读取 Modbus 数据 - * - * @param connection Modbus 连接 - * @param slaveId 从站地址 - * @param point 点位配置 - * @return 原始值（int 数组） - */ - public Future read(IotModbusTcpConnectionManager.ModbusConnection connection, - Integer slaveId, - IotModbusPointRespDTO point) { - return connection.executeBlocking(tcpConnection -> { - try { - // 1. 创建请求 - ModbusRequest request = createReadRequest(point.getFunctionCode(), - point.getRegisterAddress(), point.getRegisterCount()); - request.setUnitID(slaveId); - - // 2. 执行事务（请求） - ModbusTCPTransaction transaction = new ModbusTCPTransaction(tcpConnection); - transaction.setRequest(request); - transaction.execute(); - - // 3. 解析响应 - ModbusResponse response = transaction.getResponse(); - return extractValues(response, point.getFunctionCode()); - } catch (Exception e) { - throw new RuntimeException(String.format("Modbus 读取失败 [slaveId=%d, identifier=%s, address=%d]", - slaveId, point.getIdentifier(), point.getRegisterAddress()), e); - } - }); - } - - /** - * 写入 Modbus 数据 - * - * @param connection Modbus 连接 - * @param slaveId 从站地址 - * @param point 点位配置 - * @param values 要写入的值 - * @return 是否成功 - */ - public Future write(IotModbusTcpConnectionManager.ModbusConnection connection, - Integer slaveId, - IotModbusPointRespDTO point, - int[] values) { - return connection.executeBlocking(tcpConnection -> { - try { - // 1. 创建请求 - ModbusRequest request = createWriteRequest(point.getFunctionCode(), - point.getRegisterAddress(), point.getRegisterCount(), values); - if (request == null) { - throw new RuntimeException("功能码 " + point.getFunctionCode() + " 不支持写操作"); - } - request.setUnitID(slaveId); - - // 2. 执行事务（请求） - ModbusTCPTransaction transaction = new ModbusTCPTransaction(tcpConnection); - transaction.setRequest(request); - transaction.execute(); - return true; - } catch (Exception e) { - throw new RuntimeException(String.format("Modbus 写入失败 [slaveId=%d, identifier=%s, address=%d]", - slaveId, point.getIdentifier(), point.getRegisterAddress()), e); - } - }); - } - - /** - * 创建读取请求 - */ - @SuppressWarnings("EnhancedSwitchMigration") - private ModbusRequest createReadRequest(Integer functionCode, Integer address, Integer count) { - IotModbusFunctionCodeEnum functionCodeEnum = IotModbusFunctionCodeEnum.valueOf(functionCode); - if (functionCodeEnum == null) { - throw new IllegalArgumentException("不支持的功能码: " + functionCode); - } - switch (functionCodeEnum) { - case READ_COILS: - return new ReadCoilsRequest(address, count); - case READ_DISCRETE_INPUTS: - return new ReadInputDiscretesRequest(address, count); - case READ_HOLDING_REGISTERS: - return new ReadMultipleRegistersRequest(address, count); - case READ_INPUT_REGISTERS: - return new ReadInputRegistersRequest(address, count); - default: - throw new IllegalArgumentException("不支持的功能码: " + functionCode); - } - } - - /** - * 创建写入请求 - */ - private ModbusRequest createWriteRequest(Integer functionCode, Integer address, Integer count, int[] values) { - IotModbusFunctionCodeEnum functionCodeEnum = IotModbusFunctionCodeEnum.valueOf(functionCode); - if (functionCodeEnum == null) { - throw new IllegalArgumentException("不支持的功能码: " + functionCode); - } - switch (functionCodeEnum) { - case READ_COILS: // 写线圈（使用功能码 5 或 15） - if (count == 1) { - return new WriteCoilRequest(address, values[0] != 0); - } else { - BitVector bv = new BitVector(count); - for (int i = 0; i < Math.min(values.length, count); i++) { - bv.setBit(i, values[i] != 0); - } - return new WriteMultipleCoilsRequest(address, bv); - } - case READ_HOLDING_REGISTERS: // 写保持寄存器（使用功能码 6 或 16） - if (count == 1) { - return new WriteSingleRegisterRequest(address, new SimpleRegister(values[0])); - } else { - Register[] registers = new SimpleRegister[count]; - for (int i = 0; i < count; i++) { - registers[i] = new SimpleRegister(i < values.length ? values[i] : 0); - } - return new WriteMultipleRegistersRequest(address, registers); - } - case READ_DISCRETE_INPUTS: // 只读 - case READ_INPUT_REGISTERS: // 只读 - return null; - default: - throw new IllegalArgumentException("不支持的功能码: " + functionCode); - } - } - - /** - * 从响应中提取值 - */ - private int[] extractValues(ModbusResponse response, Integer functionCode) { - IotModbusFunctionCodeEnum functionCodeEnum = IotModbusFunctionCodeEnum.valueOf(functionCode); - if (functionCodeEnum == null) { - throw new IllegalArgumentException("不支持的功能码: " + functionCode); - } - switch (functionCodeEnum) { - case READ_COILS: - ReadCoilsResponse coilsResponse = (ReadCoilsResponse) response; - int bitCount = coilsResponse.getBitCount(); - int[] coilValues = new int[bitCount]; - for (int i = 0; i < bitCount; i++) { - coilValues[i] = coilsResponse.getCoilStatus(i) ? 1 : 0; - } - return coilValues; - case READ_DISCRETE_INPUTS: - ReadInputDiscretesResponse discretesResponse = (ReadInputDiscretesResponse) response; - int discreteCount = discretesResponse.getBitCount(); - int[] discreteValues = new int[discreteCount]; - for (int i = 0; i < discreteCount; i++) { - discreteValues[i] = discretesResponse.getDiscreteStatus(i) ? 1 : 0; - } - return discreteValues; - case READ_HOLDING_REGISTERS: - ReadMultipleRegistersResponse holdingResponse = (ReadMultipleRegistersResponse) response; - InputRegister[] holdingRegisters = holdingResponse.getRegisters(); - int[] holdingValues = new int[holdingRegisters.length]; - for (int i = 0; i < holdingRegisters.length; i++) { - holdingValues[i] = holdingRegisters[i].getValue(); - } - return holdingValues; - case READ_INPUT_REGISTERS: - ReadInputRegistersResponse inputResponse = (ReadInputRegistersResponse) response; - InputRegister[] inputRegisters = inputResponse.getRegisters(); - int[] inputValues = new int[inputRegisters.length]; - for (int i = 0; i < inputRegisters.length; i++) { - inputValues[i] = inputRegisters[i].getValue(); - } - return inputValues; - default: - throw new IllegalArgumentException("不支持的功能码: " + functionCode); - } - } - -} diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpConfigCacheService.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpConfigCacheService.java deleted file mode 100644 index 5422af1..0000000 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpConfigCacheService.java +++ /dev/null @@ -1,99 +0,0 @@ -package cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp; - -import cn.iocoder.yudao.framework.common.pojo.CommonResult; -import cn.iocoder.yudao.module.iot.core.biz.IotDeviceCommonApi; -import cn.iocoder.yudao.module.iot.core.biz.dto.IotModbusDeviceConfigRespDTO; -import lombok.RequiredArgsConstructor; -import lombok.extern.slf4j.Slf4j; - -import java.util.*; -import java.util.concurrent.ConcurrentHashMap; -import java.util.function.Consumer; - -import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertSet; - -/** - * IoT Modbus TCP 配置缓存服务，负责：从 biz 拉取 Modbus 设备配置，缓存配置数据，并检测配置变更 - * - * @author 芋道源码 - */ -@RequiredArgsConstructor -@Slf4j -public class IotModbusTcpConfigCacheService { - - private final IotDeviceCommonApi deviceApi; - - /** - * 配置缓存：deviceId -> 配置 - */ - private final Map configCache = new ConcurrentHashMap<>(); - - /** - * 已知的设备 ID 集合（作用：用于检测已删除的设备） - * - * @see #cleanupRemovedDevices(List, Consumer) - */ - private final Set knownDeviceIds = ConcurrentHashMap.newKeySet(); - - /** - * 刷新配置 - * - * @return 最新的配置列表 - */ - public List refreshConfig() { - try { - // 1. 从远程获取配置 - CommonResult> result = deviceApi.getEnabledModbusDeviceConfigs(); - if (result == null || !result.isSuccess() || result.getData() == null) { - log.warn("[refreshConfig][获取 Modbus 配置失败: {}]", result); - return new ArrayList<>(configCache.values()); - } - List configs = result.getData(); - - // 2. 更新缓存（注意：不在这里更新 knownDeviceIds，由 cleanupRemovedDevices 统一管理） - for (IotModbusDeviceConfigRespDTO config : configs) { - configCache.put(config.getDeviceId(), config); - } - return configs; - } catch (Exception e) { - log.error("[refreshConfig][刷新配置失败]", e); - return new ArrayList<>(configCache.values()); - } - } - - /** - * 获取设备配置 - * - * @param deviceId 设备 ID - * @return 配置 - */ - public IotModbusDeviceConfigRespDTO getConfig(Long deviceId) { - return configCache.get(deviceId); - } - - /** - * 清理已删除设备的资源，并更新已知设备 ID 集合 - * - * @param currentConfigs 当前有效的配置列表 - * @param cleanupAction 清理动作 - */ - public void cleanupRemovedDevices(List currentConfigs, Consumer cleanupAction) { - // 1.1 获取当前有效的设备 ID - Set currentDeviceIds = convertSet(currentConfigs, IotModbusDeviceConfigRespDTO::getDeviceId); - // 1.2 找出已删除的设备（基于旧的 knownDeviceIds） - Set removedDeviceIds = new HashSet<>(knownDeviceIds); - removedDeviceIds.removeAll(currentDeviceIds); - - // 2. 清理已删除设备 - for (Long deviceId : removedDeviceIds) { - log.info("[cleanupRemovedDevices][清理已删除设备: {}]", deviceId); - configCache.remove(deviceId); - cleanupAction.accept(deviceId); - } - - // 3. 更新已知设备 ID 集合为当前有效的设备 ID - knownDeviceIds.clear(); - knownDeviceIds.addAll(currentDeviceIds); - } - -} diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpConnectionManager.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpConnectionManager.java deleted file mode 100644 index 8452571..0000000 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpConnectionManager.java +++ /dev/null @@ -1,223 +0,0 @@ -package cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp; - -import cn.iocoder.yudao.module.iot.core.biz.dto.IotModbusDeviceConfigRespDTO; -import com.ghgande.j2mod.modbus.net.TCPMasterConnection; -import io.vertx.core.Context; -import io.vertx.core.Future; -import io.vertx.core.Vertx; -import lombok.Data; -import lombok.RequiredArgsConstructor; -import lombok.extern.slf4j.Slf4j; -import org.redisson.api.RLock; -import org.redisson.api.RedissonClient; - -import java.net.InetAddress; -import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; - -/** - * IoT Modbus TCP 连接管理器，负责： - * 1. 管理 TCP 连接（相同 ip:port 共用连接） - * 2. 分布式锁管理（连接级别），避免多节点重复创建连接 - * 3. 连接重试和故障恢复 - * - * @author 芋道源码 - */ -@RequiredArgsConstructor -@Slf4j -public class IotModbusTcpConnectionManager { - - private static final String LOCK_KEY_PREFIX = "iot:modbus-tcp:connection:"; - - private final RedissonClient redissonClient; - private final Vertx vertx; - - /** - * 连接池：key = ip:port - */ - private final Map connectionPool = new ConcurrentHashMap<>(); - - /** - * 设备 ID 到连接 key 的映射 - */ - private final Map deviceConnectionMap = new ConcurrentHashMap<>(); - - /** - * 确保连接存在 - * - * @param config 设备配置 - */ - public void ensureConnection(IotModbusDeviceConfigRespDTO config) { - // 1. 记录设备与连接的关系 - String connectionKey = buildConnectionKey(config.getIp(), config.getPort()); - deviceConnectionMap.put(config.getDeviceId(), connectionKey); - - // 2. 情况一：连接已存在，添加设备引用 - ModbusConnection connection = connectionPool.get(connectionKey); - if (connection != null) { - connection.addDevice(config.getDeviceId(), config.getSlaveId()); - return; - } - - // 3. 情况二：连接不存在，创建新连接 - // 3.1 尝试获取分布式锁 - RLock lock = redissonClient.getLock(LOCK_KEY_PREFIX + connectionKey); - if (!lock.tryLock()) { - log.debug("[ensureConnection][获取锁失败, 由其他节点负责: {}]", connectionKey); - return; - } - // 3.2 创建新连接 - try { - connection = createConnection(config, lock); - connectionPool.put(connectionKey, connection); - log.info("[ensureConnection][创建 Modbus 连接成功: {}]", connectionKey); - } catch (Exception e) { - log.error("[ensureConnection][创建 Modbus 连接失败: {}]", connectionKey, e); - lock.unlock(); - } - } - - /** - * 创建 Modbus TCP 连接 - */ - private ModbusConnection createConnection(IotModbusDeviceConfigRespDTO config, RLock lock) throws Exception { - // 1. 创建 TCP 连接 - TCPMasterConnection tcpConnection = new TCPMasterConnection(InetAddress.getByName(config.getIp())); - tcpConnection.setPort(config.getPort()); - tcpConnection.setTimeout(config.getTimeout()); - tcpConnection.connect(); - - // 2. 创建 Modbus 连接对象 - ModbusConnection connection = new ModbusConnection() - .setConnectionKey(buildConnectionKey(config.getIp(), config.getPort())) - .setTcpConnection(tcpConnection).setLock(lock).setContext(vertx.getOrCreateContext()) - .setTimeout(config.getTimeout()).setRetryInterval(config.getRetryInterval()); - connection.addDevice(config.getDeviceId(), config.getSlaveId()); - return connection; - } - - /** - * 获取连接 - */ - public ModbusConnection getConnection(Long deviceId) { - String connectionKey = deviceConnectionMap.get(deviceId); - if (connectionKey == null) { - return null; - } - return connectionPool.get(connectionKey); - } - - /** - * 获取设备的 slave ID - */ - public Integer getSlaveId(Long deviceId) { - ModbusConnection connection = getConnection(deviceId); - if (connection == null) { - return null; - } - return connection.getSlaveId(deviceId); - } - - /** - * 移除设备 - */ - public void removeDevice(Long deviceId) { - // 1. 移除设备引用 - String connectionKey = deviceConnectionMap.remove(deviceId); - if (connectionKey == null) { - return; - } - ModbusConnection connection = connectionPool.get(connectionKey); - if (connection == null) { - return; - } - connection.removeDevice(deviceId); - - // 2. 如果没有设备引用了，关闭连接 - if (connection.getDeviceCount() == 0) { - closeConnection(connectionKey); - } - } - - /** - * 关闭指定连接 - */ - private void closeConnection(String connectionKey) { - ModbusConnection connection = connectionPool.remove(connectionKey); - if (connection == null) { - return; - } - - try { - if (connection.getTcpConnection() != null) { - connection.getTcpConnection().close(); - } - // 安全释放锁：先检查锁存在且被锁定，再检查是否当前线程持有 - RLock lock = connection.getLock(); - if (lock != null && lock.isLocked() && lock.isHeldByCurrentThread()) { - lock.unlock(); - } - log.info("[closeConnection][关闭 Modbus 连接: {}]", connectionKey); - } catch (Exception e) { - log.error("[closeConnection][关闭连接失败: {}]", connectionKey, e); - } - } - - /** - * 关闭所有连接 - */ - public void closeAll() { - for (String connectionKey : connectionPool.keySet()) { - closeConnection(connectionKey); - } - deviceConnectionMap.clear(); - } - - private String buildConnectionKey(String ip, Integer port) { - return ip + ":" + port; - } - - /** - * Modbus 连接信息 - */ - @Data - public static class ModbusConnection { - - private String connectionKey; - private TCPMasterConnection tcpConnection; - private RLock lock; - private Integer timeout; - private Integer retryInterval; - private Context context; - - /** - * 设备 ID 到 slave ID 的映射 - */ - private final Map deviceSlaveMap = new ConcurrentHashMap<>(); - - public void addDevice(Long deviceId, Integer slaveId) { - deviceSlaveMap.put(deviceId, slaveId); - } - - public void removeDevice(Long deviceId) { - deviceSlaveMap.remove(deviceId); - } - - public int getDeviceCount() { - return deviceSlaveMap.size(); - } - - public Integer getSlaveId(Long deviceId) { - return deviceSlaveMap.get(deviceId); - } - - /** - * 执行 Modbus 读取操作（阻塞方式，在 Vert.x worker 线程执行） - */ - public Future executeBlocking(java.util.function.Function operation) { - // ordered=true 保证同一 Context 的操作串行执行，不同连接之间可并行 - return context.executeBlocking(() -> operation.apply(tcpConnection), true); - } - } - -} diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpDownstreamHandler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpDownstreamHandler.java deleted file mode 100644 index 5ed226a..0000000 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpDownstreamHandler.java +++ /dev/null @@ -1,117 +0,0 @@ -package cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp; - -import cn.hutool.core.collection.CollUtil; -import cn.iocoder.yudao.module.iot.core.biz.dto.IotModbusDeviceConfigRespDTO; -import cn.iocoder.yudao.module.iot.core.biz.dto.IotModbusPointRespDTO; -import cn.iocoder.yudao.module.iot.core.enums.IotModbusFunctionCodeEnum; -import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; -import lombok.RequiredArgsConstructor; -import lombok.extern.slf4j.Slf4j; - -import java.util.Map; - -/** - * IoT Modbus TCP 下行消息处理器 - * - * 负责： - * 1. 处理下行消息（如属性设置 thing.service.property.set） - * 2. 执行 Modbus 写入操作 - * - * @author 芋道源码 - */ -@RequiredArgsConstructor -@Slf4j -public class IotModbusTcpDownstreamHandler { - - private final IotModbusTcpConnectionManager connectionManager; - private final IotModbusTcpClient modbusClient; - private final IotModbusDataConverter dataConverter; - private final IotModbusTcpConfigCacheService configCacheService; - - /** - * 处理下行消息 - */ - @SuppressWarnings("unchecked") - public void handle(IotDeviceMessage message) { - // 1.1 检查是否是属性设置消息 - if (!"thing.service.property.set".equals(message.getMethod())) { - log.debug("[handle][忽略非属性设置消息: {}]", message.getMethod()); - return; - } - // 1.2 获取设备配置 - IotModbusDeviceConfigRespDTO config = configCacheService.getConfig(message.getDeviceId()); - if (config == null) { - log.warn("[handle][设备 {} 没有 Modbus 配置]", message.getDeviceId()); - return; - } - - // 2. 解析属性值并写入 - Object params = message.getParams(); - if (!(params instanceof Map)) { - log.warn("[handle][params 不是 Map 类型: {}]", params); - return; - } - Map propertyMap = (Map) params; - for (Map.Entry entry : propertyMap.entrySet()) { - String identifier = entry.getKey(); - Object value = entry.getValue(); - // 2.1.1 查找对应的点位配置 - IotModbusPointRespDTO point = findPoint(config, identifier); - if (point == null) { - log.warn("[handle][设备 {} 没有点位配置: {}]", message.getDeviceId(), identifier); - continue; - } - // 2.1.2 检查是否支持写操作 - if (!isWritable(point.getFunctionCode())) { - log.warn("[handle][点位 {} 不支持写操作, 功能码={}]", identifier, point.getFunctionCode()); - continue; - } - - // 2.2 执行写入 - writeProperty(config, point, value); - } - } - - /** - * 写入属性值 - */ - private void writeProperty(IotModbusDeviceConfigRespDTO config, IotModbusPointRespDTO point, Object value) { - // 1.1 获取连接 - IotModbusTcpConnectionManager.ModbusConnection connection = connectionManager.getConnection(config.getDeviceId()); - if (connection == null) { - log.warn("[writeProperty][设备 {} 没有连接]", config.getDeviceId()); - return; - } - // 1.2 获取 slave ID - Integer slaveId = connectionManager.getSlaveId(config.getDeviceId()); - if (slaveId == null) { - log.warn("[writeProperty][设备 {} 没有 slaveId]", config.getDeviceId()); - return; - } - - // 2.1 转换属性值为原始值 - int[] rawValues = dataConverter.convertToRawValues(value, point); - // 2.2 执行 Modbus 写入 - modbusClient.write(connection, slaveId, point, rawValues) - .onSuccess(success -> log.info("[writeProperty][写入成功, deviceId={}, identifier={}, value={}]", - config.getDeviceId(), point.getIdentifier(), value)) - .onFailure(e -> log.error("[writeProperty][写入失败, deviceId={}, identifier={}]", - config.getDeviceId(), point.getIdentifier(), e)); - } - - /** - * 查找点位配置 - */ - private IotModbusPointRespDTO findPoint(IotModbusDeviceConfigRespDTO config, String identifier) { - return CollUtil.findOne(config.getPoints(), p -> identifier.equals(p.getIdentifier())); - } - - /** - * 检查功能码是否支持写操作 - */ - private boolean isWritable(Integer functionCode) { - IotModbusFunctionCodeEnum functionCodeEnum = IotModbusFunctionCodeEnum.valueOf(functionCode); - return functionCodeEnum != null && Boolean.TRUE.equals(functionCodeEnum.getWritable()); - } - -} diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpDownstreamSubscriber.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpDownstreamSubscriber.java index 919669a..a051013 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpDownstreamSubscriber.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpDownstreamSubscriber.java @@ -4,6 +4,7 @@ import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus; import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageSubscriber; import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils; +import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.router.IotModbusTcpDownstreamHandler; import jakarta.annotation.PostConstruct; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpPollScheduler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpPollScheduler.java deleted file mode 100644 index 84516d1..0000000 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpPollScheduler.java +++ /dev/null @@ -1,178 +0,0 @@ -package cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp; - -import cn.hutool.core.collection.CollUtil; -import cn.iocoder.yudao.module.iot.core.biz.dto.IotModbusDeviceConfigRespDTO; -import cn.iocoder.yudao.module.iot.core.biz.dto.IotModbusPointRespDTO; -import io.vertx.core.Vertx; -import lombok.AllArgsConstructor; -import lombok.Data; -import lombok.RequiredArgsConstructor; -import lombok.extern.slf4j.Slf4j; - -import java.util.*; -import java.util.concurrent.ConcurrentHashMap; - -import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertSet; - -/** - * IoT Modbus TCP 轮询调度器：管理点位的轮询定时器，调度读取任务并上报结果 - * - * @author 芋道源码 - */ -@RequiredArgsConstructor -@Slf4j -public class IotModbusTcpPollScheduler { - - private final Vertx vertx; - private final IotModbusTcpConnectionManager connectionManager; - private final IotModbusTcpClient modbusClient; - private final IotModbusTcpUpstreamHandler upstreamHandler; - - /** - * 设备点位的定时器映射：deviceId -> (pointId -> PointTimerInfo) - */ - private final Map> devicePointTimers = new ConcurrentHashMap<>(); - - /** - * 点位定时器信息 - */ - @Data - @AllArgsConstructor - private static class PointTimerInfo { - - /** - * Vert.x 定时器 ID - */ - private Long timerId; - /** - * 轮询间隔（用于判断是否需要更新定时器） - */ - private Integer pollInterval; - - } - - /** - * 更新轮询任务（增量更新） - * - * 1. 【删除】点位：停止对应的轮询定时器 - * 2. 【新增】点位：创建对应的轮询定时器 - * 3. 【修改】点位：pollInterval 变化，重建对应的轮询定时器 - * 4. 其他属性变化（包括未变化的）：不处理（下次轮询时自动使用新配置） - */ - public void updatePolling(IotModbusDeviceConfigRespDTO config) { - Long deviceId = config.getDeviceId(); - List newPoints = config.getPoints(); - Map currentTimers = devicePointTimers - .computeIfAbsent(deviceId, k -> new ConcurrentHashMap<>()); - // 1.1 计算新配置（包括新增和修改的点位）中的点位 ID 集合 - Set newPointIds = convertSet(newPoints, IotModbusPointRespDTO::getId); - // 1.2 计算删除的点位 ID 集合 - Set removedPointIds = new HashSet<>(currentTimers.keySet()); - removedPointIds.removeAll(newPointIds); - - // 2. 处理删除的点位：停止不再存在的定时器 - for (Long pointId : removedPointIds) { - PointTimerInfo timerInfo = currentTimers.remove(pointId); - if (timerInfo != null) { - vertx.cancelTimer(timerInfo.getTimerId()); - log.debug("[updatePolling][设备 {} 点位 {} 定时器已删除]", deviceId, pointId); - } - } - - // 3. 处理新增和修改的点位 - if (CollUtil.isEmpty(newPoints)) { - return; - } - for (IotModbusPointRespDTO point : newPoints) { - Long pointId = point.getId(); - Integer newPollInterval = point.getPollInterval(); - PointTimerInfo existingTimer = currentTimers.get(pointId); - // 3.1 新增点位：创建定时器 - if (existingTimer == null) { - Long timerId = createPollTimer(config, point); - if (timerId != null) { - currentTimers.put(pointId, new PointTimerInfo(timerId, newPollInterval)); - log.debug("[updatePolling][设备 {} 点位 {} 定时器已创建, interval={}ms]", - deviceId, pointId, newPollInterval); - } - } else if (!Objects.equals(existingTimer.getPollInterval(), newPollInterval)) { - // 3.2 pollInterval 变化：重建定时器 - vertx.cancelTimer(existingTimer.getTimerId()); - Long timerId = createPollTimer(config, point); - if (timerId != null) { - currentTimers.put(pointId, new PointTimerInfo(timerId, newPollInterval)); - log.debug("[updatePolling][设备 {} 点位 {} 定时器已更新, interval={}ms -> {}ms]", - deviceId, pointId, existingTimer.getPollInterval(), newPollInterval); - } else { - currentTimers.remove(pointId); - } - } - // 3.3 其他属性变化：不处理（下次轮询时自动使用新配置） - } - } - - /** - * 创建轮询定时器 - */ - private Long createPollTimer(IotModbusDeviceConfigRespDTO config, IotModbusPointRespDTO point) { - if (point.getPollInterval() == null || point.getPollInterval() <= 0) { - return null; - } - return vertx.setPeriodic(point.getPollInterval(), timerId -> { - try { - pollPoint(config, point); - } catch (Exception e) { - log.error("[createPollTimer][轮询点位失败, deviceId={}, identifier={}]", - config.getDeviceId(), point.getIdentifier(), e); - } - }); - } - - /** - * 轮询单个点位 - */ - private void pollPoint(IotModbusDeviceConfigRespDTO config, IotModbusPointRespDTO point) { - // 1.1 获取连接 - IotModbusTcpConnectionManager.ModbusConnection connection = connectionManager.getConnection(config.getDeviceId()); - if (connection == null) { - log.warn("[pollPoint][设备 {} 没有连接]", config.getDeviceId()); - return; - } - // 1.2 获取 slave ID - Integer slaveId = connectionManager.getSlaveId(config.getDeviceId()); - if (slaveId == null) { - log.warn("[pollPoint][设备 {} 没有 slaveId]", config.getDeviceId()); - return; - } - - // 2. 执行 Modbus 读取 - modbusClient.read(connection, slaveId, point) - .onSuccess(rawValue -> upstreamHandler.handleReadResult(config, point, rawValue)) - .onFailure(e -> log.error("[pollPoint][读取点位失败, deviceId={}, identifier={}]", - config.getDeviceId(), point.getIdentifier(), e)); - } - - /** - * 停止设备的轮询 - */ - public void stopPolling(Long deviceId) { - Map timers = devicePointTimers.remove(deviceId); - if (CollUtil.isEmpty(timers)) { - return; - } - for (PointTimerInfo timerInfo : timers.values()) { - vertx.cancelTimer(timerInfo.getTimerId()); - } - log.debug("[stopPolling][设备 {} 停止了 {} 个轮询定时器]", deviceId, timers.size()); - } - - /** - * 停止所有轮询 - */ - public void stopAll() { - for (Long deviceId : new ArrayList<>(devicePointTimers.keySet())) { - stopPolling(deviceId); - } - } - -} diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpUpstreamHandler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpUpstreamHandler.java deleted file mode 100644 index 95ac90d..0000000 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpUpstreamHandler.java +++ /dev/null @@ -1,62 +0,0 @@ -package cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp; - -import cn.iocoder.yudao.module.iot.core.biz.dto.IotModbusDeviceConfigRespDTO; -import cn.iocoder.yudao.module.iot.core.biz.dto.IotModbusPointRespDTO; -import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; -import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService; -import lombok.Setter; -import lombok.extern.slf4j.Slf4j; - -import java.util.HashMap; -import java.util.Map; - -/** - * IoT Modbus TCP 上行数据处理器：将原始值转换为物模型属性值并上报 - * - * @author 芋道源码 - */ -@Slf4j -public class IotModbusTcpUpstreamHandler { - - private final IotDeviceMessageService messageService; - private final IotModbusDataConverter dataConverter; - - @Setter - private String serverId; - - public IotModbusTcpUpstreamHandler(IotDeviceMessageService messageService, - IotModbusDataConverter dataConverter) { - this.messageService = messageService; - this.dataConverter = dataConverter; - } - - /** - * 处理 Modbus 读取结果 - * - * @param config 设备配置 - * @param point 点位配置 - * @param rawValue 原始值（int 数组） - */ - public void handleReadResult(IotModbusDeviceConfigRespDTO config, - IotModbusPointRespDTO point, - int[] rawValue) { - try { - // 1.1 转换原始值为物模型属性值 - Object convertedValue = dataConverter.convertToPropertyValue(rawValue, point); - log.debug("[handleReadResult][设备={}, 属性={}, 原始值={}, 转换值={}]", - config.getDeviceId(), point.getIdentifier(), rawValue, convertedValue); - // 1.2 构造属性上报消息 - Map params = new HashMap<>(); - params.put(point.getIdentifier(), convertedValue); - IotDeviceMessage message = IotDeviceMessage.requestOf("thing.event.property.post", params); - - // 2. 发送到消息总线 - messageService.sendDeviceMessage(message, config.getProductKey(), - config.getDeviceName(), serverId); - } catch (Exception e) { - log.error("[handleReadResult][处理读取结果失败, deviceId={}, identifier={}]", - config.getDeviceId(), point.getIdentifier(), e); - } - } - -} diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpUpstreamProtocol.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpUpstreamProtocol.java index 43b9ad5..1d47d30 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpUpstreamProtocol.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/IotModbusTcpUpstreamProtocol.java @@ -2,6 +2,10 @@ package cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp; import cn.iocoder.yudao.module.iot.core.biz.dto.IotModbusDeviceConfigRespDTO; import cn.iocoder.yudao.module.iot.gateway.config.IotGatewayProperties; +import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.manager.IotModbusTcpConfigCacheService; +import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.manager.IotModbusTcpConnectionManager; +import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.manager.IotModbusTcpPollScheduler; +import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.router.IotModbusTcpUpstreamHandler; import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService; import io.vertx.core.Vertx; import jakarta.annotation.PostConstruct; diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/client/IotModbusTcpClient.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/client/IotModbusTcpClient.java new file mode 100644 index 0000000..951798a --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/client/IotModbusTcpClient.java @@ -0,0 +1,201 @@ +package cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.client; + +import cn.iocoder.yudao.module.iot.core.biz.dto.IotModbusPointRespDTO; +import cn.iocoder.yudao.module.iot.core.enums.IotModbusFunctionCodeEnum; +import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.manager.IotModbusTcpConnectionManager; +import com.ghgande.j2mod.modbus.io.ModbusTCPTransaction; +import com.ghgande.j2mod.modbus.msg.*; +import com.ghgande.j2mod.modbus.procimg.InputRegister; +import com.ghgande.j2mod.modbus.procimg.Register; +import com.ghgande.j2mod.modbus.procimg.SimpleRegister; +import com.ghgande.j2mod.modbus.util.BitVector; +import io.vertx.core.Future; +import lombok.extern.slf4j.Slf4j; + +/** + * IoT Modbus TCP 客户端 + *

+ * 封装 Modbus 协议读写操作： + * 1. 封装 Modbus 读/写操作 + * 2. 根据功能码，执行对应的 Modbus 请求 + * + * @author 芋道源码 + */ +@Slf4j +public class IotModbusTcpClient { + + /** + * 读取 Modbus 数据 + * + * @param connection Modbus 连接 + * @param slaveId 从站地址 + * @param point 点位配置 + * @return 原始值（int 数组） + */ + public Future read(IotModbusTcpConnectionManager.ModbusConnection connection, + Integer slaveId, + IotModbusPointRespDTO point) { + return connection.executeBlocking(tcpConnection -> { + try { + // 1. 创建请求 + ModbusRequest request = createReadRequest(point.getFunctionCode(), + point.getRegisterAddress(), point.getRegisterCount()); + request.setUnitID(slaveId); + + // 2. 执行事务（请求） + ModbusTCPTransaction transaction = new ModbusTCPTransaction(tcpConnection); + transaction.setRequest(request); + transaction.execute(); + + // 3. 解析响应 + ModbusResponse response = transaction.getResponse(); + return extractValues(response, point.getFunctionCode()); + } catch (Exception e) { + throw new RuntimeException(String.format("Modbus 读取失败 [slaveId=%d, identifier=%s, address=%d]", + slaveId, point.getIdentifier(), point.getRegisterAddress()), e); + } + }); + } + + /** + * 写入 Modbus 数据 + * + * @param connection Modbus 连接 + * @param slaveId 从站地址 + * @param point 点位配置 + * @param values 要写入的值 + * @return 是否成功 + */ + public Future write(IotModbusTcpConnectionManager.ModbusConnection connection, + Integer slaveId, + IotModbusPointRespDTO point, + int[] values) { + return connection.executeBlocking(tcpConnection -> { + try { + // 1. 创建请求 + ModbusRequest request = createWriteRequest(point.getFunctionCode(), + point.getRegisterAddress(), point.getRegisterCount(), values); + if (request == null) { + throw new RuntimeException("功能码 " + point.getFunctionCode() + " 不支持写操作"); + } + request.setUnitID(slaveId); + + // 2. 执行事务（请求） + ModbusTCPTransaction transaction = new ModbusTCPTransaction(tcpConnection); + transaction.setRequest(request); + transaction.execute(); + return true; + } catch (Exception e) { + throw new RuntimeException(String.format("Modbus 写入失败 [slaveId=%d, identifier=%s, address=%d]", + slaveId, point.getIdentifier(), point.getRegisterAddress()), e); + } + }); + } + + /** + * 创建读取请求 + */ + @SuppressWarnings("EnhancedSwitchMigration") + private ModbusRequest createReadRequest(Integer functionCode, Integer address, Integer count) { + IotModbusFunctionCodeEnum functionCodeEnum = IotModbusFunctionCodeEnum.valueOf(functionCode); + if (functionCodeEnum == null) { + throw new IllegalArgumentException("不支持的功能码: " + functionCode); + } + switch (functionCodeEnum) { + case READ_COILS: + return new ReadCoilsRequest(address, count); + case READ_DISCRETE_INPUTS: + return new ReadInputDiscretesRequest(address, count); + case READ_HOLDING_REGISTERS: + return new ReadMultipleRegistersRequest(address, count); + case READ_INPUT_REGISTERS: + return new ReadInputRegistersRequest(address, count); + default: + throw new IllegalArgumentException("不支持的功能码: " + functionCode); + } + } + + /** + * 创建写入请求 + */ + private ModbusRequest createWriteRequest(Integer functionCode, Integer address, Integer count, int[] values) { + IotModbusFunctionCodeEnum functionCodeEnum = IotModbusFunctionCodeEnum.valueOf(functionCode); + if (functionCodeEnum == null) { + throw new IllegalArgumentException("不支持的功能码: " + functionCode); + } + switch (functionCodeEnum) { + case READ_COILS: // 写线圈（使用功能码 5 或 15） + if (count == 1) { + return new WriteCoilRequest(address, values[0] != 0); + } else { + BitVector bv = new BitVector(count); + for (int i = 0; i < Math.min(values.length, count); i++) { + bv.setBit(i, values[i] != 0); + } + return new WriteMultipleCoilsRequest(address, bv); + } + case READ_HOLDING_REGISTERS: // 写保持寄存器（使用功能码 6 或 16） + if (count == 1) { + return new WriteSingleRegisterRequest(address, new SimpleRegister(values[0])); + } else { + Register[] registers = new SimpleRegister[count]; + for (int i = 0; i < count; i++) { + registers[i] = new SimpleRegister(i < values.length ? values[i] : 0); + } + return new WriteMultipleRegistersRequest(address, registers); + } + case READ_DISCRETE_INPUTS: // 只读 + case READ_INPUT_REGISTERS: // 只读 + return null; + default: + throw new IllegalArgumentException("不支持的功能码: " + functionCode); + } + } + + /** + * 从响应中提取值 + */ + private int[] extractValues(ModbusResponse response, Integer functionCode) { + IotModbusFunctionCodeEnum functionCodeEnum = IotModbusFunctionCodeEnum.valueOf(functionCode); + if (functionCodeEnum == null) { + throw new IllegalArgumentException("不支持的功能码: " + functionCode); + } + switch (functionCodeEnum) { + case READ_COILS: + ReadCoilsResponse coilsResponse = (ReadCoilsResponse) response; + int bitCount = coilsResponse.getBitCount(); + int[] coilValues = new int[bitCount]; + for (int i = 0; i < bitCount; i++) { + coilValues[i] = coilsResponse.getCoilStatus(i) ? 1 : 0; + } + return coilValues; + case READ_DISCRETE_INPUTS: + ReadInputDiscretesResponse discretesResponse = (ReadInputDiscretesResponse) response; + int discreteCount = discretesResponse.getBitCount(); + int[] discreteValues = new int[discreteCount]; + for (int i = 0; i < discreteCount; i++) { + discreteValues[i] = discretesResponse.getDiscreteStatus(i) ? 1 : 0; + } + return discreteValues; + case READ_HOLDING_REGISTERS: + ReadMultipleRegistersResponse holdingResponse = (ReadMultipleRegistersResponse) response; + InputRegister[] holdingRegisters = holdingResponse.getRegisters(); + int[] holdingValues = new int[holdingRegisters.length]; + for (int i = 0; i < holdingRegisters.length; i++) { + holdingValues[i] = holdingRegisters[i].getValue(); + } + return holdingValues; + case READ_INPUT_REGISTERS: + ReadInputRegistersResponse inputResponse = (ReadInputRegistersResponse) response; + InputRegister[] inputRegisters = inputResponse.getRegisters(); + int[] inputValues = new int[inputRegisters.length]; + for (int i = 0; i < inputRegisters.length; i++) { + inputValues[i] = inputRegisters[i].getValue(); + } + return inputValues; + default: + throw new IllegalArgumentException("不支持的功能码: " + functionCode); + } + } + +} diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/codec/IotModbusDataConverter.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/codec/IotModbusDataConverter.java new file mode 100644 index 0000000..61a4201 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/codec/IotModbusDataConverter.java @@ -0,0 +1,287 @@ +package cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.codec; + +import cn.hutool.core.util.ArrayUtil; +import cn.hutool.core.util.ObjectUtil; +import cn.iocoder.yudao.framework.common.util.object.ObjectUtils; +import cn.iocoder.yudao.module.iot.core.biz.dto.IotModbusPointRespDTO; +import cn.iocoder.yudao.module.iot.core.enums.IotModbusByteOrderEnum; +import cn.iocoder.yudao.module.iot.core.enums.IotModbusRawDataTypeEnum; +import lombok.extern.slf4j.Slf4j; + +import java.math.BigDecimal; +import java.math.RoundingMode; +import java.nio.ByteBuffer; +import java.nio.ByteOrder; + +/** + * IoT Modbus 数据转换器 + *

+ * 负责 Modbus 原始寄存器值与物模型属性值的相互转换： + * 1. 将 Modbus 原始寄存器值转换为物模型属性值 + * 2. 将物模型属性值转换为 Modbus 原始寄存器值 + * + * @author 芋道源码 + */ +@Slf4j +public class IotModbusDataConverter { + + /** + * 将原始值转换为物模型属性值 + * + * @param rawValues 原始值数组（寄存器值或线圈值） + * @param point 点位配置 + * @return 转换后的属性值 + */ + public Object convertToPropertyValue(int[] rawValues, IotModbusPointRespDTO point) { + if (ArrayUtil.isEmpty(rawValues)) { + return null; + } + String rawDataType = point.getRawDataType(); + String byteOrder = point.getByteOrder(); + BigDecimal scale = ObjectUtil.defaultIfNull(point.getScale(), BigDecimal.ONE); + + // 1. 根据原始数据类型解析原始数值 + Number rawNumber = parseRawValue(rawValues, rawDataType, byteOrder); + if (rawNumber == null) { + return null; + } + + // 2. 应用缩放因子：实际值 = 原始值 × scale + BigDecimal actualValue = new BigDecimal(rawNumber.toString()).multiply(scale); + + // 3. 根据数据类型返回合适的 Java 类型 + return formatValue(actualValue, rawDataType); + } + + /** + * 将物模型属性值转换为原始寄存器值 + * + * @param propertyValue 属性值 + * @param point 点位配置 + * @return 原始值数组 + */ + public int[] convertToRawValues(Object propertyValue, IotModbusPointRespDTO point) { + if (propertyValue == null) { + return new int[0]; + } + String rawDataType = point.getRawDataType(); + String byteOrder = point.getByteOrder(); + BigDecimal scale = ObjectUtil.defaultIfNull(point.getScale(), BigDecimal.ONE); + int registerCount = ObjectUtil.defaultIfNull(point.getRegisterCount(), 1); + + // 1. 转换为 BigDecimal + BigDecimal actualValue = new BigDecimal(propertyValue.toString()); + + // 2. 应用缩放因子：原始值 = 实际值 ÷ scale + BigDecimal rawValue = actualValue.divide(scale, 0, RoundingMode.HALF_UP); + + // 3. 根据原始数据类型编码为寄存器值 + return encodeToRegisters(rawValue, rawDataType, byteOrder, registerCount); + } + + /** + * 解析原始值 + */ + @SuppressWarnings("EnhancedSwitchMigration") + private Number parseRawValue(int[] rawValues, String rawDataType, String byteOrder) { + IotModbusRawDataTypeEnum dataTypeEnum = IotModbusRawDataTypeEnum.getByType(rawDataType); + if (dataTypeEnum == null) { + log.warn("[parseRawValue][不支持的数据类型: {}]", rawDataType); + return rawValues[0]; + } + switch (dataTypeEnum) { + case BOOLEAN: + return rawValues[0] != 0 ? 1 : 0; + case INT16: + return (short) rawValues[0]; + case UINT16: + return rawValues[0] & 0xFFFF; + case INT32: + return parseInt32(rawValues, byteOrder); + case UINT32: + return parseUint32(rawValues, byteOrder); + case FLOAT: + return parseFloat(rawValues, byteOrder); + case DOUBLE: + return parseDouble(rawValues, byteOrder); + default: + log.warn("[parseRawValue][不支持的数据类型: {}]", rawDataType); + return rawValues[0]; + } + } + + private int parseInt32(int[] rawValues, String byteOrder) { + if (rawValues.length < 2) { + return rawValues[0]; + } + byte[] bytes = reorderBytes(registersToBytes(rawValues, 2), byteOrder); + return ByteBuffer.wrap(bytes).order(ByteOrder.BIG_ENDIAN).getInt(); + } + + private long parseUint32(int[] rawValues, String byteOrder) { + if (rawValues.length < 2) { + return rawValues[0] & 0xFFFFFFFFL; + } + byte[] bytes = reorderBytes(registersToBytes(rawValues, 2), byteOrder); + return ByteBuffer.wrap(bytes).order(ByteOrder.BIG_ENDIAN).getInt() & 0xFFFFFFFFL; + } + + private float parseFloat(int[] rawValues, String byteOrder) { + if (rawValues.length < 2) { + return (float) rawValues[0]; + } + byte[] bytes = reorderBytes(registersToBytes(rawValues, 2), byteOrder); + return ByteBuffer.wrap(bytes).order(ByteOrder.BIG_ENDIAN).getFloat(); + } + + private double parseDouble(int[] rawValues, String byteOrder) { + if (rawValues.length < 4) { + return rawValues[0]; + } + byte[] bytes = reorderBytes(registersToBytes(rawValues, 4), byteOrder); + return ByteBuffer.wrap(bytes).order(ByteOrder.BIG_ENDIAN).getDouble(); + } + + /** + * 将寄存器值转换为字节数组 + */ + private byte[] registersToBytes(int[] registers, int count) { + byte[] bytes = new byte[count * 2]; + for (int i = 0; i < Math.min(registers.length, count); i++) { + bytes[i * 2] = (byte) ((registers[i] >> 8) & 0xFF); + bytes[i * 2 + 1] = (byte) (registers[i] & 0xFF); + } + return bytes; + } + + /** + * 根据字节序重排字节 + */ + @SuppressWarnings("EnhancedSwitchMigration") + private byte[] reorderBytes(byte[] bytes, String byteOrder) { + IotModbusByteOrderEnum byteOrderEnum = IotModbusByteOrderEnum.getByOrder(byteOrder); + // null 或者大端序，不需要调整 + if (ObjectUtils.equalsAny(byteOrderEnum, null, IotModbusByteOrderEnum.ABCD, IotModbusByteOrderEnum.AB)) { + return bytes; + } + + // 其他字节序调整 + byte[] result = new byte[bytes.length]; + switch (byteOrderEnum) { + case BA: // 小端序（16 位） + if (bytes.length >= 2) { + result[0] = bytes[1]; + result[1] = bytes[0]; + } + break; + case CDAB: // 大端字交换（32 位） + if (bytes.length >= 4) { + result[0] = bytes[2]; + result[1] = bytes[3]; + result[2] = bytes[0]; + result[3] = bytes[1]; + } + break; + case DCBA: // 小端序（32 位） + if (bytes.length >= 4) { + result[0] = bytes[3]; + result[1] = bytes[2]; + result[2] = bytes[1]; + result[3] = bytes[0]; + } + break; + case BADC: // 小端字交换（32 位） + if (bytes.length >= 4) { + result[0] = bytes[1]; + result[1] = bytes[0]; + result[2] = bytes[3]; + result[3] = bytes[2]; + } + break; + default: + return bytes; + } + return result; + } + + /** + * 编码为寄存器值 + */ + @SuppressWarnings("EnhancedSwitchMigration") + private int[] encodeToRegisters(BigDecimal rawValue, String rawDataType, String byteOrder, int registerCount) { + IotModbusRawDataTypeEnum dataTypeEnum = IotModbusRawDataTypeEnum.getByType(rawDataType); + if (dataTypeEnum == null) { + return new int[]{rawValue.intValue()}; + } + switch (dataTypeEnum) { + case BOOLEAN: + return new int[]{rawValue.intValue() != 0 ? 1 : 0}; + case INT16: + case UINT16: + return new int[]{rawValue.intValue() & 0xFFFF}; + case INT32: + case UINT32: + return encodeInt32(rawValue.intValue(), byteOrder); + case FLOAT: + return encodeFloat(rawValue.floatValue(), byteOrder); + case DOUBLE: + return encodeDouble(rawValue.doubleValue(), byteOrder); + default: + return new int[]{rawValue.intValue()}; + } + } + + private int[] encodeInt32(int value, String byteOrder) { + byte[] bytes = ByteBuffer.allocate(4).order(ByteOrder.BIG_ENDIAN).putInt(value).array(); + bytes = reorderBytes(bytes, byteOrder); + return bytesToRegisters(bytes); + } + + private int[] encodeFloat(float value, String byteOrder) { + byte[] bytes = ByteBuffer.allocate(4).order(ByteOrder.BIG_ENDIAN).putFloat(value).array(); + bytes = reorderBytes(bytes, byteOrder); + return bytesToRegisters(bytes); + } + + private int[] encodeDouble(double value, String byteOrder) { + byte[] bytes = ByteBuffer.allocate(8).order(ByteOrder.BIG_ENDIAN).putDouble(value).array(); + bytes = reorderBytes(bytes, byteOrder); + return bytesToRegisters(bytes); + } + + private int[] bytesToRegisters(byte[] bytes) { + int[] registers = new int[bytes.length / 2]; + for (int i = 0; i < registers.length; i++) { + registers[i] = ((bytes[i * 2] & 0xFF) << 8) | (bytes[i * 2 + 1] & 0xFF); + } + return registers; + } + + /** + * 格式化返回值 + */ + @SuppressWarnings("EnhancedSwitchMigration") + private Object formatValue(BigDecimal value, String rawDataType) { + IotModbusRawDataTypeEnum dataTypeEnum = IotModbusRawDataTypeEnum.getByType(rawDataType); + if (dataTypeEnum == null) { + return value; + } + switch (dataTypeEnum) { + case BOOLEAN: + return value.intValue() != 0; + case INT16: + case INT32: + return value.intValue(); + case UINT16: + case UINT32: + return value.longValue(); + case FLOAT: + return value.floatValue(); + case DOUBLE: + return value.doubleValue(); + default: + return value; + } + } + +} diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/manager/IotModbusTcpConfigCacheService.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/manager/IotModbusTcpConfigCacheService.java new file mode 100644 index 0000000..8712799 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/manager/IotModbusTcpConfigCacheService.java @@ -0,0 +1,99 @@ +package cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.manager; + +import cn.iocoder.yudao.framework.common.pojo.CommonResult; +import cn.iocoder.yudao.module.iot.core.biz.IotDeviceCommonApi; +import cn.iocoder.yudao.module.iot.core.biz.dto.IotModbusDeviceConfigRespDTO; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; + +import java.util.*; +import java.util.concurrent.ConcurrentHashMap; +import java.util.function.Consumer; + +import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertSet; + +/** + * IoT Modbus TCP 配置缓存服务，负责：从 biz 拉取 Modbus 设备配置，缓存配置数据，并检测配置变更 + * + * @author 芋道源码 + */ +@RequiredArgsConstructor +@Slf4j +public class IotModbusTcpConfigCacheService { + + private final IotDeviceCommonApi deviceApi; + + /** + * 配置缓存：deviceId -> 配置 + */ + private final Map configCache = new ConcurrentHashMap<>(); + + /** + * 已知的设备 ID 集合（作用：用于检测已删除的设备） + * + * @see #cleanupRemovedDevices(List, Consumer) + */ + private final Set knownDeviceIds = ConcurrentHashMap.newKeySet(); + + /** + * 刷新配置 + * + * @return 最新的配置列表 + */ + public List refreshConfig() { + try { + // 1. 从远程获取配置 + CommonResult> result = deviceApi.getEnabledModbusDeviceConfigs(); + if (result == null || !result.isSuccess() || result.getData() == null) { + log.warn("[refreshConfig][获取 Modbus 配置失败: {}]", result); + return new ArrayList<>(configCache.values()); + } + List configs = result.getData(); + + // 2. 更新缓存（注意：不在这里更新 knownDeviceIds，由 cleanupRemovedDevices 统一管理） + for (IotModbusDeviceConfigRespDTO config : configs) { + configCache.put(config.getDeviceId(), config); + } + return configs; + } catch (Exception e) { + log.error("[refreshConfig][刷新配置失败]", e); + return new ArrayList<>(configCache.values()); + } + } + + /** + * 获取设备配置 + * + * @param deviceId 设备 ID + * @return 配置 + */ + public IotModbusDeviceConfigRespDTO getConfig(Long deviceId) { + return configCache.get(deviceId); + } + + /** + * 清理已删除设备的资源，并更新已知设备 ID 集合 + * + * @param currentConfigs 当前有效的配置列表 + * @param cleanupAction 清理动作 + */ + public void cleanupRemovedDevices(List currentConfigs, Consumer cleanupAction) { + // 1.1 获取当前有效的设备 ID + Set currentDeviceIds = convertSet(currentConfigs, IotModbusDeviceConfigRespDTO::getDeviceId); + // 1.2 找出已删除的设备（基于旧的 knownDeviceIds） + Set removedDeviceIds = new HashSet<>(knownDeviceIds); + removedDeviceIds.removeAll(currentDeviceIds); + + // 2. 清理已删除设备 + for (Long deviceId : removedDeviceIds) { + log.info("[cleanupRemovedDevices][清理已删除设备: {}]", deviceId); + configCache.remove(deviceId); + cleanupAction.accept(deviceId); + } + + // 3. 更新已知设备 ID 集合为当前有效的设备 ID + knownDeviceIds.clear(); + knownDeviceIds.addAll(currentDeviceIds); + } + +} diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/manager/IotModbusTcpConnectionManager.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/manager/IotModbusTcpConnectionManager.java new file mode 100644 index 0000000..8b0351e --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/manager/IotModbusTcpConnectionManager.java @@ -0,0 +1,225 @@ +package cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.manager; + +import cn.iocoder.yudao.module.iot.core.biz.dto.IotModbusDeviceConfigRespDTO; +import com.ghgande.j2mod.modbus.net.TCPMasterConnection; +import io.vertx.core.Context; +import io.vertx.core.Future; +import io.vertx.core.Vertx; +import lombok.Data; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.redisson.api.RLock; +import org.redisson.api.RedissonClient; + +import java.net.InetAddress; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +/** + * IoT Modbus TCP 连接管理器 + *

+ * 统一管理 Modbus TCP 连接： + * 1. 管理 TCP 连接（相同 ip:port 共用连接） + * 2. 分布式锁管理（连接级别），避免多节点重复创建连接 + * 3. 连接重试和故障恢复 + * + * @author 芋道源码 + */ +@RequiredArgsConstructor +@Slf4j +public class IotModbusTcpConnectionManager { + + private static final String LOCK_KEY_PREFIX = "iot:modbus-tcp:connection:"; + + private final RedissonClient redissonClient; + private final Vertx vertx; + + /** + * 连接池：key = ip:port + */ + private final Map connectionPool = new ConcurrentHashMap<>(); + + /** + * 设备 ID 到连接 key 的映射 + */ + private final Map deviceConnectionMap = new ConcurrentHashMap<>(); + + /** + * 确保连接存在 + * + * @param config 设备配置 + */ + public void ensureConnection(IotModbusDeviceConfigRespDTO config) { + // 1. 记录设备与连接的关系 + String connectionKey = buildConnectionKey(config.getIp(), config.getPort()); + deviceConnectionMap.put(config.getDeviceId(), connectionKey); + + // 2. 情况一：连接已存在，添加设备引用 + ModbusConnection connection = connectionPool.get(connectionKey); + if (connection != null) { + connection.addDevice(config.getDeviceId(), config.getSlaveId()); + return; + } + + // 3. 情况二：连接不存在，创建新连接 + // 3.1 尝试获取分布式锁 + RLock lock = redissonClient.getLock(LOCK_KEY_PREFIX + connectionKey); + if (!lock.tryLock()) { + log.debug("[ensureConnection][获取锁失败, 由其他节点负责: {}]", connectionKey); + return; + } + // 3.2 创建新连接 + try { + connection = createConnection(config, lock); + connectionPool.put(connectionKey, connection); + log.info("[ensureConnection][创建 Modbus 连接成功: {}]", connectionKey); + } catch (Exception e) { + log.error("[ensureConnection][创建 Modbus 连接失败: {}]", connectionKey, e); + lock.unlock(); + } + } + + /** + * 创建 Modbus TCP 连接 + */ + private ModbusConnection createConnection(IotModbusDeviceConfigRespDTO config, RLock lock) throws Exception { + // 1. 创建 TCP 连接 + TCPMasterConnection tcpConnection = new TCPMasterConnection(InetAddress.getByName(config.getIp())); + tcpConnection.setPort(config.getPort()); + tcpConnection.setTimeout(config.getTimeout()); + tcpConnection.connect(); + + // 2. 创建 Modbus 连接对象 + ModbusConnection connection = new ModbusConnection() + .setConnectionKey(buildConnectionKey(config.getIp(), config.getPort())) + .setTcpConnection(tcpConnection).setLock(lock).setContext(vertx.getOrCreateContext()) + .setTimeout(config.getTimeout()).setRetryInterval(config.getRetryInterval()); + connection.addDevice(config.getDeviceId(), config.getSlaveId()); + return connection; + } + + /** + * 获取连接 + */ + public ModbusConnection getConnection(Long deviceId) { + String connectionKey = deviceConnectionMap.get(deviceId); + if (connectionKey == null) { + return null; + } + return connectionPool.get(connectionKey); + } + + /** + * 获取设备的 slave ID + */ + public Integer getSlaveId(Long deviceId) { + ModbusConnection connection = getConnection(deviceId); + if (connection == null) { + return null; + } + return connection.getSlaveId(deviceId); + } + + /** + * 移除设备 + */ + public void removeDevice(Long deviceId) { + // 1. 移除设备引用 + String connectionKey = deviceConnectionMap.remove(deviceId); + if (connectionKey == null) { + return; + } + ModbusConnection connection = connectionPool.get(connectionKey); + if (connection == null) { + return; + } + connection.removeDevice(deviceId); + + // 2. 如果没有设备引用了，关闭连接 + if (connection.getDeviceCount() == 0) { + closeConnection(connectionKey); + } + } + + /** + * 关闭指定连接 + */ + private void closeConnection(String connectionKey) { + ModbusConnection connection = connectionPool.remove(connectionKey); + if (connection == null) { + return; + } + + try { + if (connection.getTcpConnection() != null) { + connection.getTcpConnection().close(); + } + // 安全释放锁：先检查锁存在且被锁定，再检查是否当前线程持有 + RLock lock = connection.getLock(); + if (lock != null && lock.isLocked() && lock.isHeldByCurrentThread()) { + lock.unlock(); + } + log.info("[closeConnection][关闭 Modbus 连接: {}]", connectionKey); + } catch (Exception e) { + log.error("[closeConnection][关闭连接失败: {}]", connectionKey, e); + } + } + + /** + * 关闭所有连接 + */ + public void closeAll() { + for (String connectionKey : connectionPool.keySet()) { + closeConnection(connectionKey); + } + deviceConnectionMap.clear(); + } + + private String buildConnectionKey(String ip, Integer port) { + return ip + ":" + port; + } + + /** + * Modbus 连接信息 + */ + @Data + public static class ModbusConnection { + + private String connectionKey; + private TCPMasterConnection tcpConnection; + private RLock lock; + private Integer timeout; + private Integer retryInterval; + private Context context; + + /** + * 设备 ID 到 slave ID 的映射 + */ + private final Map deviceSlaveMap = new ConcurrentHashMap<>(); + + public void addDevice(Long deviceId, Integer slaveId) { + deviceSlaveMap.put(deviceId, slaveId); + } + + public void removeDevice(Long deviceId) { + deviceSlaveMap.remove(deviceId); + } + + public int getDeviceCount() { + return deviceSlaveMap.size(); + } + + public Integer getSlaveId(Long deviceId) { + return deviceSlaveMap.get(deviceId); + } + + /** + * 执行 Modbus 读取操作（阻塞方式，在 Vert.x worker 线程执行） + */ + public Future executeBlocking(java.util.function.Function operation) { + // ordered=true 保证同一 Context 的操作串行执行，不同连接之间可并行 + return context.executeBlocking(() -> operation.apply(tcpConnection), true); + } + } + +} diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/manager/IotModbusTcpPollScheduler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/manager/IotModbusTcpPollScheduler.java new file mode 100644 index 0000000..3caf030 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/manager/IotModbusTcpPollScheduler.java @@ -0,0 +1,180 @@ +package cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.manager; + +import cn.hutool.core.collection.CollUtil; +import cn.iocoder.yudao.module.iot.core.biz.dto.IotModbusDeviceConfigRespDTO; +import cn.iocoder.yudao.module.iot.core.biz.dto.IotModbusPointRespDTO; +import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.router.IotModbusTcpUpstreamHandler; +import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.client.IotModbusTcpClient; +import io.vertx.core.Vertx; +import lombok.AllArgsConstructor; +import lombok.Data; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; + +import java.util.*; +import java.util.concurrent.ConcurrentHashMap; + +import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertSet; + +/** + * IoT Modbus TCP 轮询调度器：管理点位的轮询定时器，调度读取任务并上报结果 + * + * @author 芋道源码 + */ +@RequiredArgsConstructor +@Slf4j +public class IotModbusTcpPollScheduler { + + private final Vertx vertx; + private final IotModbusTcpConnectionManager connectionManager; + private final IotModbusTcpClient modbusClient; + private final IotModbusTcpUpstreamHandler upstreamHandler; + + /** + * 设备点位的定时器映射：deviceId -> (pointId -> PointTimerInfo) + */ + private final Map> devicePointTimers = new ConcurrentHashMap<>(); + + /** + * 点位定时器信息 + */ + @Data + @AllArgsConstructor + private static class PointTimerInfo { + + /** + * Vert.x 定时器 ID + */ + private Long timerId; + /** + * 轮询间隔（用于判断是否需要更新定时器） + */ + private Integer pollInterval; + + } + + /** + * 更新轮询任务（增量更新） + * + * 1. 【删除】点位：停止对应的轮询定时器 + * 2. 【新增】点位：创建对应的轮询定时器 + * 3. 【修改】点位：pollInterval 变化，重建对应的轮询定时器 + * 4. 其他属性变化（包括未变化的）：不处理（下次轮询时自动使用新配置） + */ + public void updatePolling(IotModbusDeviceConfigRespDTO config) { + Long deviceId = config.getDeviceId(); + List newPoints = config.getPoints(); + Map currentTimers = devicePointTimers + .computeIfAbsent(deviceId, k -> new ConcurrentHashMap<>()); + // 1.1 计算新配置（包括新增和修改的点位）中的点位 ID 集合 + Set newPointIds = convertSet(newPoints, IotModbusPointRespDTO::getId); + // 1.2 计算删除的点位 ID 集合 + Set removedPointIds = new HashSet<>(currentTimers.keySet()); + removedPointIds.removeAll(newPointIds); + + // 2. 处理删除的点位：停止不再存在的定时器 + for (Long pointId : removedPointIds) { + PointTimerInfo timerInfo = currentTimers.remove(pointId); + if (timerInfo != null) { + vertx.cancelTimer(timerInfo.getTimerId()); + log.debug("[updatePolling][设备 {} 点位 {} 定时器已删除]", deviceId, pointId); + } + } + + // 3. 处理新增和修改的点位 + if (CollUtil.isEmpty(newPoints)) { + return; + } + for (IotModbusPointRespDTO point : newPoints) { + Long pointId = point.getId(); + Integer newPollInterval = point.getPollInterval(); + PointTimerInfo existingTimer = currentTimers.get(pointId); + // 3.1 新增点位：创建定时器 + if (existingTimer == null) { + Long timerId = createPollTimer(config, point); + if (timerId != null) { + currentTimers.put(pointId, new PointTimerInfo(timerId, newPollInterval)); + log.debug("[updatePolling][设备 {} 点位 {} 定时器已创建, interval={}ms]", + deviceId, pointId, newPollInterval); + } + } else if (!Objects.equals(existingTimer.getPollInterval(), newPollInterval)) { + // 3.2 pollInterval 变化：重建定时器 + vertx.cancelTimer(existingTimer.getTimerId()); + Long timerId = createPollTimer(config, point); + if (timerId != null) { + currentTimers.put(pointId, new PointTimerInfo(timerId, newPollInterval)); + log.debug("[updatePolling][设备 {} 点位 {} 定时器已更新, interval={}ms -> {}ms]", + deviceId, pointId, existingTimer.getPollInterval(), newPollInterval); + } else { + currentTimers.remove(pointId); + } + } + // 3.3 其他属性变化：不处理（下次轮询时自动使用新配置） + } + } + + /** + * 创建轮询定时器 + */ + private Long createPollTimer(IotModbusDeviceConfigRespDTO config, IotModbusPointRespDTO point) { + if (point.getPollInterval() == null || point.getPollInterval() <= 0) { + return null; + } + return vertx.setPeriodic(point.getPollInterval(), timerId -> { + try { + pollPoint(config, point); + } catch (Exception e) { + log.error("[createPollTimer][轮询点位失败, deviceId={}, identifier={}]", + config.getDeviceId(), point.getIdentifier(), e); + } + }); + } + + /** + * 轮询单个点位 + */ + private void pollPoint(IotModbusDeviceConfigRespDTO config, IotModbusPointRespDTO point) { + // 1.1 获取连接 + IotModbusTcpConnectionManager.ModbusConnection connection = connectionManager.getConnection(config.getDeviceId()); + if (connection == null) { + log.warn("[pollPoint][设备 {} 没有连接]", config.getDeviceId()); + return; + } + // 1.2 获取 slave ID + Integer slaveId = connectionManager.getSlaveId(config.getDeviceId()); + if (slaveId == null) { + log.warn("[pollPoint][设备 {} 没有 slaveId]", config.getDeviceId()); + return; + } + + // 2. 执行 Modbus 读取 + modbusClient.read(connection, slaveId, point) + .onSuccess(rawValue -> upstreamHandler.handleReadResult(config, point, rawValue)) + .onFailure(e -> log.error("[pollPoint][读取点位失败, deviceId={}, identifier={}]", + config.getDeviceId(), point.getIdentifier(), e)); + } + + /** + * 停止设备的轮询 + */ + public void stopPolling(Long deviceId) { + Map timers = devicePointTimers.remove(deviceId); + if (CollUtil.isEmpty(timers)) { + return; + } + for (PointTimerInfo timerInfo : timers.values()) { + vertx.cancelTimer(timerInfo.getTimerId()); + } + log.debug("[stopPolling][设备 {} 停止了 {} 个轮询定时器]", deviceId, timers.size()); + } + + /** + * 停止所有轮询 + */ + public void stopAll() { + for (Long deviceId : new ArrayList<>(devicePointTimers.keySet())) { + stopPolling(deviceId); + } + } + +} diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/package-info.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/package-info.java new file mode 100644 index 0000000..823d3f7 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/package-info.java @@ -0,0 +1,9 @@ +/** + * Modbus TCP 协议实现包 + *

+ * 提供基于 j2mod 的 Modbus TCP 主站（Master）功能，支持： + * 1. 定时轮询 Modbus 从站设备数据 + * 2. 下发属性设置命令到从站设备 + * 3. 数据格式转换（寄存器值 ↔ 物模型属性值） + */ +package cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp; \ No newline at end of file diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/router/IotModbusTcpDownstreamHandler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/router/IotModbusTcpDownstreamHandler.java new file mode 100644 index 0000000..2355a3d --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/router/IotModbusTcpDownstreamHandler.java @@ -0,0 +1,122 @@ +package cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.router; + +import cn.hutool.core.collection.CollUtil; +import cn.iocoder.yudao.module.iot.core.biz.dto.IotModbusDeviceConfigRespDTO; +import cn.iocoder.yudao.module.iot.core.biz.dto.IotModbusPointRespDTO; +import cn.iocoder.yudao.module.iot.core.enums.IotModbusFunctionCodeEnum; +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; +import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.codec.IotModbusDataConverter; +import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.client.IotModbusTcpClient; +import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.manager.IotModbusTcpConfigCacheService; +import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.manager.IotModbusTcpConnectionManager; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; + +import java.util.Map; + +/** + * IoT Modbus TCP 下行消息处理器 + * + * 负责： + * 1. 处理下行消息（如属性设置 thing.service.property.set） + * 2. 执行 Modbus 写入操作 + * + * @author 芋道源码 + */ +@RequiredArgsConstructor +@Slf4j +public class IotModbusTcpDownstreamHandler { + + private final IotModbusTcpConnectionManager connectionManager; + private final IotModbusTcpClient modbusClient; + private final IotModbusDataConverter dataConverter; + private final IotModbusTcpConfigCacheService configCacheService; + + /** + * 处理下行消息 + */ + @SuppressWarnings("unchecked") + public void handle(IotDeviceMessage message) { + // 1.1 检查是否是属性设置消息 + // TODO @AI：要使用枚举 + if (!"thing.service.property.set".equals(message.getMethod())) { + log.debug("[handle][忽略非属性设置消息: {}]", message.getMethod()); + return; + } + // 1.2 获取设备配置 + IotModbusDeviceConfigRespDTO config = configCacheService.getConfig(message.getDeviceId()); + if (config == null) { + log.warn("[handle][设备 {} 没有 Modbus 配置]", message.getDeviceId()); + return; + } + + // 2. 解析属性值并写入 + Object params = message.getParams(); + if (!(params instanceof Map)) { + log.warn("[handle][params 不是 Map 类型: {}]", params); + return; + } + Map propertyMap = (Map) params; + for (Map.Entry entry : propertyMap.entrySet()) { + String identifier = entry.getKey(); + Object value = entry.getValue(); + // 2.1.1 查找对应的点位配置 + IotModbusPointRespDTO point = findPoint(config, identifier); + if (point == null) { + log.warn("[handle][设备 {} 没有点位配置: {}]", message.getDeviceId(), identifier); + continue; + } + // 2.1.2 检查是否支持写操作 + if (!isWritable(point.getFunctionCode())) { + log.warn("[handle][点位 {} 不支持写操作, 功能码={}]", identifier, point.getFunctionCode()); + continue; + } + + // 2.2 执行写入 + writeProperty(config, point, value); + } + } + + /** + * 写入属性值 + */ + private void writeProperty(IotModbusDeviceConfigRespDTO config, IotModbusPointRespDTO point, Object value) { + // 1.1 获取连接 + IotModbusTcpConnectionManager.ModbusConnection connection = connectionManager.getConnection(config.getDeviceId()); + if (connection == null) { + log.warn("[writeProperty][设备 {} 没有连接]", config.getDeviceId()); + return; + } + // 1.2 获取 slave ID + Integer slaveId = connectionManager.getSlaveId(config.getDeviceId()); + if (slaveId == null) { + log.warn("[writeProperty][设备 {} 没有 slaveId]", config.getDeviceId()); + return; + } + + // 2.1 转换属性值为原始值 + int[] rawValues = dataConverter.convertToRawValues(value, point); + // 2.2 执行 Modbus 写入 + modbusClient.write(connection, slaveId, point, rawValues) + .onSuccess(success -> log.info("[writeProperty][写入成功, deviceId={}, identifier={}, value={}]", + config.getDeviceId(), point.getIdentifier(), value)) + .onFailure(e -> log.error("[writeProperty][写入失败, deviceId={}, identifier={}]", + config.getDeviceId(), point.getIdentifier(), e)); + } + + /** + * 查找点位配置 + */ + private IotModbusPointRespDTO findPoint(IotModbusDeviceConfigRespDTO config, String identifier) { + return CollUtil.findOne(config.getPoints(), p -> identifier.equals(p.getIdentifier())); + } + + /** + * 检查功能码是否支持写操作 + */ + private boolean isWritable(Integer functionCode) { + IotModbusFunctionCodeEnum functionCodeEnum = IotModbusFunctionCodeEnum.valueOf(functionCode); + return functionCodeEnum != null && Boolean.TRUE.equals(functionCodeEnum.getWritable()); + } + +} diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/router/IotModbusTcpUpstreamHandler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/router/IotModbusTcpUpstreamHandler.java new file mode 100644 index 0000000..5da669e --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/router/IotModbusTcpUpstreamHandler.java @@ -0,0 +1,63 @@ +package cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.router; + +import cn.hutool.core.map.MapUtil; +import cn.iocoder.yudao.module.iot.core.biz.dto.IotModbusDeviceConfigRespDTO; +import cn.iocoder.yudao.module.iot.core.biz.dto.IotModbusPointRespDTO; +import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum; +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; +import cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp.codec.IotModbusDataConverter; +import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService; +import lombok.Setter; +import lombok.extern.slf4j.Slf4j; + +import java.util.Map; + +/** + * IoT Modbus TCP 上行数据处理器：将原始值转换为物模型属性值并上报 + * + * @author 芋道源码 + */ +@Slf4j +public class IotModbusTcpUpstreamHandler { + + private final IotDeviceMessageService messageService; + private final IotModbusDataConverter dataConverter; + + @Setter + private String serverId; + + public IotModbusTcpUpstreamHandler(IotDeviceMessageService messageService, + IotModbusDataConverter dataConverter) { + this.messageService = messageService; + this.dataConverter = dataConverter; + } + + /** + * 处理 Modbus 读取结果 + * + * @param config 设备配置 + * @param point 点位配置 + * @param rawValue 原始值（int 数组） + */ + public void handleReadResult(IotModbusDeviceConfigRespDTO config, + IotModbusPointRespDTO point, + int[] rawValue) { + try { + // 1.1 转换原始值为物模型属性值 + Object convertedValue = dataConverter.convertToPropertyValue(rawValue, point); + log.debug("[handleReadResult][设备={}, 属性={}, 原始值={}, 转换值={}]", + config.getDeviceId(), point.getIdentifier(), rawValue, convertedValue); + // 1.2 构造属性上报消息 + Map params = MapUtil.of(point.getIdentifier(), convertedValue); + IotDeviceMessage message = IotDeviceMessage.requestOf(IotDeviceMessageMethodEnum.PROPERTY_POST.getMethod(), params); + + // 2. 发送到消息总线 + messageService.sendDeviceMessage(message, config.getProductKey(), + config.getDeviceName(), serverId); + } catch (Exception e) { + log.error("[handleReadResult][处理读取结果失败, deviceId={}, identifier={}]", + config.getDeviceId(), point.getIdentifier(), e); + } + } + +} diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/ModbusTcpSlaveSimulatorTest.java b/yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/ModbusTcpSlaveSimulatorTest.java new file mode 100644 index 0000000..509e600 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbustcp/ModbusTcpSlaveSimulatorTest.java @@ -0,0 +1,96 @@ +package cn.iocoder.yudao.module.iot.gateway.protocol.modbustcp; + +import com.ghgande.j2mod.modbus.procimg.*; +import com.ghgande.j2mod.modbus.slave.ModbusSlave; +import com.ghgande.j2mod.modbus.slave.ModbusSlaveFactory; + +/** + * Modbus TCP 从站模拟器 + * + * 用于测试 Modbus TCP 网关的连接和数据读写功能 + * + * @author 芋道源码 + */ +public class ModbusTcpSlaveSimulatorTest { + + private static final int PORT = 5020; + private static final int SLAVE_ID = 1; + + @SuppressWarnings({"InfiniteLoopStatement", "BusyWait", "CommentedOutCode"}) + public static void main(String[] args) throws Exception { + // 1. 创建进程映像（Process Image），存储寄存器数据 + SimpleProcessImage spi = new SimpleProcessImage(SLAVE_ID); + + // 2.1 初始化线圈（Coil，功能码 01/05）- 离散输出，可读写 + // 地址 0-9，共 10 个线圈 + for (int i = 0; i < 10; i++) { + spi.addDigitalOut(new SimpleDigitalOut(i % 2 == 0)); // 交替 true/false + } + + // 2.2 初始化离散输入（Discrete Input，功能码 02）- 只读 + // 地址 0-9，共 10 个离散输入 + for (int i = 0; i < 10; i++) { + spi.addDigitalIn(new SimpleDigitalIn(i % 3 == 0)); // 每 3 个一个 true + } + + // 2.3 初始化保持寄存器（Holding Register，功能码 03/06/16）- 可读写 + // 地址 0-19，共 20 个寄存器 + for (int i = 0; i < 20; i++) { + spi.addRegister(new SimpleRegister(i * 100)); // 值为 0, 100, 200, ... + } + + // 2.4 初始化输入寄存器（Input Register，功能码 04）- 只读 + // 地址 0-19，共 20 个寄存器 + SimpleInputRegister[] inputRegisters = new SimpleInputRegister[20]; + for (int i = 0; i < 20; i++) { + inputRegisters[i] = new SimpleInputRegister(i * 10 + 1); // 值为 1, 11, 21, ... + spi.addInputRegister(inputRegisters[i]); + } + + // 3.1 创建 Modbus TCP 从站 + ModbusSlave slave = ModbusSlaveFactory.createTCPSlave(PORT, 5); + slave.addProcessImage(SLAVE_ID, spi); + // 3.2 启动从站 + slave.open(); + + System.out.println("==================================================="); + System.out.println("Modbus TCP 从站模拟器已启动"); + System.out.println("端口: " + PORT); + System.out.println("从站地址 (Slave ID): " + SLAVE_ID); + System.out.println("==================================================="); + System.out.println("可用寄存器:"); + System.out.println(" - 线圈 (Coil, 功能码 01/05): 地址 0-9"); + System.out.println(" - 离散输入 (Discrete Input, 功能码 02): 地址 0-9"); + System.out.println(" - 保持寄存器 (Holding Register, 功能码 03/06/16): 地址 0-19"); + System.out.println(" - 输入寄存器 (Input Register, 功能码 04): 地址 0-19"); + System.out.println("==================================================="); + System.out.println("按 Ctrl+C 停止模拟器"); + + // 4. 添加关闭钩子 + Runtime.getRuntime().addShutdownHook(new Thread(() -> { + System.out.println("\n正在关闭模拟器..."); + slave.close(); + System.out.println("模拟器已关闭"); + })); + + // 5. 保持运行，定时更新输入寄存器模拟数据变化 + int counter = 0; + while (true) { + Thread.sleep(5000); // 每 5 秒更新一次 + counter++; + + // 更新输入寄存器的值，模拟传感器数据变化 + for (int i = 0; i < 20; i++) { + int newValue = (i * 10 + 1) + counter; + inputRegisters[i].setValue(newValue); + } + + // 更新保持寄存器的第一个值 + spi.getRegister(0).setValue(counter * 100); +// System.out.println("[" + java.time.LocalTime.now() + "] 数据已更新, counter=" + counter +// + ", 保持寄存器[0]=" + (counter * 100) +// + ", 输入寄存器[0]=" + (1 + counter)); + } + } + +} -- libgit2 0.21.4