From c608b81c4e9e347d42a56a0847cf94de449ee4f7 Mon Sep 17 00:00:00 2001 From: YunaiV Date: Sun, 8 Feb 2026 16:34:22 +0800 Subject: [PATCH] feat（iot）：modbus-tcp-slave 优化大量代码，主要是 polling 轮询的复用； --- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/AbstractIotProtocolDownstreamSubscriber.java | 79 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/IotProtocolDownstreamSubscriber.java | 79 ------------------------------------------------------------------------------- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/coap/handler/downstream/IotCoapDownstreamSubscriber.java | 4 ++-- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/downstream/IotEmqxDownstreamSubscriber.java | 4 ++-- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/http/handler/downstream/IotHttpDownstreamSubscriber.java | 4 ++-- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/common/IotModbusUtils.java | 515 ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/common/manager/AbstractIotModbusPollScheduler.java | 269 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/common/utils/IotModbusCommonUtils.java | 526 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/common/utils/IotModbusTcpMasterUtils.java | 195 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpmaster/client/IotModbusTcpClientUtils.java | 195 --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpmaster/handler/downstream/IotModbusTcpDownstreamHandler.java | 12 ++++++------ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpmaster/handler/downstream/IotModbusTcpDownstreamSubscriber.java | 4 ++-- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpmaster/handler/upstream/IotModbusTcpUpstreamHandler.java | 4 ++-- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpmaster/manager/IotModbusTcpPollScheduler.java | 252 +++++++++++------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/codec/IotModbusFrame.java | 4 ++-- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/codec/IotModbusFrameDecoder.java | 20 ++++++++++---------- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/codec/IotModbusFrameEncoder.java | 8 ++++---- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/handler/downstream/IotModbusTcpSlaveDownstreamHandler.java | 12 ++++++------ yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/handler/downstream/IotModbusTcpSlaveDownstreamSubscriber.java | 4 ++-- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/handler/upstream/IotModbusTcpSlaveUpstreamHandler.java | 9 ++++----- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/manager/IotModbusTcpSlaveConnectionManager.java | 4 ++-- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/manager/IotModbusTcpSlavePollScheduler.java | 282 ++++++++++++++++++++---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/handler/downstream/IotMqttDownstreamSubscriber.java | 4 ++-- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/handler/downstream/IotTcpDownstreamSubscriber.java | 4 ++-- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/udp/handler/downstream/IotUdpDownstreamSubscriber.java | 4 ++-- yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/websocket/handler/downstream/IotWebSocketDownstreamSubscriber.java | 4 ++-- yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/IotModbusTcpSlaveModbusRtuIntegrationTest.java | 4 ++-- 27 files changed, 1156 insertions(+), 1349 deletions(-) create mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/AbstractIotProtocolDownstreamSubscriber.java delete mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/IotProtocolDownstreamSubscriber.java delete mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/common/IotModbusUtils.java create mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/common/manager/AbstractIotModbusPollScheduler.java create mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/common/utils/IotModbusCommonUtils.java create mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/common/utils/IotModbusTcpMasterUtils.java delete mode 100644 yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpmaster/client/IotModbusTcpClientUtils.java diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/AbstractIotProtocolDownstreamSubscriber.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/AbstractIotProtocolDownstreamSubscriber.java new file mode 100644 index 0000000..efd61e1 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/AbstractIotProtocolDownstreamSubscriber.java @@ -0,0 +1,79 @@ +package cn.iocoder.yudao.module.iot.gateway.protocol; + +import cn.hutool.core.util.StrUtil; +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 lombok.AllArgsConstructor; +import lombok.extern.slf4j.Slf4j; + +/** + * IoT 协议下行消息订阅者抽象类 + * + * 负责接收来自消息总线的下行消息，并委托给子类进行业务处理 + * + * @author 芋道源码 + */ +@AllArgsConstructor +@Slf4j +public abstract class AbstractIotProtocolDownstreamSubscriber implements IotMessageSubscriber { + + private final IotProtocol protocol; + + private final IotMessageBus messageBus; + + @Override + public String getTopic() { + return IotDeviceMessageUtils.buildMessageBusGatewayDeviceMessageTopic(protocol.getServerId()); + } + + /** + * 保证点对点消费，需要保证独立的 Group，所以使用 Topic 作为 Group + */ + @Override + public String getGroup() { + return getTopic(); + } + + @Override + public void start() { + messageBus.register(this); + log.info("[start][{} 下行消息订阅成功，Topic：{}]", protocol.getType().name(), getTopic()); + } + + @Override + public void stop() { + messageBus.unregister(this); + log.info("[stop][{} 下行消息订阅已停止，Topic：{}]", protocol.getType().name(), getTopic()); + } + + @Override + public void onMessage(IotDeviceMessage message) { + log.debug("[onMessage][接收到下行消息, messageId: {}, method: {}, deviceId: {}]", + message.getId(), message.getMethod(), message.getDeviceId()); + try { + // 1. 校验 + String method = message.getMethod(); + if (StrUtil.isBlank(method)) { + log.warn("[onMessage][消息方法为空, messageId: {}, deviceId: {}]", + message.getId(), message.getDeviceId()); + return; + } + + // 2. 处理下行消息 + handleMessage(message); + } catch (Exception e) { + log.error("[onMessage][处理下行消息失败, messageId: {}, method: {}, deviceId: {}]", + message.getId(), message.getMethod(), message.getDeviceId(), e); + } + } + + /** + * 处理下行消息 + * + * @param message 下行消息 + */ + protected abstract void handleMessage(IotDeviceMessage message); + +} diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/IotProtocolDownstreamSubscriber.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/IotProtocolDownstreamSubscriber.java deleted file mode 100644 index 2e2150f..0000000 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/IotProtocolDownstreamSubscriber.java +++ /dev/null @@ -1,79 +0,0 @@ -package cn.iocoder.yudao.module.iot.gateway.protocol; - -import cn.hutool.core.util.StrUtil; -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 lombok.AllArgsConstructor; -import lombok.extern.slf4j.Slf4j; - -/** - * IoT 协议下行消息订阅者抽象类 - * - * 负责接收来自消息总线的下行消息，并委托给子类进行业务处理 - * - * @author 芋道源码 - */ -@AllArgsConstructor -@Slf4j -public abstract class IotProtocolDownstreamSubscriber implements IotMessageSubscriber { - - private final IotProtocol protocol; - - private final IotMessageBus messageBus; - - @Override - public String getTopic() { - return IotDeviceMessageUtils.buildMessageBusGatewayDeviceMessageTopic(protocol.getServerId()); - } - - /** - * 保证点对点消费，需要保证独立的 Group，所以使用 Topic 作为 Group - */ - @Override - public String getGroup() { - return getTopic(); - } - - @Override - public void start() { - messageBus.register(this); - log.info("[start][{} 下行消息订阅成功，Topic：{}]", protocol.getType().name(), getTopic()); - } - - @Override - public void stop() { - messageBus.unregister(this); - log.info("[stop][{} 下行消息订阅已停止，Topic：{}]", protocol.getType().name(), getTopic()); - } - - @Override - public void onMessage(IotDeviceMessage message) { - log.debug("[onMessage][接收到下行消息, messageId: {}, method: {}, deviceId: {}]", - message.getId(), message.getMethod(), message.getDeviceId()); - try { - // 1. 校验 - String method = message.getMethod(); - if (StrUtil.isBlank(method)) { - log.warn("[onMessage][消息方法为空, messageId: {}, deviceId: {}]", - message.getId(), message.getDeviceId()); - return; - } - - // 2. 处理下行消息 - handleMessage(message); - } catch (Exception e) { - log.error("[onMessage][处理下行消息失败, messageId: {}, method: {}, deviceId: {}]", - message.getId(), message.getMethod(), message.getDeviceId(), e); - } - } - - /** - * 处理下行消息 - * - * @param message 下行消息 - */ - protected abstract void handleMessage(IotDeviceMessage message); - -} diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/coap/handler/downstream/IotCoapDownstreamSubscriber.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/coap/handler/downstream/IotCoapDownstreamSubscriber.java index 188d2e6..3309d2c 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/coap/handler/downstream/IotCoapDownstreamSubscriber.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/coap/handler/downstream/IotCoapDownstreamSubscriber.java @@ -2,7 +2,7 @@ package cn.iocoder.yudao.module.iot.gateway.protocol.coap.handler.downstream; import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus; import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; -import cn.iocoder.yudao.module.iot.gateway.protocol.IotProtocolDownstreamSubscriber; +import cn.iocoder.yudao.module.iot.gateway.protocol.AbstractIotProtocolDownstreamSubscriber; import cn.iocoder.yudao.module.iot.gateway.protocol.coap.IotCoapProtocol; import lombok.extern.slf4j.Slf4j; @@ -12,7 +12,7 @@ import lombok.extern.slf4j.Slf4j; * @author 芋道源码 */ @Slf4j -public class IotCoapDownstreamSubscriber extends IotProtocolDownstreamSubscriber { +public class IotCoapDownstreamSubscriber extends AbstractIotProtocolDownstreamSubscriber { public IotCoapDownstreamSubscriber(IotCoapProtocol protocol, IotMessageBus messageBus) { super(protocol, messageBus); diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/downstream/IotEmqxDownstreamSubscriber.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/downstream/IotEmqxDownstreamSubscriber.java index 55aaaac..e7e5de9 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/downstream/IotEmqxDownstreamSubscriber.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/downstream/IotEmqxDownstreamSubscriber.java @@ -2,7 +2,7 @@ package cn.iocoder.yudao.module.iot.gateway.protocol.emqx.handler.downstream; import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus; import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; -import cn.iocoder.yudao.module.iot.gateway.protocol.IotProtocolDownstreamSubscriber; +import cn.iocoder.yudao.module.iot.gateway.protocol.AbstractIotProtocolDownstreamSubscriber; import cn.iocoder.yudao.module.iot.gateway.protocol.emqx.IotEmqxProtocol; import lombok.extern.slf4j.Slf4j; @@ -12,7 +12,7 @@ import lombok.extern.slf4j.Slf4j; * @author 芋道源码 */ @Slf4j -public class IotEmqxDownstreamSubscriber extends IotProtocolDownstreamSubscriber { +public class IotEmqxDownstreamSubscriber extends AbstractIotProtocolDownstreamSubscriber { private final IotEmqxDownstreamHandler downstreamHandler; diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/http/handler/downstream/IotHttpDownstreamSubscriber.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/http/handler/downstream/IotHttpDownstreamSubscriber.java index bfac16c..fe94fe6 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/http/handler/downstream/IotHttpDownstreamSubscriber.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/http/handler/downstream/IotHttpDownstreamSubscriber.java @@ -3,7 +3,7 @@ package cn.iocoder.yudao.module.iot.gateway.protocol.http.handler.downstream; import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus; import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; import cn.iocoder.yudao.module.iot.gateway.protocol.IotProtocol; -import cn.iocoder.yudao.module.iot.gateway.protocol.IotProtocolDownstreamSubscriber; +import cn.iocoder.yudao.module.iot.gateway.protocol.AbstractIotProtocolDownstreamSubscriber; import lombok.extern.slf4j.Slf4j; /** @@ -13,7 +13,7 @@ import lombok.extern.slf4j.Slf4j; */ @Slf4j -public class IotHttpDownstreamSubscriber extends IotProtocolDownstreamSubscriber { +public class IotHttpDownstreamSubscriber extends AbstractIotProtocolDownstreamSubscriber { public IotHttpDownstreamSubscriber(IotProtocol protocol, IotMessageBus messageBus) { super(protocol, messageBus); diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/common/IotModbusUtils.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/common/IotModbusUtils.java deleted file mode 100644 index 227a886..0000000 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/common/IotModbusUtils.java +++ /dev/null @@ -1,515 +0,0 @@ -package cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common; - -import cn.hutool.core.collection.CollUtil; -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.IotModbusDeviceConfigRespDTO; -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 cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpslave.codec.IotModbusFrame; -import lombok.experimental.UtilityClass; -import lombok.extern.slf4j.Slf4j; - -import java.math.BigDecimal; -import java.math.RoundingMode; -import java.nio.ByteBuffer; -import java.nio.ByteOrder; - -/** - * IoT Modbus 协议工具类 - *

- * 提供 Modbus 协议全链路能力： - *

    - *
  • 协议常量：功能码（FC01~FC16）、异常掩码等
    • - *
  • 功能码判断：读/写/异常分类、可写判断、写功能码映射
    • - *
  • CRC-16/MODBUS 计算和校验
    • - *
  • 数据转换：原始值 ↔ 物模型属性值（{@link #convertToPropertyValue} / {@link #convertToRawValues}）
    • - *
  • 帧值提取：从 Modbus 帧提取寄存器/线圈值（{@link #extractValues}）
    • - *
  • 点位查找（{@link #findPoint}）
    • - *
- * - * @author 芋道源码 - */ -@UtilityClass -@Slf4j -public class IotModbusUtils { - - /** FC01: 读线圈 */ - public static final int FC_READ_COILS = 1; - /** FC02: 读离散输入 */ - public static final int FC_READ_DISCRETE_INPUTS = 2; - /** FC03: 读保持寄存器 */ - public static final int FC_READ_HOLDING_REGISTERS = 3; - /** FC04: 读输入寄存器 */ - public static final int FC_READ_INPUT_REGISTERS = 4; - - /** FC05: 写单个线圈 */ - public static final int FC_WRITE_SINGLE_COIL = 5; - /** FC06: 写单个寄存器 */ - public static final int FC_WRITE_SINGLE_REGISTER = 6; - /** FC15: 写多个线圈 */ - public static final int FC_WRITE_MULTIPLE_COILS = 15; - /** FC16: 写多个寄存器 */ - public static final int FC_WRITE_MULTIPLE_REGISTERS = 16; - - /** - * 异常响应掩码：响应帧的功能码最高位为 1 时，表示异常响应 - * 例如：请求 FC=0x03，异常响应 FC=0x83（0x03 | 0x80） - */ - public static final int FC_EXCEPTION_MASK = 0x80; - - /** - * 功能码掩码：用于从异常响应中提取原始功能码 - * 例如：异常 FC=0x83，原始 FC = 0x83 & 0x7F = 0x03 - */ - public static final int FC_MASK = 0x7F; - - // ==================== 功能码分类判断 ==================== - - /** - * 判断是否为读响应（FC01-04） - */ - public static boolean isReadResponse(int functionCode) { - return functionCode >= FC_READ_COILS && functionCode <= FC_READ_INPUT_REGISTERS; - } - - /** - * 判断是否为写响应（FC05/06/15/16） - */ - public static boolean isWriteResponse(int functionCode) { - return functionCode == FC_WRITE_SINGLE_COIL || functionCode == FC_WRITE_SINGLE_REGISTER - || functionCode == FC_WRITE_MULTIPLE_COILS || functionCode == FC_WRITE_MULTIPLE_REGISTERS; - } - - /** - * 判断是否为异常响应 - */ - public static boolean isExceptionResponse(int functionCode) { - return (functionCode & FC_EXCEPTION_MASK) != 0; - } - - /** - * 从异常响应中提取原始功能码 - */ - public static int extractOriginalFunctionCode(int exceptionFunctionCode) { - return exceptionFunctionCode & FC_MASK; - } - - /** - * 判断读功能码是否支持写操作 - *

- * FC01（读线圈）和 FC03（读保持寄存器）支持写操作； - * FC02（读离散输入）和 FC04（读输入寄存器）为只读。 - * - * @param readFunctionCode 读功能码（FC01-04） - * @return 是否支持写操作 - */ - @SuppressWarnings("BooleanMethodIsAlwaysInverted") - public static boolean isWritable(int readFunctionCode) { - return readFunctionCode == FC_READ_COILS || readFunctionCode == FC_READ_HOLDING_REGISTERS; - } - - /** - * 获取单写功能码 - *

- * FC01（读线圈）→ FC05（写单个线圈）； - * FC03（读保持寄存器）→ FC06（写单个寄存器）； - * 其他返回 null（不支持写）。 - * - * @param readFunctionCode 读功能码 - * @return 单写功能码，不支持写时返回 null - */ - @SuppressWarnings("EnhancedSwitchMigration") - public static Integer getWriteSingleFunctionCode(int readFunctionCode) { - switch (readFunctionCode) { - case FC_READ_COILS: - return FC_WRITE_SINGLE_COIL; - case FC_READ_HOLDING_REGISTERS: - return FC_WRITE_SINGLE_REGISTER; - default: - return null; - } - } - - /** - * 获取多写功能码 - *

- * FC01（读线圈）→ FC15（写多个线圈）； - * FC03（读保持寄存器）→ FC16（写多个寄存器）； - * 其他返回 null（不支持写）。 - * - * @param readFunctionCode 读功能码 - * @return 多写功能码，不支持写时返回 null - */ - @SuppressWarnings("EnhancedSwitchMigration") - public static Integer getWriteMultipleFunctionCode(int readFunctionCode) { - switch (readFunctionCode) { - case FC_READ_COILS: - return FC_WRITE_MULTIPLE_COILS; - case FC_READ_HOLDING_REGISTERS: - return FC_WRITE_MULTIPLE_REGISTERS; - default: - return null; - } - } - - // ==================== CRC16 工具 ==================== - - /** - * 计算 CRC-16/MODBUS - * - * @param data 数据 - * @param length 计算长度 - * @return CRC16 值 - */ - public static int calculateCrc16(byte[] data, int length) { - int crc = 0xFFFF; - for (int i = 0; i < length; i++) { - crc ^= (data[i] & 0xFF); - for (int j = 0; j < 8; j++) { - if ((crc & 0x0001) != 0) { - crc >>= 1; - crc ^= 0xA001; - } else { - crc >>= 1; - } - } - } - return crc; - } - - /** - * 校验 CRC16 - * - * @param data 包含 CRC 的完整数据 - * @return 校验是否通过 - */ - public static boolean verifyCrc16(byte[] data) { - if (data.length < 3) { - return false; - } - int computed = calculateCrc16(data, data.length - 2); - int received = (data[data.length - 2] & 0xFF) | ((data[data.length - 1] & 0xFF) << 8); - return computed == received; - } - - // ==================== 数据转换 ==================== - - /** - * 将原始值转换为物模型属性值 - * - * @param rawValues 原始值数组（寄存器值或线圈值） - * @param point 点位配置 - * @return 转换后的属性值 - */ - public static 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 static 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 static 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 static 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 static 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 static 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 static 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 static 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 static 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 static 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 static 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 static 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 static 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 static 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 static 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; - } - } - - // ==================== 帧值提取 ==================== - - /** - * 从帧中提取寄存器值（FC01-04 读响应） - * - * @param frame 解码后的 Modbus 帧 - * @return 寄存器值数组（int[]），失败返回 null - */ - @SuppressWarnings("EnhancedSwitchMigration") - public static int[] extractValues(IotModbusFrame frame) { - if (frame == null || frame.isException()) { - return null; - } - byte[] pdu = frame.getPdu(); - if (pdu == null || pdu.length < 1) { - return null; - } - - int functionCode = frame.getFunctionCode(); - switch (functionCode) { - case FC_READ_COILS: - case FC_READ_DISCRETE_INPUTS: - return extractCoilValues(pdu); - case FC_READ_HOLDING_REGISTERS: - case FC_READ_INPUT_REGISTERS: - return extractRegisterValues(pdu); - default: - log.warn("[extractValues][不支持的功能码: {}]", functionCode); - return null; - } - } - - private static int[] extractCoilValues(byte[] pdu) { - if (pdu.length < 2) { - return null; - } - int byteCount = pdu[0] & 0xFF; - int bitCount = byteCount * 8; - int[] values = new int[bitCount]; - for (int i = 0; i < bitCount && (1 + i / 8) < pdu.length; i++) { - values[i] = ((pdu[1 + i / 8] >> (i % 8)) & 0x01); - } - return values; - } - - private static int[] extractRegisterValues(byte[] pdu) { - if (pdu.length < 2) { - return null; - } - int byteCount = pdu[0] & 0xFF; - int registerCount = byteCount / 2; - int[] values = new int[registerCount]; - for (int i = 0; i < registerCount && (1 + i * 2 + 1) < pdu.length; i++) { - values[i] = ((pdu[1 + i * 2] & 0xFF) << 8) | (pdu[1 + i * 2 + 1] & 0xFF); - } - return values; - } - - // ==================== 点位查找 ==================== - - /** - * 查找点位配置 - * - * @param config 设备 Modbus 配置 - * @param identifier 点位标识符 - * @return 匹配的点位配置，未找到返回 null - */ - public static IotModbusPointRespDTO findPoint(IotModbusDeviceConfigRespDTO config, String identifier) { - return CollUtil.findOne(config.getPoints(), p -> identifier.equals(p.getIdentifier())); - } - -} diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/common/manager/AbstractIotModbusPollScheduler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/common/manager/AbstractIotModbusPollScheduler.java new file mode 100644 index 0000000..dedac7a --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/common/manager/AbstractIotModbusPollScheduler.java @@ -0,0 +1,269 @@ +package cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.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 io.vertx.core.Vertx; +import lombok.AllArgsConstructor; +import lombok.Data; +import lombok.extern.slf4j.Slf4j; + +import java.util.*; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedQueue; + +import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertSet; + +/** + * Modbus 轮询调度器基类 + *

+ * 封装通用的定时器管理、per-device 请求队列限速逻辑。 + * 子类只需实现 {@link #pollPoint(Long, Long)} 定义具体的轮询动作。 + *

+ * + * @author 芋道源码 + */ +@Slf4j +public abstract class AbstractIotModbusPollScheduler { + + protected final Vertx vertx; + + /** + * 同设备最小请求间隔（毫秒），防止 Modbus 设备性能不足时请求堆积 + */ + private static final long MIN_REQUEST_INTERVAL = 1000; + + /** + * 设备点位的定时器映射：deviceId -> (pointId -> PointTimerInfo) + */ + private final Map> devicePointTimers = new ConcurrentHashMap<>(); + + /** + * per-device 请求队列：deviceId -> 待执行请求队列 + */ + private final Map> deviceRequestQueues = new ConcurrentHashMap<>(); + /** + * per-device 上次请求时间戳：deviceId -> lastRequestTimeMs + */ + private final Map deviceLastRequestTime = new ConcurrentHashMap<>(); + /** + * per-device 延迟 timer 标记：deviceId -> 是否有延迟 timer 在等待 + */ + private final Map deviceDelayTimerActive = new ConcurrentHashMap<>(); + + protected AbstractIotModbusPollScheduler(Vertx vertx) { + this.vertx = vertx; + } + + /** + * 点位定时器信息 + */ + @Data + @AllArgsConstructor + private static class PointTimerInfo { + + /** + * Vert.x 定时器 ID + */ + private Long timerId; + /** + * 轮询间隔（用于判断是否需要更新定时器） + */ + private Integer pollInterval; + + } + + // ========== 轮询管理 ========== + + /** + * 更新轮询任务（增量更新） + * + * 1. 【删除】点位：停止对应的轮询定时器 + * 2. 【新增】点位：创建对应的轮询定时器 + * 3. 【修改】点位：pollInterval 变化，重建对应的轮询定时器 + * 【修改】其他属性变化：不需要重建定时器（pollPoint 运行时从 configCache 取最新 point） + */ + 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(deviceId, pointId, newPollInterval); + 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(deviceId, pointId, newPollInterval); + 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 其他属性变化：无需重建定时器，因为 pollPoint() 运行时从 configCache 获取最新 point，自动使用新配置 + } + } + + /** + * 创建轮询定时器 + */ + private Long createPollTimer(Long deviceId, Long pointId, Integer pollInterval) { + if (pollInterval == null || pollInterval <= 0) { + return null; + } + return vertx.setPeriodic(pollInterval, timerId -> { + try { + submitPollRequest(deviceId, pointId); + } catch (Exception e) { + log.error("[createPollTimer][轮询点位失败, deviceId={}, pointId={}]", deviceId, pointId, e); + } + }); + } + + // ========== 请求队列（per-device 限速） ========== + + /** + * 提交轮询请求到设备请求队列（保证同设备请求间隔） + */ + private void submitPollRequest(Long deviceId, Long pointId) { + // 1. 【重要】将请求添加到设备的请求队列 + Queue queue = deviceRequestQueues.computeIfAbsent(deviceId, k -> new ConcurrentLinkedQueue<>()); + queue.offer(() -> pollPoint(deviceId, pointId)); + + // 2. 处理设备请求队列（如果没有延迟 timer 在等待） + processDeviceQueue(deviceId); + } + + /** + * 处理设备请求队列 + */ + private void processDeviceQueue(Long deviceId) { + Queue queue = deviceRequestQueues.get(deviceId); + if (CollUtil.isEmpty(queue)) { + return; + } + // 检查是否已有延迟 timer 在等待 + if (Boolean.TRUE.equals(deviceDelayTimerActive.get(deviceId))) { + return; + } + + // 不满足间隔要求，延迟执行 + long now = System.currentTimeMillis(); + long lastTime = deviceLastRequestTime.getOrDefault(deviceId, 0L); + long elapsed = now - lastTime; + if (elapsed < MIN_REQUEST_INTERVAL) { + scheduleNextRequest(deviceId, MIN_REQUEST_INTERVAL - elapsed); + return; + } + + // 满足间隔要求，立即执行 + Runnable task = queue.poll(); + if (task == null) { + return; + } + deviceLastRequestTime.put(deviceId, now); + task.run(); + // 继续处理队列中的下一个（如果有的话，需要延迟） + if (CollUtil.isNotEmpty(queue)) { + scheduleNextRequest(deviceId); + } + } + + private void scheduleNextRequest(Long deviceId) { + scheduleNextRequest(deviceId, MIN_REQUEST_INTERVAL); + } + + private void scheduleNextRequest(Long deviceId, long delayMs) { + deviceDelayTimerActive.put(deviceId, true); + vertx.setTimer(delayMs, id -> { + deviceDelayTimerActive.put(deviceId, false); + Queue queue = deviceRequestQueues.get(deviceId); + if (CollUtil.isEmpty(queue)) { + return; + } + + // 满足间隔要求，立即执行 + Runnable task = queue.poll(); + if (task == null) { + return; + } + deviceLastRequestTime.put(deviceId, System.currentTimeMillis()); + task.run(); + // 继续处理队列中的下一个（如果有的话，需要延迟） + if (CollUtil.isNotEmpty(queue)) { + scheduleNextRequest(deviceId); + } + }); + } + + // ========== 轮询执行 ========== + + /** + * 轮询单个点位（子类实现具体的读取逻辑） + * + * @param deviceId 设备 ID + * @param pointId 点位 ID + */ + protected abstract void pollPoint(Long deviceId, Long pointId); + + // ========== 停止 ========== + + /** + * 停止设备的轮询 + */ + public void stopPolling(Long deviceId) { + Map timers = devicePointTimers.remove(deviceId); + if (CollUtil.isEmpty(timers)) { + return; + } + for (PointTimerInfo timerInfo : timers.values()) { + vertx.cancelTimer(timerInfo.getTimerId()); + } + // 清理请求队列 + deviceRequestQueues.remove(deviceId); + deviceLastRequestTime.remove(deviceId); + deviceDelayTimerActive.remove(deviceId); + 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/modbus/common/utils/IotModbusCommonUtils.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/common/utils/IotModbusCommonUtils.java new file mode 100644 index 0000000..236277b --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/common/utils/IotModbusCommonUtils.java @@ -0,0 +1,526 @@ +package cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.utils; + +import cn.hutool.core.collection.CollUtil; +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.IotModbusDeviceConfigRespDTO; +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 cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpslave.codec.IotModbusFrame; +import lombok.experimental.UtilityClass; +import lombok.extern.slf4j.Slf4j; + +import java.math.BigDecimal; +import java.math.RoundingMode; +import java.nio.ByteBuffer; +import java.nio.ByteOrder; + +/** + * IoT Modbus 协议工具类 + *

+ * 提供 Modbus 协议全链路能力： + *

    + *
  • 协议常量：功能码（FC01~FC16）、异常掩码等
    • + *
  • 功能码判断：读/写/异常分类、可写判断、写功能码映射
    • + *
  • CRC-16/MODBUS 计算和校验
    • + *
  • 数据转换：原始值 ↔ 物模型属性值（{@link #convertToPropertyValue} / {@link #convertToRawValues}）
    • + *
  • 帧值提取：从 Modbus 帧提取寄存器/线圈值（{@link #extractValues}）
    • + *
  • 点位查找（{@link #findPoint}）
    • + *
+ * + * @author 芋道源码 + */ +@UtilityClass +@Slf4j +public class IotModbusCommonUtils { + + /** FC01: 读线圈 */ + public static final int FC_READ_COILS = 1; + /** FC02: 读离散输入 */ + public static final int FC_READ_DISCRETE_INPUTS = 2; + /** FC03: 读保持寄存器 */ + public static final int FC_READ_HOLDING_REGISTERS = 3; + /** FC04: 读输入寄存器 */ + public static final int FC_READ_INPUT_REGISTERS = 4; + + /** FC05: 写单个线圈 */ + public static final int FC_WRITE_SINGLE_COIL = 5; + /** FC06: 写单个寄存器 */ + public static final int FC_WRITE_SINGLE_REGISTER = 6; + /** FC15: 写多个线圈 */ + public static final int FC_WRITE_MULTIPLE_COILS = 15; + /** FC16: 写多个寄存器 */ + public static final int FC_WRITE_MULTIPLE_REGISTERS = 16; + + /** + * 异常响应掩码：响应帧的功能码最高位为 1 时，表示异常响应 + * 例如：请求 FC=0x03，异常响应 FC=0x83（0x03 | 0x80） + */ + public static final int FC_EXCEPTION_MASK = 0x80; + + /** + * 功能码掩码：用于从异常响应中提取原始功能码 + * 例如：异常 FC=0x83，原始 FC = 0x83 & 0x7F = 0x03 + */ + public static final int FC_MASK = 0x7F; + + // ==================== 功能码分类判断 ==================== + + /** + * 判断是否为读响应（FC01-04） + */ + public static boolean isReadResponse(int functionCode) { + return functionCode >= FC_READ_COILS && functionCode <= FC_READ_INPUT_REGISTERS; + } + + /** + * 判断是否为写响应（FC05/06/15/16） + */ + public static boolean isWriteResponse(int functionCode) { + return functionCode == FC_WRITE_SINGLE_COIL || functionCode == FC_WRITE_SINGLE_REGISTER + || functionCode == FC_WRITE_MULTIPLE_COILS || functionCode == FC_WRITE_MULTIPLE_REGISTERS; + } + + /** + * 判断是否为异常响应 + */ + public static boolean isExceptionResponse(int functionCode) { + return (functionCode & FC_EXCEPTION_MASK) != 0; + } + + /** + * 从异常响应中提取原始功能码 + */ + public static int extractOriginalFunctionCode(int exceptionFunctionCode) { + return exceptionFunctionCode & FC_MASK; + } + + /** + * 判断读功能码是否支持写操作 + *

+ * FC01（读线圈）和 FC03（读保持寄存器）支持写操作； + * FC02（读离散输入）和 FC04（读输入寄存器）为只读。 + * + * @param readFunctionCode 读功能码（FC01-04） + * @return 是否支持写操作 + */ + @SuppressWarnings("BooleanMethodIsAlwaysInverted") + public static boolean isWritable(int readFunctionCode) { + return readFunctionCode == FC_READ_COILS || readFunctionCode == FC_READ_HOLDING_REGISTERS; + } + + /** + * 获取单写功能码 + *

+ * FC01（读线圈）→ FC05（写单个线圈）； + * FC03（读保持寄存器）→ FC06（写单个寄存器）； + * 其他返回 null（不支持写）。 + * + * @param readFunctionCode 读功能码 + * @return 单写功能码，不支持写时返回 null + */ + @SuppressWarnings("EnhancedSwitchMigration") + public static Integer getWriteSingleFunctionCode(int readFunctionCode) { + switch (readFunctionCode) { + case FC_READ_COILS: + return FC_WRITE_SINGLE_COIL; + case FC_READ_HOLDING_REGISTERS: + return FC_WRITE_SINGLE_REGISTER; + default: + return null; + } + } + + /** + * 获取多写功能码 + *

+ * FC01（读线圈）→ FC15（写多个线圈）； + * FC03（读保持寄存器）→ FC16（写多个寄存器）； + * 其他返回 null（不支持写）。 + * + * @param readFunctionCode 读功能码 + * @return 多写功能码，不支持写时返回 null + */ + @SuppressWarnings("EnhancedSwitchMigration") + public static Integer getWriteMultipleFunctionCode(int readFunctionCode) { + switch (readFunctionCode) { + case FC_READ_COILS: + return FC_WRITE_MULTIPLE_COILS; + case FC_READ_HOLDING_REGISTERS: + return FC_WRITE_MULTIPLE_REGISTERS; + default: + return null; + } + } + + // ==================== CRC16 工具 ==================== + + /** + * 计算 CRC-16/MODBUS + * + * @param data 数据 + * @param length 计算长度 + * @return CRC16 值 + */ + public static int calculateCrc16(byte[] data, int length) { + int crc = 0xFFFF; + for (int i = 0; i < length; i++) { + crc ^= (data[i] & 0xFF); + for (int j = 0; j < 8; j++) { + if ((crc & 0x0001) != 0) { + crc >>= 1; + crc ^= 0xA001; + } else { + crc >>= 1; + } + } + } + return crc; + } + + /** + * 校验 CRC16 + * + * @param data 包含 CRC 的完整数据 + * @return 校验是否通过 + */ + public static boolean verifyCrc16(byte[] data) { + if (data.length < 3) { + return false; + } + int computed = calculateCrc16(data, data.length - 2); + int received = (data[data.length - 2] & 0xFF) | ((data[data.length - 1] & 0xFF) << 8); + return computed == received; + } + + // ==================== 数据转换 ==================== + + /** + * 将原始值转换为物模型属性值 + * + * @param rawValues 原始值数组（寄存器值或线圈值） + * @param point 点位配置 + * @return 转换后的属性值 + */ + public static 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 static 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 static 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 static 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 static 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 static 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 static 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 static 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 static 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 static 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 static 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 static 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 static 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 static 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 static 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; + } + } + + // ==================== 帧值提取 ==================== + + /** + * 从帧中提取寄存器值（FC01-04 读响应） + * + * @param frame 解码后的 Modbus 帧 + * @return 寄存器值数组（int[]），失败返回 null + */ + @SuppressWarnings("EnhancedSwitchMigration") + public static int[] extractValues(IotModbusFrame frame) { + if (frame == null || frame.isException()) { + return null; + } + byte[] pdu = frame.getPdu(); + if (pdu == null || pdu.length < 1) { + return null; + } + + int functionCode = frame.getFunctionCode(); + switch (functionCode) { + case FC_READ_COILS: + case FC_READ_DISCRETE_INPUTS: + return extractCoilValues(pdu); + case FC_READ_HOLDING_REGISTERS: + case FC_READ_INPUT_REGISTERS: + return extractRegisterValues(pdu); + default: + log.warn("[extractValues][不支持的功能码: {}]", functionCode); + return null; + } + } + + private static int[] extractCoilValues(byte[] pdu) { + if (pdu.length < 2) { + return null; + } + int byteCount = pdu[0] & 0xFF; + int bitCount = byteCount * 8; + int[] values = new int[bitCount]; + for (int i = 0; i < bitCount && (1 + i / 8) < pdu.length; i++) { + values[i] = ((pdu[1 + i / 8] >> (i % 8)) & 0x01); + } + return values; + } + + private static int[] extractRegisterValues(byte[] pdu) { + if (pdu.length < 2) { + return null; + } + int byteCount = pdu[0] & 0xFF; + int registerCount = byteCount / 2; + int[] values = new int[registerCount]; + for (int i = 0; i < registerCount && (1 + i * 2 + 1) < pdu.length; i++) { + values[i] = ((pdu[1 + i * 2] & 0xFF) << 8) | (pdu[1 + i * 2 + 1] & 0xFF); + } + return values; + } + + // ==================== 点位查找 ==================== + + /** + * 查找点位配置 + * + * @param config 设备 Modbus 配置 + * @param identifier 点位标识符 + * @return 匹配的点位配置，未找到返回 null + */ + public static IotModbusPointRespDTO findPoint(IotModbusDeviceConfigRespDTO config, String identifier) { + return CollUtil.findOne(config.getPoints(), p -> identifier.equals(p.getIdentifier())); + } + + /** + * 根据点位 ID 查找点位配置 + * + * @param config 设备 Modbus 配置 + * @param pointId 点位 ID + * @return 匹配的点位配置，未找到返回 null + */ + public static IotModbusPointRespDTO findPointById(IotModbusDeviceConfigRespDTO config, Long pointId) { + return CollUtil.findOne(config.getPoints(), p -> p.getId().equals(pointId)); + } + +} diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/common/utils/IotModbusTcpMasterUtils.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/common/utils/IotModbusTcpMasterUtils.java new file mode 100644 index 0000000..5804ea6 --- /dev/null +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/common/utils/IotModbusTcpMasterUtils.java @@ -0,0 +1,195 @@ +package cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.utils; + +import cn.iocoder.yudao.module.iot.core.biz.dto.IotModbusPointRespDTO; +import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpmaster.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.experimental.UtilityClass; +import lombok.extern.slf4j.Slf4j; + +import static cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.utils.IotModbusCommonUtils.*; + +/** + * IoT Modbus TCP 客户端工具类 + *

+ * 封装基于 j2mod 的 Modbus TCP 读写操作： + * 1. 根据功能码创建对应的 Modbus 读/写请求 + * 2. 通过 {@link IotModbusTcpConnectionManager.ModbusConnection} 执行事务 + * 3. 从响应中提取原始值 + * + * @author 芋道源码 + */ +@UtilityClass +@Slf4j +public class IotModbusTcpMasterUtils { + + /** + * 读取 Modbus 数据 + * + * @param connection Modbus 连接 + * @param slaveId 从站地址 + * @param point 点位配置 + * @return 原始值（int 数组） + */ + public static 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 static 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 static ModbusRequest createReadRequest(Integer functionCode, Integer address, Integer count) { + switch (functionCode) { + case FC_READ_COILS: + return new ReadCoilsRequest(address, count); + case FC_READ_DISCRETE_INPUTS: + return new ReadInputDiscretesRequest(address, count); + case FC_READ_HOLDING_REGISTERS: + return new ReadMultipleRegistersRequest(address, count); + case FC_READ_INPUT_REGISTERS: + return new ReadInputRegistersRequest(address, count); + default: + throw new IllegalArgumentException("不支持的功能码: " + functionCode); + } + } + + /** + * 创建写入请求 + */ + @SuppressWarnings("EnhancedSwitchMigration") + private static ModbusRequest createWriteRequest(Integer functionCode, Integer address, Integer count, int[] values) { + switch (functionCode) { + case FC_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 FC_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 FC_READ_DISCRETE_INPUTS: // 只读 + case FC_READ_INPUT_REGISTERS: // 只读 + return null; + default: + throw new IllegalArgumentException("不支持的功能码: " + functionCode); + } + } + + /** + * 从响应中提取值 + */ + @SuppressWarnings("EnhancedSwitchMigration") + private static int[] extractValues(ModbusResponse response, Integer functionCode) { + switch (functionCode) { + case FC_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 FC_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 FC_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 FC_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/modbus/tcpmaster/client/IotModbusTcpClientUtils.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpmaster/client/IotModbusTcpClientUtils.java deleted file mode 100644 index 4472839..0000000 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpmaster/client/IotModbusTcpClientUtils.java +++ /dev/null @@ -1,195 +0,0 @@ -package cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpmaster.client; - -import cn.iocoder.yudao.module.iot.core.biz.dto.IotModbusPointRespDTO; -import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpmaster.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.experimental.UtilityClass; -import lombok.extern.slf4j.Slf4j; - -import static cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.IotModbusUtils.*; - -/** - * IoT Modbus TCP 客户端工具类 - *

- * 封装基于 j2mod 的 Modbus TCP 读写操作： - * 1. 根据功能码创建对应的 Modbus 读/写请求 - * 2. 通过 {@link IotModbusTcpConnectionManager.ModbusConnection} 执行事务 - * 3. 从响应中提取原始值 - * - * @author 芋道源码 - */ -@UtilityClass -@Slf4j -public class IotModbusTcpClientUtils { - - /** - * 读取 Modbus 数据 - * - * @param connection Modbus 连接 - * @param slaveId 从站地址 - * @param point 点位配置 - * @return 原始值（int 数组） - */ - public static 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 static 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 static ModbusRequest createReadRequest(Integer functionCode, Integer address, Integer count) { - switch (functionCode) { - case FC_READ_COILS: - return new ReadCoilsRequest(address, count); - case FC_READ_DISCRETE_INPUTS: - return new ReadInputDiscretesRequest(address, count); - case FC_READ_HOLDING_REGISTERS: - return new ReadMultipleRegistersRequest(address, count); - case FC_READ_INPUT_REGISTERS: - return new ReadInputRegistersRequest(address, count); - default: - throw new IllegalArgumentException("不支持的功能码: " + functionCode); - } - } - - /** - * 创建写入请求 - */ - @SuppressWarnings("EnhancedSwitchMigration") - private static ModbusRequest createWriteRequest(Integer functionCode, Integer address, Integer count, int[] values) { - switch (functionCode) { - case FC_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 FC_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 FC_READ_DISCRETE_INPUTS: // 只读 - case FC_READ_INPUT_REGISTERS: // 只读 - return null; - default: - throw new IllegalArgumentException("不支持的功能码: " + functionCode); - } - } - - /** - * 从响应中提取值 - */ - @SuppressWarnings("EnhancedSwitchMigration") - private static int[] extractValues(ModbusResponse response, Integer functionCode) { - switch (functionCode) { - case FC_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 FC_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 FC_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 FC_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/modbus/tcpmaster/handler/downstream/IotModbusTcpDownstreamHandler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpmaster/handler/downstream/IotModbusTcpDownstreamHandler.java index 27df39e..32dfde5 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpmaster/handler/downstream/IotModbusTcpDownstreamHandler.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpmaster/handler/downstream/IotModbusTcpDownstreamHandler.java @@ -4,8 +4,8 @@ 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.modbus.common.IotModbusUtils; -import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpmaster.client.IotModbusTcpClientUtils; +import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.utils.IotModbusCommonUtils; +import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.utils.IotModbusTcpMasterUtils; import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpmaster.manager.IotModbusTcpConfigCacheService; import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpmaster.manager.IotModbusTcpConnectionManager; import lombok.RequiredArgsConstructor; @@ -57,13 +57,13 @@ public class IotModbusTcpDownstreamHandler { String identifier = entry.getKey(); Object value = entry.getValue(); // 2.1 查找对应的点位配置 - IotModbusPointRespDTO point = IotModbusUtils.findPoint(config, identifier); + IotModbusPointRespDTO point = IotModbusCommonUtils.findPoint(config, identifier); if (point == null) { log.warn("[handle][设备 {} 没有点位配置: {}]", message.getDeviceId(), identifier); continue; } // 2.2 检查是否支持写操作 - if (!IotModbusUtils.isWritable(point.getFunctionCode())) { + if (!IotModbusCommonUtils.isWritable(point.getFunctionCode())) { log.warn("[handle][点位 {} 不支持写操作, 功能码={}]", identifier, point.getFunctionCode()); continue; } @@ -91,9 +91,9 @@ public class IotModbusTcpDownstreamHandler { } // 2.1 转换属性值为原始值 - int[] rawValues = IotModbusUtils.convertToRawValues(value, point); + int[] rawValues = IotModbusCommonUtils.convertToRawValues(value, point); // 2.2 执行 Modbus 写入 - IotModbusTcpClientUtils.write(connection, slaveId, point, rawValues) + IotModbusTcpMasterUtils.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={}]", diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpmaster/handler/downstream/IotModbusTcpDownstreamSubscriber.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpmaster/handler/downstream/IotModbusTcpDownstreamSubscriber.java index c8e22af..74afcbb 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpmaster/handler/downstream/IotModbusTcpDownstreamSubscriber.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpmaster/handler/downstream/IotModbusTcpDownstreamSubscriber.java @@ -2,7 +2,7 @@ package cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpmaster.handler.do import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus; import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; -import cn.iocoder.yudao.module.iot.gateway.protocol.IotProtocolDownstreamSubscriber; +import cn.iocoder.yudao.module.iot.gateway.protocol.AbstractIotProtocolDownstreamSubscriber; import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpmaster.IotModbusTcpMasterProtocol; import lombok.extern.slf4j.Slf4j; @@ -12,7 +12,7 @@ import lombok.extern.slf4j.Slf4j; * @author 芋道源码 */ @Slf4j -public class IotModbusTcpDownstreamSubscriber extends IotProtocolDownstreamSubscriber { +public class IotModbusTcpDownstreamSubscriber extends AbstractIotProtocolDownstreamSubscriber { private final IotModbusTcpDownstreamHandler downstreamHandler; diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpmaster/handler/upstream/IotModbusTcpUpstreamHandler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpmaster/handler/upstream/IotModbusTcpUpstreamHandler.java index 8762481..de02af0 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpmaster/handler/upstream/IotModbusTcpUpstreamHandler.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpmaster/handler/upstream/IotModbusTcpUpstreamHandler.java @@ -5,7 +5,7 @@ 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.modbus.common.IotModbusUtils; +import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.utils.IotModbusCommonUtils; import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService; import lombok.extern.slf4j.Slf4j; @@ -40,7 +40,7 @@ public class IotModbusTcpUpstreamHandler { int[] rawValue) { try { // 1.1 转换原始值为物模型属性值 - Object convertedValue = IotModbusUtils.convertToPropertyValue(rawValue, point); + Object convertedValue = IotModbusCommonUtils.convertToPropertyValue(rawValue, point); log.debug("[handleReadResult][设备={}, 属性={}, 原始值={}, 转换值={}]", config.getDeviceId(), point.getIdentifier(), rawValue, convertedValue); // 1.2 构造属性上报消息 diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpmaster/manager/IotModbusTcpPollScheduler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpmaster/manager/IotModbusTcpPollScheduler.java index afe4a9c..2f04956 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpmaster/manager/IotModbusTcpPollScheduler.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpmaster/manager/IotModbusTcpPollScheduler.java @@ -1,242 +1,45 @@ package cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpmaster.manager; import cn.hutool.core.collection.CollUtil; +import cn.hutool.core.lang.Assert; 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.modbus.tcpmaster.client.IotModbusTcpClientUtils; +import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.manager.AbstractIotModbusPollScheduler; +import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.utils.IotModbusCommonUtils; +import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.utils.IotModbusTcpMasterUtils; import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpmaster.handler.upstream.IotModbusTcpUpstreamHandler; import io.vertx.core.Vertx; -import lombok.AllArgsConstructor; -import lombok.Data; import lombok.extern.slf4j.Slf4j; -import java.util.*; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentLinkedQueue; - -import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertSet; - -// TODO @AI：类的命名上，要体现上 master。其它类似 /Users/yunai/Java/ruoyi-vue-pro-jdk25/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpmaster 也要！ /** * IoT Modbus TCP Master 轮询调度器：管理点位的轮询定时器，调度读取任务并上报结果 * * @author 芋道源码 */ @Slf4j -public class IotModbusTcpPollScheduler { +public class IotModbusTcpPollScheduler extends AbstractIotModbusPollScheduler { - private final Vertx vertx; private final IotModbusTcpConnectionManager connectionManager; private final IotModbusTcpUpstreamHandler upstreamHandler; private final IotModbusTcpConfigCacheService configCacheService; - /** - * 同设备最小请求间隔（毫秒），防止 Modbus 设备性能不足时请求堆积 - */ - private static final long MIN_REQUEST_INTERVAL = 100; - - /** - * 设备点位的定时器映射：deviceId -> (pointId -> PointTimerInfo) - */ - private final Map> devicePointTimers = new ConcurrentHashMap<>(); - - /** - * per-device 请求队列：deviceId -> 待执行请求队列 - */ - private final Map> deviceRequestQueues = new ConcurrentHashMap<>(); - - /** - * per-device 上次请求时间戳：deviceId -> lastRequestTimeMs - */ - private final Map deviceLastRequestTime = new ConcurrentHashMap<>(); - - /** - * per-device 延迟 timer 标记：deviceId -> 是否有延迟 timer 在等待 - */ - private final Map deviceDelayTimerActive = new ConcurrentHashMap<>(); - public IotModbusTcpPollScheduler(Vertx vertx, IotModbusTcpConnectionManager connectionManager, IotModbusTcpUpstreamHandler upstreamHandler, IotModbusTcpConfigCacheService configCacheService) { - this.vertx = vertx; + super(vertx); this.connectionManager = connectionManager; this.upstreamHandler = upstreamHandler; this.configCacheService = configCacheService; } - // ========== 点位定时器 ========== - - /** - * 点位定时器信息 - */ - @Data - @AllArgsConstructor - private static class PointTimerInfo { - - /** - * Vert.x 定时器 ID - */ - private Long timerId; - /** - * 轮询间隔（用于判断是否需要更新定时器） - */ - private Integer pollInterval; - - } - - // ========== 轮询管理 ========== - - /** - * 更新轮询任务（增量更新） - * - * 1. 【删除】点位：停止对应的轮询定时器 - * 2. 【新增】点位：创建对应的轮询定时器 - * 3. 【修改】点位：pollInterval 变化，重建对应的轮询定时器 - * 4. 其他属性变化：不需要重建定时器（pollPoint 运行时从 configCache 取最新 point） - */ - 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(deviceId, pointId, newPollInterval); - 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(deviceId, pointId, newPollInterval); - 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 其他属性变化：无需重建定时器，因为 pollPoint() 运行时从 configCache 获取最新 point，自动使用新配置 - } - } - - /** - * 创建轮询定时器 - *

- * 闭包只捕获 deviceId 和 pointId，运行时从 configCache 获取最新配置，避免旧快照问题。 - */ - private Long createPollTimer(Long deviceId, Long pointId, Integer pollInterval) { - if (pollInterval == null || pollInterval <= 0) { - return null; - } - return vertx.setPeriodic(pollInterval, timerId -> { - try { - submitPollRequest(deviceId, pointId); - } catch (Exception e) { - log.error("[createPollTimer][轮询点位失败, deviceId={}, pointId={}]", deviceId, pointId, e); - } - }); - } - - // ========== 请求队列（per-device 限速） ========== - - /** - * 提交轮询请求到设备请求队列（保证同设备请求间隔） - */ - private void submitPollRequest(Long deviceId, Long pointId) { - Queue queue = deviceRequestQueues.computeIfAbsent(deviceId, k -> new ConcurrentLinkedQueue<>()); - queue.offer(() -> pollPoint(deviceId, pointId)); - processDeviceQueue(deviceId); - } - - /** - * 处理设备请求队列 - */ - private void processDeviceQueue(Long deviceId) { - Queue queue = deviceRequestQueues.get(deviceId); - if (CollUtil.isEmpty(queue)) { - return; - } - // 检查是否已有延迟 timer 在等待 - if (Boolean.TRUE.equals(deviceDelayTimerActive.get(deviceId))) { - return; - } - long now = System.currentTimeMillis(); - long lastTime = deviceLastRequestTime.getOrDefault(deviceId, 0L); - long elapsed = now - lastTime; - if (elapsed >= MIN_REQUEST_INTERVAL) { - // 满足间隔要求，立即执行 - Runnable task = queue.poll(); - if (task != null) { - deviceLastRequestTime.put(deviceId, now); - task.run(); - // 继续处理队列中的下一个（如果有的话，需要延迟） - if (!queue.isEmpty()) { - scheduleNextRequest(deviceId); - } - } - } else { - // 需要延迟 - scheduleNextRequest(deviceId, MIN_REQUEST_INTERVAL - elapsed); - } - } - - private void scheduleNextRequest(Long deviceId) { - scheduleNextRequest(deviceId, MIN_REQUEST_INTERVAL); - } - - private void scheduleNextRequest(Long deviceId, long delayMs) { - deviceDelayTimerActive.put(deviceId, true); - vertx.setTimer(delayMs, id -> { - deviceDelayTimerActive.put(deviceId, false); - Queue queue = deviceRequestQueues.get(deviceId); - if (CollUtil.isEmpty(queue)) { - Runnable task = queue.poll(); - if (task != null) { - deviceLastRequestTime.put(deviceId, System.currentTimeMillis()); - task.run(); - } - // 继续处理 - if (queue != null && !queue.isEmpty()) { - scheduleNextRequest(deviceId); - } - } - }); - } - // ========== 轮询执行 ========== /** * 轮询单个点位 */ - private void pollPoint(Long deviceId, Long pointId) { + @Override + protected void pollPoint(Long deviceId, Long pointId) { // 1.1 从 configCache 获取最新配置 IotModbusDeviceConfigRespDTO config = configCacheService.getConfig(deviceId); if (config == null || CollUtil.isEmpty(config.getPoints())) { @@ -244,8 +47,7 @@ public class IotModbusTcpPollScheduler { return; } // 1.2 查找点位 - // TODO @AI：是不是这里，可以抽到 IotModbusUtils 里？感觉应该有几个地方需要的； - IotModbusPointRespDTO point = CollUtil.findOne(config.getPoints(), p -> p.getId().equals(pointId)); + IotModbusPointRespDTO point = IotModbusCommonUtils.findPointById(config, pointId); if (point == null) { log.warn("[pollPoint][设备 {} 点位 {} 未找到]", deviceId, pointId); return; @@ -259,45 +61,13 @@ public class IotModbusTcpPollScheduler { } // 2.2 获取 slave ID Integer slaveId = connectionManager.getSlaveId(deviceId); - if (slaveId == null) { - log.warn("[pollPoint][设备 {} 没有 slaveId]", deviceId); - return; - } + Assert.notNull(slaveId, "设备 {} 没有配置 slaveId", deviceId); // 3. 执行 Modbus 读取 - IotModbusTcpClientUtils.read(connection, slaveId, point) + IotModbusTcpMasterUtils.read(connection, slaveId, point) .onSuccess(rawValue -> upstreamHandler.handleReadResult(config, point, rawValue)) .onFailure(e -> log.error("[pollPoint][读取点位失败, deviceId={}, identifier={}]", deviceId, 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()); - } - // 清理请求队列 - deviceRequestQueues.remove(deviceId); - deviceLastRequestTime.remove(deviceId); - deviceDelayTimerActive.remove(deviceId); - 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/modbus/tcpslave/codec/IotModbusFrame.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/codec/IotModbusFrame.java index 6ac6930..b7661ab 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/codec/IotModbusFrame.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/codec/IotModbusFrame.java @@ -1,7 +1,7 @@ package cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpslave.codec; import cn.iocoder.yudao.module.iot.core.enums.IotModbusFrameFormatEnum; -import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.IotModbusUtils; +import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.utils.IotModbusCommonUtils; import lombok.Data; import lombok.experimental.Accessors; @@ -39,7 +39,7 @@ public class IotModbusFrame { * 当功能码最高位为 1 时（异常响应），此字段存储异常码。 * 为 null 表示非异常响应。 * - * @see IotModbusUtils#FC_EXCEPTION_MASK + * @see IotModbusCommonUtils#FC_EXCEPTION_MASK */ private Integer exceptionCode; diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/codec/IotModbusFrameDecoder.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/codec/IotModbusFrameDecoder.java index 09d98e0..fc5219e 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/codec/IotModbusFrameDecoder.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/codec/IotModbusFrameDecoder.java @@ -1,7 +1,7 @@ package cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpslave.codec; import cn.iocoder.yudao.module.iot.core.enums.IotModbusFrameFormatEnum; -import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.IotModbusUtils; +import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.utils.IotModbusCommonUtils; import io.vertx.core.Handler; import io.vertx.core.buffer.Buffer; import io.vertx.core.parsetools.RecordParser; @@ -96,7 +96,7 @@ public class IotModbusFrameDecoder { return null; } // 校验 CRC - if (!IotModbusUtils.verifyCrc16(data)) { + if (!IotModbusCommonUtils.verifyCrc16(data)) { log.warn("[decodeRtuResponse][CRC 校验失败]"); return null; } @@ -119,8 +119,8 @@ public class IotModbusFrameDecoder { .setPdu(pdu) .setTransactionId(transactionId); // 异常响应 - if (IotModbusUtils.isExceptionResponse(functionCode)) { - frame.setFunctionCode(IotModbusUtils.extractOriginalFunctionCode(functionCode)); + if (IotModbusCommonUtils.isExceptionResponse(functionCode)) { + frame.setFunctionCode(IotModbusCommonUtils.extractOriginalFunctionCode(functionCode)); if (pdu.length >= 1) { frame.setExceptionCode(pdu[0] & 0xFF); } @@ -281,7 +281,7 @@ public class IotModbusFrameDecoder { this.slaveId = bytes[0]; this.functionCode = bytes[1]; int fc = functionCode & 0xFF; - if (IotModbusUtils.isExceptionResponse(fc)) { + if (IotModbusCommonUtils.isExceptionResponse(fc)) { // 异常响应：完整帧 = slaveId(1) + FC(1) + exceptionCode(1) + CRC(2) = 5 字节 // 已有 6 字节（多 1 字节），取前 5 字节组装 Buffer frame = Buffer.buffer(5); @@ -290,7 +290,7 @@ public class IotModbusFrameDecoder { frame.appendBytes(bytes, 2, 3); // exceptionCode + CRC emitFrame(frame); resetToHeader(); - } else if (IotModbusUtils.isReadResponse(fc) || fc == customFunctionCode) { + } else if (IotModbusCommonUtils.isReadResponse(fc) || fc == customFunctionCode) { // 读响应或自定义 FC：bytes[2] = byteCount this.byteCount = bytes[2]; int bc = byteCount & 0xFF; @@ -315,7 +315,7 @@ public class IotModbusFrameDecoder { this.expectedDataLen = bc + 2; // byteCount 个数据 + 2 CRC parser.fixedSizeMode(remaining); } - } else if (IotModbusUtils.isWriteResponse(fc)) { + } else if (IotModbusCommonUtils.isWriteResponse(fc)) { // 写响应：总长 = slaveId(1) + FC(1) + addr(2) + value/qty(2) + CRC(2) = 8 字节 // 已有 6 字节，还需 2 字节 state = STATE_WRITE_BODY; @@ -356,15 +356,15 @@ public class IotModbusFrameDecoder { this.slaveId = header[0]; this.functionCode = header[1]; int fc = functionCode & 0xFF; - if (IotModbusUtils.isExceptionResponse(fc)) { + if (IotModbusCommonUtils.isExceptionResponse(fc)) { // 异常响应 state = STATE_EXCEPTION_BODY; parser.fixedSizeMode(3); // exceptionCode(1) + CRC(2) - } else if (IotModbusUtils.isReadResponse(fc) || fc == customFunctionCode) { + } else if (IotModbusCommonUtils.isReadResponse(fc) || fc == customFunctionCode) { // 读响应或自定义 FC state = STATE_READ_BYTE_COUNT; parser.fixedSizeMode(1); // byteCount - } else if (IotModbusUtils.isWriteResponse(fc)) { + } else if (IotModbusCommonUtils.isWriteResponse(fc)) { // 写响应 state = STATE_WRITE_BODY; pendingData = Buffer.buffer(); diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/codec/IotModbusFrameEncoder.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/codec/IotModbusFrameEncoder.java index 195c823..36323cb 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/codec/IotModbusFrameEncoder.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/codec/IotModbusFrameEncoder.java @@ -1,7 +1,7 @@ package cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpslave.codec; import cn.iocoder.yudao.module.iot.core.enums.IotModbusFrameFormatEnum; -import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.IotModbusUtils; +import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.utils.IotModbusCommonUtils; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; @@ -60,7 +60,7 @@ public class IotModbusFrameEncoder { public byte[] encodeWriteSingleRequest(int slaveId, int functionCode, int address, int value, IotModbusFrameFormatEnum format, Integer transactionId) { // FC05 单写线圈：Modbus 标准要求 value 为 0xFF00（ON）或 0x0000（OFF） - if (functionCode == IotModbusUtils.FC_WRITE_SINGLE_COIL) { + if (functionCode == IotModbusCommonUtils.FC_WRITE_SINGLE_COIL) { value = (value != 0) ? 0xFF00 : 0x0000; } // PDU: [FC(1)] [Address(2)] [Value(2)] @@ -120,7 +120,7 @@ public class IotModbusFrameEncoder { int quantity = values.length; int byteCount = (quantity + 7) / 8; // 向上取整 byte[] pdu = new byte[6 + byteCount]; - pdu[0] = (byte) IotModbusUtils.FC_WRITE_MULTIPLE_COILS; // FC15 + pdu[0] = (byte) IotModbusCommonUtils.FC_WRITE_MULTIPLE_COILS; // FC15 pdu[1] = (byte) ((address >> 8) & 0xFF); pdu[2] = (byte) (address & 0xFF); pdu[3] = (byte) ((quantity >> 8) & 0xFF); @@ -204,7 +204,7 @@ public class IotModbusFrameEncoder { frame[0] = (byte) slaveId; System.arraycopy(pdu, 0, frame, 1, pdu.length); // 计算并追加 CRC16 - int crc = IotModbusUtils.calculateCrc16(frame, frame.length - 2); + int crc = IotModbusCommonUtils.calculateCrc16(frame, frame.length - 2); frame[frame.length - 2] = (byte) (crc & 0xFF); // CRC Low frame[frame.length - 1] = (byte) ((crc >> 8) & 0xFF); // CRC High return frame; diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/handler/downstream/IotModbusTcpSlaveDownstreamHandler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/handler/downstream/IotModbusTcpSlaveDownstreamHandler.java index ca9edde..770f13e 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/handler/downstream/IotModbusTcpSlaveDownstreamHandler.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/handler/downstream/IotModbusTcpSlaveDownstreamHandler.java @@ -7,7 +7,7 @@ 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.enums.IotModbusFrameFormatEnum; import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; -import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.IotModbusUtils; +import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.utils.IotModbusCommonUtils; import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpslave.codec.IotModbusFrameEncoder; import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpslave.manager.IotModbusTcpSlaveConfigCacheService; import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpslave.manager.IotModbusTcpSlaveConnectionManager; @@ -82,13 +82,13 @@ public class IotModbusTcpSlaveDownstreamHandler { String identifier = entry.getKey(); Object value = entry.getValue(); // 2.1 查找对应的点位配置 - IotModbusPointRespDTO point = IotModbusUtils.findPoint(config, identifier); + IotModbusPointRespDTO point = IotModbusCommonUtils.findPoint(config, identifier); if (point == null) { log.warn("[handle][设备 {} 没有点位配置: {}]", message.getDeviceId(), identifier); continue; } // 2.2 检查是否支持写操作 - if (!IotModbusUtils.isWritable(point.getFunctionCode())) { + if (!IotModbusCommonUtils.isWritable(point.getFunctionCode())) { log.warn("[handle][点位 {} 不支持写操作, 功能码={}]", identifier, point.getFunctionCode()); continue; } @@ -104,7 +104,7 @@ public class IotModbusTcpSlaveDownstreamHandler { private void writeProperty(Long deviceId, ConnectionInfo connInfo, IotModbusPointRespDTO point, Object value) { // 1.1 转换属性值为原始值 - int[] rawValues = IotModbusUtils.convertToRawValues(value, point); + int[] rawValues = IotModbusCommonUtils.convertToRawValues(value, point); // 1.2 确定帧格式和事务 ID IotModbusFrameFormatEnum frameFormat = connInfo.getFrameFormat(); @@ -117,8 +117,8 @@ public class IotModbusTcpSlaveDownstreamHandler { // 1.3 编码写请求 byte[] data; int readFunctionCode = point.getFunctionCode(); - Integer writeSingleCode = IotModbusUtils.getWriteSingleFunctionCode(readFunctionCode); - Integer writeMultipleCode = IotModbusUtils.getWriteMultipleFunctionCode(readFunctionCode); + Integer writeSingleCode = IotModbusCommonUtils.getWriteSingleFunctionCode(readFunctionCode); + Integer writeMultipleCode = IotModbusCommonUtils.getWriteMultipleFunctionCode(readFunctionCode); if (rawValues.length == 1 && writeSingleCode != null) { // 单个值：使用单写功能码（FC05/FC06） data = frameEncoder.encodeWriteSingleRequest(slaveId, writeSingleCode, diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/handler/downstream/IotModbusTcpSlaveDownstreamSubscriber.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/handler/downstream/IotModbusTcpSlaveDownstreamSubscriber.java index 4e7b882..2a11cff 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/handler/downstream/IotModbusTcpSlaveDownstreamSubscriber.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/handler/downstream/IotModbusTcpSlaveDownstreamSubscriber.java @@ -2,7 +2,7 @@ package cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpslave.handler.dow import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus; import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; -import cn.iocoder.yudao.module.iot.gateway.protocol.IotProtocolDownstreamSubscriber; +import cn.iocoder.yudao.module.iot.gateway.protocol.AbstractIotProtocolDownstreamSubscriber; import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpslave.IotModbusTcpSlaveProtocol; import lombok.extern.slf4j.Slf4j; @@ -12,7 +12,7 @@ import lombok.extern.slf4j.Slf4j; * @author 芋道源码 */ @Slf4j -public class IotModbusTcpSlaveDownstreamSubscriber extends IotProtocolDownstreamSubscriber { +public class IotModbusTcpSlaveDownstreamSubscriber extends AbstractIotProtocolDownstreamSubscriber { private final IotModbusTcpSlaveDownstreamHandler downstreamHandler; diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/handler/upstream/IotModbusTcpSlaveUpstreamHandler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/handler/upstream/IotModbusTcpSlaveUpstreamHandler.java index 67b2d52..6ede74c 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/handler/upstream/IotModbusTcpSlaveUpstreamHandler.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/handler/upstream/IotModbusTcpSlaveUpstreamHandler.java @@ -19,7 +19,7 @@ import cn.iocoder.yudao.module.iot.core.enums.IotModbusFrameFormatEnum; import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; import cn.iocoder.yudao.module.iot.core.topic.IotDeviceIdentity; import cn.iocoder.yudao.module.iot.core.util.IotDeviceAuthUtils; -import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.IotModbusUtils; +import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.utils.IotModbusCommonUtils; import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpslave.codec.IotModbusFrame; import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpslave.codec.IotModbusFrameEncoder; import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpslave.manager.IotModbusTcpSlaveConfigCacheService; @@ -237,7 +237,7 @@ public class IotModbusTcpSlaveUpstreamHandler { return; } // 2.2 提取寄存器值 - int[] rawValues = IotModbusUtils.extractValues(frame); + int[] rawValues = IotModbusCommonUtils.extractValues(frame); if (rawValues == null) { log.warn("[handlePollingResponse][提取寄存器值失败, deviceId={}, identifier={}]", info.getDeviceId(), request.getIdentifier()); @@ -248,14 +248,13 @@ public class IotModbusTcpSlaveUpstreamHandler { if (config == null || CollUtil.isEmpty(config.getPoints())) { return; } - IotModbusPointRespDTO point = CollUtil.findOne(config.getPoints(), - p -> p.getId().equals(request.getPointId())); + IotModbusPointRespDTO point = IotModbusCommonUtils.findPointById(config, request.getPointId()); if (point == null) { return; } // 3.1 点位翻译 - Object convertedValue = IotModbusUtils.convertToPropertyValue(rawValues, point); + Object convertedValue = IotModbusCommonUtils.convertToPropertyValue(rawValues, point); // 3.2 上报属性 Map params = MapUtil.of(request.getIdentifier(), convertedValue); IotDeviceMessage message = IotDeviceMessage.requestOf( diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/manager/IotModbusTcpSlaveConnectionManager.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/manager/IotModbusTcpSlaveConnectionManager.java index 241d8d7..1a6a4cc 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/manager/IotModbusTcpSlaveConnectionManager.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/manager/IotModbusTcpSlaveConnectionManager.java @@ -75,8 +75,8 @@ public class IotModbusTcpSlaveConnectionManager { try { oldSocket.close(); } catch (Exception e) { - // TODO @AI：这里日志可以打的更完整一点，方便追溯；比如：设备 ID、旧连接地址等 - log.warn("[registerConnection][关闭旧 socket 失败]", e); + log.warn("[registerConnection][关闭旧 socket 失败, deviceId={}, oldRemote={}]", + info.getDeviceId(), oldSocket.remoteAddress(), e); } } diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/manager/IotModbusTcpSlavePollScheduler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/manager/IotModbusTcpSlavePollScheduler.java index d0f38e3..8a9cba5 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/manager/IotModbusTcpSlavePollScheduler.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/manager/IotModbusTcpSlavePollScheduler.java @@ -1,42 +1,28 @@ package cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpslave.manager; import cn.hutool.core.collection.CollUtil; +import cn.hutool.core.lang.Assert; 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.IotModbusFrameFormatEnum; +import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.manager.AbstractIotModbusPollScheduler; +import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.utils.IotModbusCommonUtils; import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpslave.codec.IotModbusFrameEncoder; import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpslave.manager.IotModbusTcpSlaveConnectionManager.ConnectionInfo; import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpslave.manager.IotModbusTcpSlavePendingRequestManager.PendingRequest; import io.vertx.core.Vertx; -import lombok.AllArgsConstructor; -import lombok.Data; import lombok.extern.slf4j.Slf4j; -import java.util.*; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.atomic.AtomicInteger; -import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertSet; - /** - * IoT Modbus TCP Slave 轮询调度器 - *

- * 管理点位的轮询定时器，为云端轮询模式的设备调度读取任务。 - * 与 tcpmaster 的 {@code IotModbusTcpPollScheduler} 不同，这里不通过 j2mod 直接读取，而是： - * 1. 编码 Modbus 读请求帧 - * 2. 通过 ConnectionManager 发送到设备的 TCP 连接 - * 3. 将请求注册到 PendingRequestManager，等待设备响应 - *

- * 闭包只捕获 deviceId + pointId，运行时从 configCacheService 获取最新 config 和 point， - * 避免闭包捕获旧快照导致上报消息用旧身份的问题。 + * IoT Modbus TCP Slave 轮询调度器：编码读请求帧，通过 TCP 连接发送到设备，注册 PendingRequest 等待响应 * * @author 芋道源码 */ @Slf4j -public class IotModbusTcpSlavePollScheduler { +public class IotModbusTcpSlavePollScheduler extends AbstractIotModbusPollScheduler { - private final Vertx vertx; private final IotModbusTcpSlaveConnectionManager connectionManager; private final IotModbusFrameEncoder frameEncoder; private final IotModbusTcpSlavePendingRequestManager pendingRequestManager; @@ -47,29 +33,6 @@ public class IotModbusTcpSlavePollScheduler { */ private final AtomicInteger transactionIdCounter; - /** - * 同设备最小请求间隔（毫秒），防止 Modbus 设备性能不足时请求堆积 - */ - private static final long MIN_REQUEST_INTERVAL = 200; - - /** - * 设备点位的定时器映射：deviceId -> (pointId -> PointTimerInfo) - */ - private final Map> devicePointTimers = new ConcurrentHashMap<>(); - - /** - * per-device 请求队列：deviceId -> 待执行请求队列 - */ - private final Map> deviceRequestQueues = new ConcurrentHashMap<>(); - /** - * per-device 上次请求时间戳：deviceId -> lastRequestTimeMs - */ - private final Map deviceLastRequestTime = new ConcurrentHashMap<>(); - /** - * per-device 延迟 timer 标记：deviceId -> 是否有延迟 timer 在等待 - */ - private final Map deviceDelayTimerActive = new ConcurrentHashMap<>(); - public IotModbusTcpSlavePollScheduler(Vertx vertx, IotModbusTcpSlaveConnectionManager connectionManager, IotModbusFrameEncoder frameEncoder, @@ -77,7 +40,7 @@ public class IotModbusTcpSlavePollScheduler { int requestTimeout, AtomicInteger transactionIdCounter, IotModbusTcpSlaveConfigCacheService configCacheService) { - this.vertx = vertx; + super(vertx); this.connectionManager = connectionManager; this.frameEncoder = frameEncoder; this.pendingRequestManager = pendingRequestManager; @@ -86,185 +49,13 @@ public class IotModbusTcpSlavePollScheduler { this.configCacheService = configCacheService; } - /** - * 点位定时器信息 - */ - @Data - @AllArgsConstructor - private static class PointTimerInfo { - - /** - * Vert.x 定时器 ID - */ - private Long timerId; - /** - * 轮询间隔（用于判断是否需要更新定时器） - */ - private Integer pollInterval; - - } - - // ========== 轮询管理 ========== - - /** - * 更新轮询任务（增量更新） - * - * 1. 【删除】点位：停止对应的轮询定时器 - * 2. 【新增】点位：创建对应的轮询定时器 - * 3. 【修改】点位：pollInterval 变化，重建对应的轮询定时器 - * 4. 其他属性变化：不需要重建定时器（pollPoint 运行时从 configCache 取最新 point） - */ - 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(deviceId, pointId, newPollInterval); - 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(deviceId, pointId, newPollInterval); - 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 其他属性变化：无需重建定时器，因为 pollPoint() 运行时从 configCache 获取最新 point，自动使用新配置 - } - } - - /** - * 创建轮询定时器 - *

- * 闭包只捕获 deviceId 和 pointId，运行时从 configCache 获取最新配置，避免旧快照问题。 - */ - private Long createPollTimer(Long deviceId, Long pointId, Integer pollInterval) { - if (pollInterval == null || pollInterval <= 0) { - return null; - } - return vertx.setPeriodic(pollInterval, timerId -> { - try { - submitPollRequest(deviceId, pointId); - } catch (Exception e) { - log.error("[createPollTimer][轮询点位失败, deviceId={}, pointId={}]", deviceId, pointId, e); - } - }); - } - - // ========== 请求队列（per-device 限速） ========== - - /** - * 提交轮询请求到设备请求队列（保证同设备请求间隔） - */ - private void submitPollRequest(Long deviceId, Long pointId) { - // 1. 将请求添加到设备的请求队列 - Queue queue = deviceRequestQueues.computeIfAbsent(deviceId, k -> new ConcurrentLinkedQueue<>()); - queue.offer(() -> pollPoint(deviceId, pointId)); - - // 2. 处理设备请求队列（如果没有延迟 timer 在等待） - processDeviceQueue(deviceId); - } - - /** - * 处理设备请求队列 - */ - private void processDeviceQueue(Long deviceId) { - Queue queue = deviceRequestQueues.get(deviceId); - if (CollUtil.isEmpty(queue)) { - return; - } - // 检查是否已有延迟 timer 在等待 - if (Boolean.TRUE.equals(deviceDelayTimerActive.get(deviceId))) { - return; - } - - // TODO @AI：可以改成不满足间隔，然后 return，简化括号层级； - long now = System.currentTimeMillis(); - long lastTime = deviceLastRequestTime.getOrDefault(deviceId, 0L); - long elapsed = now - lastTime; - - if (elapsed >= MIN_REQUEST_INTERVAL) { - // 满足间隔要求，立即执行 - Runnable task = queue.poll(); - if (task != null) { - deviceLastRequestTime.put(deviceId, now); - task.run(); - // 继续处理队列中的下一个（如果有的话，需要延迟） - if (!queue.isEmpty()) { - scheduleNextRequest(deviceId); - } - } - } else { - // 需要延迟 - scheduleNextRequest(deviceId, MIN_REQUEST_INTERVAL - elapsed); - } - } - - private void scheduleNextRequest(Long deviceId) { - scheduleNextRequest(deviceId, MIN_REQUEST_INTERVAL); - } - - private void scheduleNextRequest(Long deviceId, long delayMs) { - deviceDelayTimerActive.put(deviceId, true); - vertx.setTimer(delayMs, id -> { - deviceDelayTimerActive.put(deviceId, false); - Queue queue = deviceRequestQueues.get(deviceId); - // TODO @AI：if return？简化下？ - if (CollUtil.isEmpty(queue)) { - Runnable task = queue.poll(); - if (task != null) { - deviceLastRequestTime.put(deviceId, System.currentTimeMillis()); - task.run(); - } - // 继续处理 - if (queue != null && !queue.isEmpty()) { - scheduleNextRequest(deviceId); - } - } - }); - } - // ========== 轮询执行 ========== /** * 轮询单个点位 - *

- * 运行时从 configCacheService 获取最新的 config 和 point，而非使用闭包捕获的旧引用。 */ - private void pollPoint(Long deviceId, Long pointId) { + @Override + protected void pollPoint(Long deviceId, Long pointId) { // 1.1 从 configCache 获取最新配置 IotModbusDeviceConfigRespDTO config = configCacheService.getConfig(deviceId); if (config == null || CollUtil.isEmpty(config.getPoints())) { @@ -272,34 +63,31 @@ public class IotModbusTcpSlavePollScheduler { return; } // 1.2 查找点位 - IotModbusPointRespDTO point = CollUtil.findOne(config.getPoints(), p -> p.getId().equals(pointId)); + IotModbusPointRespDTO point = IotModbusCommonUtils.findPointById(config, pointId); if (point == null) { log.warn("[pollPoint][设备 {} 点位 {} 未找到]", deviceId, pointId); return; } - // 2. 获取连接信息 - ConnectionInfo connInfo = connectionManager.getConnectionInfoByDeviceId(deviceId); - if (connInfo == null) { + // 2.1 获取连接 + ConnectionInfo connection = connectionManager.getConnectionInfoByDeviceId(deviceId); + if (connection == null) { log.debug("[pollPoint][设备 {} 没有连接，跳过轮询]", deviceId); return; } + // 2.2 获取 slave ID + IotModbusFrameFormatEnum frameFormat = connection.getFrameFormat(); + Assert.notNull(frameFormat, "设备 {} 的帧格式不能为空", deviceId); + int slaveId = connection.getSlaveId(); + Assert.notNull(connection.getSlaveId(), "设备 {} 的 slaveId 不能为空", deviceId); - // 3.1 确定帧格式和事务 ID - IotModbusFrameFormatEnum frameFormat = connInfo.getFrameFormat(); - if (frameFormat == null) { - log.warn("[pollPoint][设备 {} 帧格式为空，跳过轮询]", deviceId); - return; - } + // 3.1 编码读请求 Integer transactionId = frameFormat == IotModbusFrameFormatEnum.MODBUS_TCP ? (transactionIdCounter.incrementAndGet() & 0xFFFF) : null; - // TODO @AI：这里断言必须非空！ - int slaveId = connInfo.getSlaveId() != null ? connInfo.getSlaveId() : 1; - // 3.2 编码读请求 byte[] data = frameEncoder.encodeReadRequest(slaveId, point.getFunctionCode(), point.getRegisterAddress(), point.getRegisterCount(), frameFormat, transactionId); - // 3.3 注册 PendingRequest + // 3.2 注册 PendingRequest PendingRequest pendingRequest = new PendingRequest( deviceId, point.getId(), point.getIdentifier(), slaveId, point.getFunctionCode(), @@ -307,41 +95,11 @@ public class IotModbusTcpSlavePollScheduler { transactionId, System.currentTimeMillis() + requestTimeout); pendingRequestManager.addRequest(pendingRequest); - - // 4. 发送读请求 + // 3.3 发送读请求 connectionManager.sendToDevice(deviceId, data); log.debug("[pollPoint][设备={}, 点位={}, FC={}, 地址={}, 数量={}]", deviceId, point.getIdentifier(), point.getFunctionCode(), point.getRegisterAddress(), point.getRegisterCount()); } - // ========== 停止 ========== - - /** - * 停止设备的轮询 - */ - public void stopPolling(Long deviceId) { - Map timers = devicePointTimers.remove(deviceId); - if (CollUtil.isEmpty(timers)) { - return; - } - for (PointTimerInfo timerInfo : timers.values()) { - vertx.cancelTimer(timerInfo.getTimerId()); - } - // 清理请求队列 - deviceRequestQueues.remove(deviceId); - deviceLastRequestTime.remove(deviceId); - deviceDelayTimerActive.remove(deviceId); - 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/mqtt/handler/downstream/IotMqttDownstreamSubscriber.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/handler/downstream/IotMqttDownstreamSubscriber.java index c8aa299..5f0b547 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/handler/downstream/IotMqttDownstreamSubscriber.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/handler/downstream/IotMqttDownstreamSubscriber.java @@ -2,7 +2,7 @@ package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.handler.downstream; import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus; import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; -import cn.iocoder.yudao.module.iot.gateway.protocol.IotProtocolDownstreamSubscriber; +import cn.iocoder.yudao.module.iot.gateway.protocol.AbstractIotProtocolDownstreamSubscriber; import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttProtocol; import lombok.extern.slf4j.Slf4j; @@ -12,7 +12,7 @@ import lombok.extern.slf4j.Slf4j; * @author 芋道源码 */ @Slf4j -public class IotMqttDownstreamSubscriber extends IotProtocolDownstreamSubscriber { +public class IotMqttDownstreamSubscriber extends AbstractIotProtocolDownstreamSubscriber { private final IotMqttDownstreamHandler downstreamHandler; diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/handler/downstream/IotTcpDownstreamSubscriber.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/handler/downstream/IotTcpDownstreamSubscriber.java index 7a29e6c..39a7384 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/handler/downstream/IotTcpDownstreamSubscriber.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/handler/downstream/IotTcpDownstreamSubscriber.java @@ -3,7 +3,7 @@ package cn.iocoder.yudao.module.iot.gateway.protocol.tcp.handler.downstream; import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus; import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; import cn.iocoder.yudao.module.iot.gateway.protocol.IotProtocol; -import cn.iocoder.yudao.module.iot.gateway.protocol.IotProtocolDownstreamSubscriber; +import cn.iocoder.yudao.module.iot.gateway.protocol.AbstractIotProtocolDownstreamSubscriber; import lombok.extern.slf4j.Slf4j; /** @@ -12,7 +12,7 @@ import lombok.extern.slf4j.Slf4j; * @author 芋道源码 */ @Slf4j -public class IotTcpDownstreamSubscriber extends IotProtocolDownstreamSubscriber { +public class IotTcpDownstreamSubscriber extends AbstractIotProtocolDownstreamSubscriber { private final IotTcpDownstreamHandler downstreamHandler; diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/udp/handler/downstream/IotUdpDownstreamSubscriber.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/udp/handler/downstream/IotUdpDownstreamSubscriber.java index ea0bc99..cc21df6 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/udp/handler/downstream/IotUdpDownstreamSubscriber.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/udp/handler/downstream/IotUdpDownstreamSubscriber.java @@ -3,7 +3,7 @@ package cn.iocoder.yudao.module.iot.gateway.protocol.udp.handler.downstream; import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus; import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; import cn.iocoder.yudao.module.iot.gateway.protocol.IotProtocol; -import cn.iocoder.yudao.module.iot.gateway.protocol.IotProtocolDownstreamSubscriber; +import cn.iocoder.yudao.module.iot.gateway.protocol.AbstractIotProtocolDownstreamSubscriber; import lombok.extern.slf4j.Slf4j; /** @@ -12,7 +12,7 @@ import lombok.extern.slf4j.Slf4j; * @author 芋道源码 */ @Slf4j -public class IotUdpDownstreamSubscriber extends IotProtocolDownstreamSubscriber { +public class IotUdpDownstreamSubscriber extends AbstractIotProtocolDownstreamSubscriber { private final IotUdpDownstreamHandler downstreamHandler; diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/websocket/handler/downstream/IotWebSocketDownstreamSubscriber.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/websocket/handler/downstream/IotWebSocketDownstreamSubscriber.java index efe5f43..c565be2 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/websocket/handler/downstream/IotWebSocketDownstreamSubscriber.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/websocket/handler/downstream/IotWebSocketDownstreamSubscriber.java @@ -2,7 +2,7 @@ package cn.iocoder.yudao.module.iot.gateway.protocol.websocket.handler.downstrea import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus; import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage; -import cn.iocoder.yudao.module.iot.gateway.protocol.IotProtocolDownstreamSubscriber; +import cn.iocoder.yudao.module.iot.gateway.protocol.AbstractIotProtocolDownstreamSubscriber; import cn.iocoder.yudao.module.iot.gateway.protocol.websocket.IotWebSocketProtocol; import lombok.extern.slf4j.Slf4j; @@ -12,7 +12,7 @@ import lombok.extern.slf4j.Slf4j; * @author 芋道源码 */ @Slf4j -public class IotWebSocketDownstreamSubscriber extends IotProtocolDownstreamSubscriber { +public class IotWebSocketDownstreamSubscriber extends AbstractIotProtocolDownstreamSubscriber { private final IotWebSocketDownstreamHandler downstreamHandler; diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/IotModbusTcpSlaveModbusRtuIntegrationTest.java b/yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/IotModbusTcpSlaveModbusRtuIntegrationTest.java index faea89e..e103d6b 100644 --- a/yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/IotModbusTcpSlaveModbusRtuIntegrationTest.java +++ b/yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/protocol/modbus/tcpslave/IotModbusTcpSlaveModbusRtuIntegrationTest.java @@ -8,7 +8,7 @@ import cn.iocoder.yudao.module.iot.core.util.IotDeviceAuthUtils; import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpslave.codec.IotModbusFrame; import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpslave.codec.IotModbusFrameDecoder; import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.tcpslave.codec.IotModbusFrameEncoder; -import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.IotModbusUtils; +import cn.iocoder.yudao.module.iot.gateway.protocol.modbus.common.utils.IotModbusCommonUtils; import io.vertx.core.Vertx; import io.vertx.core.buffer.Buffer; import io.vertx.core.net.NetClient; @@ -285,7 +285,7 @@ public class IotModbusTcpSlaveModbusRtuIntegrationTest { frame[3 + i * 2 + 1] = (byte) (registerValues[i] & 0xFF); } // 计算 CRC16 - int crc = IotModbusUtils.calculateCrc16(frame, totalLength - 2); + int crc = IotModbusCommonUtils.calculateCrc16(frame, totalLength - 2); frame[totalLength - 2] = (byte) (crc & 0xFF); // CRC Low frame[totalLength - 1] = (byte) ((crc >> 8) & 0xFF); // CRC High return frame; -- libgit2 0.21.4