🎉 Initial commit

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/biz/IotDeviceCommonApi.java

@@ -0,0 +1,58 @@
+package cn.iocoder.yudao.module.iot.core.biz;
+
+import cn.iocoder.yudao.framework.common.pojo.CommonResult;
+import cn.iocoder.yudao.module.iot.core.biz.dto.*;
+import cn.iocoder.yudao.module.iot.core.topic.auth.IotDeviceRegisterReqDTO;
+import cn.iocoder.yudao.module.iot.core.topic.auth.IotDeviceRegisterRespDTO;
+import cn.iocoder.yudao.module.iot.core.topic.auth.IotSubDeviceRegisterRespDTO;
+
+import java.util.List;
+
+/**
+ * IoT 设备通用 API
+ *
+ * @author haohao
+ */
+public interface IotDeviceCommonApi {
+
+    /**
+     * 设备认证
+     *
+     * @param authReqDTO 认证请求
+     * @return 认证结果
+     */
+    CommonResult<Boolean> authDevice(IotDeviceAuthReqDTO authReqDTO);
+
+    /**
+     * 获取设备信息
+     *
+     * @param infoReqDTO 设备信息请求
+     * @return 设备信息
+     */
+    CommonResult<IotDeviceRespDTO> getDevice(IotDeviceGetReqDTO infoReqDTO);
+
+    /**
+     * 直连/网关设备动态注册（一型一密）
+     *
+     * @param reqDTO 动态注册请求
+     * @return 注册结果（包含 DeviceSecret）
+     */
+    CommonResult<IotDeviceRegisterRespDTO> registerDevice(IotDeviceRegisterReqDTO reqDTO);
+
+    /**
+     * 网关子设备动态注册（网关代理转发）
+     *
+     * @param reqDTO 子设备注册请求（包含网关标识和子设备列表）
+     * @return 注册结果列表
+     */
+    CommonResult<List<IotSubDeviceRegisterRespDTO>> registerSubDevices(IotSubDeviceRegisterFullReqDTO reqDTO);
+
+    /**
+     * 获取 Modbus 设备配置列表
+     *
+     * @param listReqDTO 查询参数
+     * @return Modbus 设备配置列表
+     */
+    CommonResult<List<IotModbusDeviceConfigRespDTO>> getModbusDeviceConfigList(IotModbusDeviceConfigListReqDTO listReqDTO);
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/biz/dto/IotDeviceAuthReqDTO.java

@@ -0,0 +1,36 @@
+package cn.iocoder.yudao.module.iot.core.biz.dto;
+
+import jakarta.validation.constraints.NotEmpty;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * IoT 设备认证 Request DTO
+ *
+ * @author 芋道源码
+ */
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class IotDeviceAuthReqDTO {
+
+    /**
+     * 客户端 ID
+     */
+    @NotEmpty(message = "客户端 ID 不能为空")
+    private String clientId;
+
+    /**
+     * 用户名
+     */
+    @NotEmpty(message = "用户名不能为空")
+    private String username;
+
+    /**
+     * 密码
+     */
+    @NotEmpty(message = "密码不能为空")
+    private String password;
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/biz/dto/IotDeviceGetReqDTO.java

@@ -0,0 +1,27 @@
+package cn.iocoder.yudao.module.iot.core.biz.dto;
+
+import lombok.Data;
+
+/**
+ * IoT 设备信息查询 Request DTO
+ *
+ * @author 芋道源码
+ */
+@Data
+public class IotDeviceGetReqDTO {
+
+    /**
+     * 设备编号
+     */
+    private Long id;
+
+    /**
+     * 产品标识
+     */
+    private String productKey;
+    /**
+     * 设备名称
+     */
+    private String deviceName;
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/biz/dto/IotDeviceRespDTO.java

@@ -0,0 +1,45 @@
+package cn.iocoder.yudao.module.iot.core.biz.dto;
+
+import lombok.Data;
+
+/**
+ * IoT 设备信息 Response DTO
+ *
+ * @author 芋道源码
+ */
+@Data
+public class IotDeviceRespDTO {
+
+    /**
+     * 设备编号
+     */
+    private Long id;
+    /**
+     * 产品标识
+     */
+    private String productKey;
+    /**
+     * 设备名称
+     */
+    private String deviceName;
+    /**
+     * 租户编号
+     */
+    private Long tenantId;
+
+    // ========== 产品相关字段 ==========
+
+    /**
+     * 产品编号
+     */
+    private Long productId;
+    /**
+     * 协议类型
+     */
+    private String protocolType;
+    /**
+     * 序列化类型
+     */
+    private String serializeType;
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/biz/dto/IotModbusDeviceConfigListReqDTO.java

@@ -0,0 +1,37 @@
+package cn.iocoder.yudao.module.iot.core.biz.dto;
+
+import lombok.Data;
+import lombok.experimental.Accessors;
+
+import java.util.Set;
+
+/**
+ * IoT Modbus 设备配置列表查询 Request DTO
+ *
+ * @author 芋道源码
+ */
+@Data
+@Accessors(chain = true)
+public class IotModbusDeviceConfigListReqDTO {
+
+    /**
+     * 状态
+     */
+    private Integer status;
+
+    /**
+     * 模式
+     */
+    private Integer mode;
+
+    /**
+     * 协议类型
+     */
+    private String protocolType;
+
+    /**
+     * 设备 ID 集合
+     */
+    private Set<Long> deviceIds;
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/biz/dto/IotModbusDeviceConfigRespDTO.java

@@ -0,0 +1,66 @@
+package cn.iocoder.yudao.module.iot.core.biz.dto;
+
+import lombok.Data;
+
+import java.util.List;
+
+/**
+ * IoT Modbus 设备配置 Response DTO
+ *
+ * @author 芋道源码
+ */
+@Data
+public class IotModbusDeviceConfigRespDTO {
+
+    /**
+     * 设备编号
+     */
+    private Long deviceId;
+    /**
+     * 产品标识
+     */
+    private String productKey;
+    /**
+     * 设备名称
+     */
+    private String deviceName;
+
+    // ========== Modbus 连接配置 ==========
+
+    /**
+     * Modbus 服务器 IP 地址
+     */
+    private String ip;
+    /**
+     * Modbus 服务器端口
+     */
+    private Integer port;
+    /**
+     * 从站地址
+     */
+    private Integer slaveId;
+    /**
+     * 连接超时时间，单位：毫秒
+     */
+    private Integer timeout;
+    /**
+     * 重试间隔，单位：毫秒
+     */
+    private Integer retryInterval;
+    /**
+     * 模式
+     */
+    private Integer mode;
+    /**
+     * 数据帧格式
+     */
+    private Integer frameFormat;
+
+    // ========== Modbus 点位配置 ==========
+
+    /**
+     * 点位列表
+     */
+    private List<IotModbusPointRespDTO> points;
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/biz/dto/IotModbusPointRespDTO.java

@@ -0,0 +1,67 @@
+package cn.iocoder.yudao.module.iot.core.biz.dto;
+
+import cn.iocoder.yudao.module.iot.core.enums.modbus.IotModbusByteOrderEnum;
+import cn.iocoder.yudao.module.iot.core.enums.modbus.IotModbusRawDataTypeEnum;
+import lombok.Data;
+
+import java.math.BigDecimal;
+
+/**
+ * IoT Modbus 点位配置 Response DTO
+ *
+ * @author 芋道源码
+ */
+@Data
+public class IotModbusPointRespDTO {
+
+    /**
+     * 点位编号
+     */
+    private Long id;
+    /**
+     * 属性标识符（物模型的 identifier）
+     */
+    private String identifier;
+    /**
+     * 属性名称（物模型的 name）
+     */
+    private String name;
+
+    // ========== Modbus 协议配置 ==========
+
+    /**
+     * Modbus 功能码
+     *
+     * 取值范围：FC01-04（读线圈、读离散输入、读保持寄存器、读输入寄存器）
+     */
+    private Integer functionCode;
+    /**
+     * 寄存器起始地址
+     */
+    private Integer registerAddress;
+    /**
+     * 寄存器数量
+     */
+    private Integer registerCount;
+    /**
+     * 字节序
+     *
+     * 枚举 {@link IotModbusByteOrderEnum}
+     */
+    private String byteOrder;
+    /**
+     * 原始数据类型
+     *
+     * 枚举 {@link IotModbusRawDataTypeEnum}
+     */
+    private String rawDataType;
+    /**
+     * 缩放因子
+     */
+    private BigDecimal scale;
+    /**
+     * 轮询间隔（毫秒）
+     */
+    private Integer pollInterval;
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/biz/dto/IotSubDeviceRegisterFullReqDTO.java

@@ -0,0 +1,38 @@
+package cn.iocoder.yudao.module.iot.core.biz.dto;
+
+import cn.iocoder.yudao.module.iot.core.topic.auth.IotSubDeviceRegisterReqDTO;
+import jakarta.validation.constraints.NotEmpty;
+import jakarta.validation.constraints.NotNull;
+import lombok.Data;
+
+import java.util.List;
+
+/**
+ * IoT 子设备动态注册 Request DTO
+ * <p>
+ * 额外包含了网关设备的标识信息
+ *
+ * @author 芋道源码
+ */
+@Data
+public class IotSubDeviceRegisterFullReqDTO {
+
+    /**
+     * 网关设备 ProductKey
+     */
+    @NotEmpty(message = "网关产品标识不能为空")
+    private String gatewayProductKey;
+
+    /**
+     * 网关设备 DeviceName
+     */
+    @NotEmpty(message = "网关设备名称不能为空")
+    private String gatewayDeviceName;
+
+    /**
+     * 子设备注册列表
+     */
+    @NotNull(message = "子设备注册列表不能为空")
+    private List<IotSubDeviceRegisterReqDTO> subDevices;
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/IotDeviceMessageMethodEnum.java

@@ -0,0 +1,103 @@
+package cn.iocoder.yudao.module.iot.core.enums;
+
+import cn.hutool.core.util.ArrayUtil;
+import cn.iocoder.yudao.framework.common.core.ArrayValuable;
+import cn.iocoder.yudao.framework.common.util.collection.SetUtils;
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+
+import java.util.Arrays;
+import java.util.Set;
+
+/**
+ * IoT 设备消息的方法枚举
+ *
+ * @author haohao
+ */
+@Getter
+@AllArgsConstructor
+public enum IotDeviceMessageMethodEnum implements ArrayValuable<String> {
+
+    // ========== 设备状态 ==========
+
+    STATE_UPDATE("thing.state.update", "设备状态更新", true),
+
+    // TODO 芋艿：要不要加个 ping 消息；
+
+    // ========== 拓扑管理 ==========
+    // 可参考：https://help.aliyun.com/zh/iot/user-guide/manage-topological-relationships
+
+    TOPO_ADD("thing.topo.add", "添加拓扑关系", true),
+    TOPO_DELETE("thing.topo.delete", "删除拓扑关系", true),
+    TOPO_GET("thing.topo.get", "获取拓扑关系", true),
+    TOPO_CHANGE("thing.topo.change", "拓扑关系变更通知", false),
+
+    // ========== 设备注册 ==========
+    // 可参考：https://help.aliyun.com/zh/iot/user-guide/unique-certificate-per-product-verification
+
+    DEVICE_REGISTER("thing.auth.register", "设备动态注册", true),
+    SUB_DEVICE_REGISTER("thing.auth.register.sub", "子设备动态注册", true),
+
+    // ========== 设备属性 ==========
+    // 可参考：https://help.aliyun.com/zh/iot/user-guide/device-properties-events-and-services
+
+    PROPERTY_POST("thing.property.post", "属性上报", true),
+    PROPERTY_SET("thing.property.set", "属性设置", false),
+
+    PROPERTY_PACK_POST("thing.event.property.pack.post", "批量上报（属性 + 事件 + 子设备）", true), // 网关独有
+
+    // ========== 设备事件 ==========
+    // 可参考：https://help.aliyun.com/zh/iot/user-guide/device-properties-events-and-services
+
+    EVENT_POST("thing.event.post", "事件上报", true),
+
+    // ========== 设备服务调用 ==========
+    // 可参考：https://help.aliyun.com/zh/iot/user-guide/device-properties-events-and-services
+
+    SERVICE_INVOKE("thing.service.invoke", "服务调用", false),
+
+    // ========== 设备配置 ==========
+    // 可参考：https://help.aliyun.com/zh/iot/user-guide/remote-configuration-1
+
+    CONFIG_PUSH("thing.config.push", "配置推送", false),
+
+    // ========== OTA 固件 ==========
+    // 可参考：https://help.aliyun.com/zh/iot/user-guide/perform-ota-updates
+
+    OTA_UPGRADE("thing.ota.upgrade", "OTA 固件信息推送", false),
+    OTA_PROGRESS("thing.ota.progress", "OTA 升级进度上报", true),
+
+    ;
+
+    public static final String[] ARRAYS = Arrays.stream(values()).map(IotDeviceMessageMethodEnum::getMethod)
+            .toArray(String[]::new);
+
+    /**
+     * 不进行 reply 回复的方法集合
+     */
+    public static final Set<String> REPLY_DISABLED = SetUtils.asSet(
+            STATE_UPDATE.getMethod(),
+            OTA_PROGRESS.getMethod() // 参考阿里云，OTA 升级进度上报，不进行回复
+    );
+
+    private final String method;
+
+    private final String name;
+
+    private final Boolean upstream;
+
+    @Override
+    public String[] array() {
+        return ARRAYS;
+    }
+
+    public static IotDeviceMessageMethodEnum of(String method) {
+        return ArrayUtil.firstMatch(item -> item.getMethod().equals(method),
+                IotDeviceMessageMethodEnum.values());
+    }
+
+    public static boolean isReplyDisabled(String method) {
+        return REPLY_DISABLED.contains(method);
+    }
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/IotProtocolTypeEnum.java

@@ -0,0 +1,47 @@
+package cn.iocoder.yudao.module.iot.core.enums;
+
+import cn.hutool.core.util.ArrayUtil;
+import cn.iocoder.yudao.framework.common.core.ArrayValuable;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+import java.util.Arrays;
+
+/**
+ * IoT 协议类型枚举
+ *
+ * 用于定义传输层协议类型
+ *
+ * @author 芋道源码
+ */
+@RequiredArgsConstructor
+@Getter
+public enum IotProtocolTypeEnum implements ArrayValuable<String> {
+
+    TCP("tcp"),
+    UDP("udp"),
+    WEBSOCKET("websocket"),
+    HTTP("http"),
+    MQTT("mqtt"),
+    EMQX("emqx"),
+    COAP("coap"),
+    MODBUS_TCP_CLIENT("modbus_tcp_client"),
+    MODBUS_TCP_SERVER("modbus_tcp_server");
+
+    public static final String[] ARRAYS = Arrays.stream(values()).map(IotProtocolTypeEnum::getType).toArray(String[]::new);
+
+    /**
+     * 类型
+     */
+    private final String type;
+
+    @Override
+    public String[] array() {
+        return ARRAYS;
+    }
+
+    public static IotProtocolTypeEnum of(String type) {
+        return ArrayUtil.firstMatch(e -> e.getType().equals(type), values());
+    }
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/IotSerializeTypeEnum.java

@@ -0,0 +1,40 @@
+package cn.iocoder.yudao.module.iot.core.enums;
+
+import cn.hutool.core.util.ArrayUtil;
+import cn.iocoder.yudao.framework.common.core.ArrayValuable;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+import java.util.Arrays;
+
+/**
+ * IoT 序列化类型枚举
+ *
+ * 用于定义设备消息的序列化格式
+ *
+ * @author 芋道源码
+ */
+@RequiredArgsConstructor
+@Getter
+public enum IotSerializeTypeEnum implements ArrayValuable<String> {
+
+    JSON("json"),
+    BINARY("binary");
+
+    public static final String[] ARRAYS = Arrays.stream(values()).map(IotSerializeTypeEnum::getType).toArray(String[]::new);
+
+    /**
+     * 类型
+     */
+    private final String type;
+
+    @Override
+    public String[] array() {
+        return ARRAYS;
+    }
+
+    public static IotSerializeTypeEnum of(String type) {
+        return ArrayUtil.firstMatch(e -> e.getType().equals(type), values());
+    }
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/device/IotDeviceStateEnum.java

@@ -0,0 +1,46 @@
+package cn.iocoder.yudao.module.iot.core.enums.device;
+
+import cn.iocoder.yudao.framework.common.core.ArrayValuable;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+import java.util.Arrays;
+
+/**
+ * IoT 设备状态枚举
+ *
+ * @author haohao
+ */
+@RequiredArgsConstructor
+@Getter
+public enum IotDeviceStateEnum implements ArrayValuable<Integer> {
+
+    INACTIVE(0, "未激活"),
+    ONLINE(1, "在线"),
+    OFFLINE(2, "离线");
+
+    public static final Integer[] ARRAYS = Arrays.stream(values()).map(IotDeviceStateEnum::getState).toArray(Integer[]::new);
+
+    /**
+     * 状态
+     */
+    private final Integer state;
+    /**
+     * 状态名
+     */
+    private final String name;
+
+    @Override
+    public Integer[] array() {
+        return ARRAYS;
+    }
+
+    public static boolean isOnline(Integer state) {
+        return ONLINE.getState().equals(state);
+    }
+
+    public static boolean isNotOnline(Integer state) {
+        return !isOnline(state);
+    }
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/modbus/IotModbusByteOrderEnum.java

@@ -0,0 +1,54 @@
+package cn.iocoder.yudao.module.iot.core.enums.modbus;
+
+import cn.iocoder.yudao.framework.common.core.ArrayValuable;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+import java.util.Arrays;
+
+/**
+ * IoT Modbus 字节序枚举
+ *
+ * @author 芋道源码
+ */
+@Getter
+@RequiredArgsConstructor
+public enum IotModbusByteOrderEnum implements ArrayValuable<String> {
+
+    AB("AB", "大端序（16位）", 2),
+    BA("BA", "小端序（16位）", 2),
+    ABCD("ABCD", "大端序（32位）", 4),
+    CDAB("CDAB", "大端字交换（32位）", 4),
+    DCBA("DCBA", "小端序（32位）", 4),
+    BADC("BADC", "小端字交换（32位）", 4);
+
+    public static final String[] ARRAYS = Arrays.stream(values())
+            .map(IotModbusByteOrderEnum::getOrder)
+            .toArray(String[]::new);
+
+    /**
+     * 字节序
+     */
+    private final String order;
+    /**
+     * 名称
+     */
+    private final String name;
+    /**
+     * 字节数
+     */
+    private final Integer byteCount;
+
+    @Override
+    public String[] array() {
+        return ARRAYS;
+    }
+
+    public static IotModbusByteOrderEnum getByOrder(String order) {
+        return Arrays.stream(values())
+                .filter(e -> e.getOrder().equals(order))
+                .findFirst()
+                .orElse(null);
+    }
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/modbus/IotModbusFrameFormatEnum.java

@@ -0,0 +1,35 @@
+package cn.iocoder.yudao.module.iot.core.enums.modbus;
+
+import cn.iocoder.yudao.framework.common.core.ArrayValuable;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+import java.util.Arrays;
+
+/**
+ * IoT Modbus 数据帧格式枚举
+ *
+ * @author 芋道源码
+ */
+@Getter
+@RequiredArgsConstructor
+public enum IotModbusFrameFormatEnum implements ArrayValuable<Integer> {
+
+    MODBUS_TCP(1),
+    MODBUS_RTU(2);
+
+    public static final Integer[] ARRAYS = Arrays.stream(values())
+            .map(IotModbusFrameFormatEnum::getFormat)
+            .toArray(Integer[]::new);
+
+    /**
+     * 格式
+     */
+    private final Integer format;
+
+    @Override
+    public Integer[] array() {
+        return ARRAYS;
+    }
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/modbus/IotModbusModeEnum.java

@@ -0,0 +1,39 @@
+package cn.iocoder.yudao.module.iot.core.enums.modbus;
+
+import cn.iocoder.yudao.framework.common.core.ArrayValuable;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+import java.util.Arrays;
+
+/**
+ * IoT Modbus 工作模式枚举
+ *
+ * @author 芋道源码
+ */
+@Getter
+@RequiredArgsConstructor
+public enum IotModbusModeEnum implements ArrayValuable<Integer> {
+
+    POLLING(1, "云端轮询"),
+    ACTIVE_REPORT(2, "边缘采集");
+
+    public static final Integer[] ARRAYS = Arrays.stream(values())
+            .map(IotModbusModeEnum::getMode)
+            .toArray(Integer[]::new);
+
+    /**
+     * 工作模式
+     */
+    private final Integer mode;
+    /**
+     * 模式名称
+     */
+    private final String name;
+
+    @Override
+    public Integer[] array() {
+        return ARRAYS;
+    }
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/modbus/IotModbusRawDataTypeEnum.java

@@ -0,0 +1,56 @@
+package cn.iocoder.yudao.module.iot.core.enums.modbus;
+
+import cn.iocoder.yudao.framework.common.core.ArrayValuable;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+import java.util.Arrays;
+
+/**
+ * IoT Modbus 原始数据类型枚举
+ *
+ * @author 芋道源码
+ */
+@Getter
+@RequiredArgsConstructor
+public enum IotModbusRawDataTypeEnum implements ArrayValuable<String> {
+
+    INT16("INT16", "有符号 16 位整数", 1),
+    UINT16("UINT16", "无符号 16 位整数", 1),
+    INT32("INT32", "有符号 32 位整数", 2),
+    UINT32("UINT32", "无符号 32 位整数", 2),
+    FLOAT("FLOAT", "32 位浮点数", 2),
+    DOUBLE("DOUBLE", "64 位浮点数", 4),
+    BOOLEAN("BOOLEAN", "布尔值（用于线圈）", 1),
+    STRING("STRING", "字符串", null); // null 表示可变长度
+
+    public static final String[] ARRAYS = Arrays.stream(values())
+            .map(IotModbusRawDataTypeEnum::getType)
+            .toArray(String[]::new);
+
+    /**
+     * 数据类型
+     */
+    private final String type;
+    /**
+     * 名称
+     */
+    private final String name;
+    /**
+     * 寄存器数量（null 表示可变）
+     */
+    private final Integer registerCount;
+
+    @Override
+    public String[] array() {
+        return ARRAYS;
+    }
+
+    public static IotModbusRawDataTypeEnum getByType(String type) {
+        return Arrays.stream(values())
+                .filter(e -> e.getType().equals(type))
+                .findFirst()
+                .orElse(null);
+    }
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/config/IotMessageBusAutoConfiguration.java

@@ -0,0 +1,174 @@
+package cn.iocoder.yudao.module.iot.core.messagebus.config;
+
+import cn.iocoder.yudao.framework.mq.redis.core.RedisMQTemplate;
+import cn.iocoder.yudao.framework.mq.redis.core.job.RedisPendingMessageResendJob;
+import cn.iocoder.yudao.framework.mq.redis.core.job.RedisStreamMessageCleanupJob;
+import cn.iocoder.yudao.framework.mq.redis.core.stream.AbstractRedisStreamMessage;
+import cn.iocoder.yudao.framework.mq.redis.core.stream.AbstractRedisStreamMessageListener;
+import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;
+import cn.iocoder.yudao.module.iot.core.messagebus.core.kafka.IotKafkaMessageBus;
+import cn.iocoder.yudao.module.iot.core.messagebus.core.local.IotLocalMessageBus;
+import cn.iocoder.yudao.module.iot.core.messagebus.core.rabbitmq.IotRabbitMQMessageBus;
+import cn.iocoder.yudao.module.iot.core.messagebus.core.redis.IotRedisMessageBus;
+import cn.iocoder.yudao.module.iot.core.messagebus.core.rocketmq.IotRocketMQMessageBus;
+import cn.iocoder.yudao.module.iot.core.mq.producer.IotDeviceMessageProducer;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
+import org.apache.rocketmq.spring.core.RocketMQTemplate;
+import org.redisson.api.RedissonClient;
+import org.springframework.amqp.rabbit.core.RabbitAdmin;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.boot.autoconfigure.AutoConfiguration;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.boot.kafka.autoconfigure.KafkaProperties;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.kafka.core.KafkaTemplate;
+
+import java.util.List;
+
+import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertList;
+
+/**
+ * IoT 消息总线自动配置
+ *
+ * @author 芋道源码
+ */
+@AutoConfiguration
+@EnableConfigurationProperties(IotMessageBusProperties.class)
+@Slf4j
+public class IotMessageBusAutoConfiguration {
+
+    @Bean
+    public IotDeviceMessageProducer deviceMessageProducer(IotMessageBus messageBus) {
+        return new IotDeviceMessageProducer(messageBus);
+    }
+
+    // ==================== Local 实现 ====================
+
+    @Configuration
+    @ConditionalOnProperty(prefix = "yudao.iot.message-bus", name = "type", havingValue = "local", matchIfMissing = true)
+    public static class IotLocalMessageBusConfiguration {
+
+        @Bean
+        public IotLocalMessageBus iotLocalMessageBus(ApplicationContext applicationContext) {
+            log.info("[iotLocalMessageBus][创建 IoT Local 消息总线]");
+            return new IotLocalMessageBus(applicationContext);
+        }
+
+    }
+
+    // ==================== RocketMQ 实现 ====================
+
+    @Configuration
+    @ConditionalOnProperty(prefix = "yudao.iot.message-bus", name = "type", havingValue = "rocketmq")
+    @ConditionalOnClass(RocketMQTemplate.class)
+    public static class IotRocketMQMessageBusConfiguration {
+
+        @Bean
+        public IotRocketMQMessageBus iotRocketMQMessageBus(RocketMQProperties rocketMQProperties,
+                                                           RocketMQTemplate rocketMQTemplate) {
+            log.info("[iotRocketMQMessageBus][创建 IoT RocketMQ 消息总线]");
+            return new IotRocketMQMessageBus(rocketMQProperties, rocketMQTemplate);
+        }
+
+    }
+
+    // ==================== Kafka 实现 ====================
+
+    @Configuration
+    @ConditionalOnProperty(prefix = "yudao.iot.message-bus", name = "type", havingValue = "kafka")
+    @ConditionalOnClass(KafkaTemplate.class)
+    public static class IotKafkaMessageBusConfiguration {
+
+        @Bean
+        public IotKafkaMessageBus iotKafkaMessageBus(KafkaProperties kafkaProperties) {
+            log.info("[iotKafkaMessageBus][创建 IoT Kafka 消息总线]");
+            return new IotKafkaMessageBus(kafkaProperties);
+        }
+
+    }
+
+    // ==================== Redis 实现 ====================
+
+    /**
+     * 特殊：由于 YudaoRedisMQConsumerAutoConfiguration 关于 Redis stream 的消费是动态注册，所以这里只能拷贝相关的逻辑！！！
+     *
+     * @see cn.iocoder.yudao.framework.mq.redis.config.YudaoRedisMQConsumerAutoConfiguration
+     */
+    @Configuration
+    @ConditionalOnProperty(prefix = "yudao.iot.message-bus", name = "type", havingValue = "redis")
+    @ConditionalOnClass(RedisTemplate.class)
+    public static class IotRedisMessageBusConfiguration {
+
+        @Bean
+        public IotRedisMessageBus iotRedisMessageBus(StringRedisTemplate redisTemplate) {
+            log.info("[iotRedisMessageBus][创建 IoT Redis 消息总线]");
+            return new IotRedisMessageBus(redisTemplate);
+        }
+
+        /**
+         * 创建 Redis Stream 重新消费的任务
+         */
+        @Bean
+        public RedisPendingMessageResendJob iotRedisPendingMessageResendJob(IotRedisMessageBus messageBus,
+                                                                            RedisMQTemplate redisTemplate,
+                                                                            RedissonClient redissonClient) {
+            List<AbstractRedisStreamMessageListener<?>> listeners = getListeners(messageBus);
+            return new RedisPendingMessageResendJob(listeners, redisTemplate, redissonClient,
+                    RedisPendingMessageResendJob.IOT_RESEND_LOCK_KEY);
+        }
+
+        /**
+         * 创建 Redis Stream 消息清理任务
+         */
+        @Bean
+        public RedisStreamMessageCleanupJob iotRedisStreamMessageCleanupJob(IotRedisMessageBus messageBus,
+                                                                            RedisMQTemplate redisTemplate,
+                                                                            RedissonClient redissonClient) {
+            List<AbstractRedisStreamMessageListener<?>> listeners = getListeners(messageBus);
+            return new RedisStreamMessageCleanupJob(listeners, redisTemplate, redissonClient,
+                    RedisStreamMessageCleanupJob.IOT_CLEANUP_LOCK_KEY);
+        }
+
+        private List<AbstractRedisStreamMessageListener<?>> getListeners(IotRedisMessageBus messageBus) {
+            return convertList(messageBus.getSubscribers(), subscriber ->
+                    new AbstractRedisStreamMessageListener<>(subscriber.getTopic(), subscriber.getGroup()) {
+
+                        @Override
+                        public void onMessage(AbstractRedisStreamMessage message) {
+                            throw new UnsupportedOperationException("不应该调用！！！");
+                        }
+                    });
+        }
+
+    }
+
+    // ==================== RabbitMQ 实现 ====================
+
+    @Configuration
+    @ConditionalOnProperty(prefix = "yudao.iot.message-bus", name = "type", havingValue = "rabbitmq")
+    @ConditionalOnClass(RabbitTemplate.class)
+    public static class IotRabbitMQMessageBusConfiguration {
+
+        @Bean
+        @ConditionalOnMissingBean
+        public RabbitAdmin rabbitAdmin(RabbitTemplate rabbitTemplate) {
+            return new RabbitAdmin(rabbitTemplate);
+        }
+
+        @Bean
+        public IotRabbitMQMessageBus iotRabbitMQMessageBus(RabbitTemplate rabbitTemplate, RabbitAdmin rabbitAdmin) {
+            log.info("[iotRabbitMQMessageBus][创建 IoT RabbitMQ 消息总线]");
+            return new IotRabbitMQMessageBus(rabbitTemplate, rabbitAdmin);
+        }
+
+    }
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/config/IotMessageBusProperties.java

@@ -0,0 +1,28 @@
+package cn.iocoder.yudao.module.iot.core.messagebus.config;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.validation.annotation.Validated;
+
+import jakarta.validation.constraints.NotEmpty;
+import jakarta.validation.constraints.NotNull;
+
+/**
+ * IoT 消息总线配置属性
+ *
+ * @author 芋道源码
+ */
+@ConfigurationProperties("yudao.iot.message-bus")
+@Data
+@Validated
+public class IotMessageBusProperties {
+
+    /**
+     * 消息总线类型
+     *
+     * 可选值：local、redis、rocketmq、kafka、rabbitmq
+     */
+    @NotNull(message = "IoT 消息总线类型不能为空")
+    private String type = "local";
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/IotMessageBus.java

@@ -0,0 +1,37 @@
+package cn.iocoder.yudao.module.iot.core.messagebus.core;
+
+/**
+ * IoT 消息总线接口
+ *
+ * 用于在 IoT 系统中发布和订阅消息，支持多种消息中间件实现
+ *
+ * @author 芋道源码
+ */
+public interface IotMessageBus {
+
+    /**
+     * 发布消息到消息总线
+     *
+     * @param topic 主题
+     * @param message 消息内容
+     */
+    void post(String topic, Object message);
+
+    /**
+     * 注册消息订阅者
+     *
+     * @param subscriber 订阅者
+     */
+    void register(IotMessageSubscriber<?> subscriber);
+
+    /**
+     * 取消注册消息订阅者
+     *
+     * @param subscriber 订阅者
+     */
+    default void unregister(IotMessageSubscriber<?> subscriber) {
+        // TODO 芋艿：暂时不实现，需求量不大，但是
+        // throw new UnsupportedOperationException("取消注册消息订阅者功能，尚未实现");
+    }
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/IotMessageSubscriber.java

@@ -0,0 +1,41 @@
+package cn.iocoder.yudao.module.iot.core.messagebus.core;
+
+/**
+ * IoT 消息总线订阅者接口
+ *
+ * 用于处理从消息总线接收到的消息
+ *
+ * @author 芋道源码
+ */
+public interface IotMessageSubscriber<T> {
+
+    /**
+     * @return 主题
+     */
+    String getTopic();
+
+    /**
+     * @return 分组
+     */
+    String getGroup();
+
+    /**
+     * 处理接收到的消息
+     *
+     * @param message 消息内容
+     */
+    void onMessage(T message);
+
+    /**
+     * 启动订阅
+     */
+    default void start() {
+    }
+
+    /**
+     * 停止订阅
+     */
+    default void stop() {
+    }
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/kafka/IotKafkaMessageBus.java

@@ -0,0 +1,127 @@
+package cn.iocoder.yudao.module.iot.core.messagebus.core.kafka;
+
+import cn.hutool.core.util.TypeUtil;
+import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
+import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;
+import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageSubscriber;
+import jakarta.annotation.PreDestroy;
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.springframework.boot.kafka.autoconfigure.KafkaProperties;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.listener.AcknowledgingMessageListener;
+import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
+import org.springframework.kafka.listener.ContainerProperties;
+
+import java.lang.reflect.Type;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+
+/**
+ * 基于 Kafka 的 {@link IotMessageBus} 实现类
+ *
+ * @author 芋道源码
+ */
+@Slf4j
+public class IotKafkaMessageBus implements IotMessageBus {
+
+    private final KafkaTemplate<String, String> kafkaTemplate;
+
+    private final KafkaProperties kafkaProperties;
+
+    @Getter
+    private final List<IotMessageSubscriber<?>> subscribers = new ArrayList<>();
+
+    private final List<ConcurrentMessageListenerContainer<String, String>> containers = new ArrayList<>();
+
+    public IotKafkaMessageBus(KafkaProperties kafkaProperties) {
+        this.kafkaProperties = kafkaProperties;
+        this.kafkaTemplate = new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(buildProducerProperties(kafkaProperties)));
+    }
+
+    @Override
+    public void post(String topic, Object message) {
+        String messageJson = JsonUtils.toJsonString(message);
+        try {
+            kafkaTemplate.send(topic, messageJson).get();
+            log.info("[post][topic({}) 发送消息({})]", topic, message);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new IllegalStateException(String.format("发送 Kafka 消息失败，topic(%s) message(%s)", topic, message), e);
+        } catch (ExecutionException e) {
+            throw new IllegalStateException(String.format("发送 Kafka 消息失败，topic(%s) message(%s)", topic, message), e);
+        }
+    }
+
+    @Override
+    public void register(IotMessageSubscriber<?> subscriber) {
+        Type type = TypeUtil.getTypeArgument(subscriber.getClass(), 0);
+        if (type == null) {
+            throw new IllegalStateException(String.format("类型(%s) 需要设置消息类型", getClass().getName()));
+        }
+
+        // 1. 创建消费容器
+        ContainerProperties containerProperties = new ContainerProperties(subscriber.getTopic());
+        containerProperties.setGroupId(subscriber.getGroup());
+        containerProperties.setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
+        containerProperties.setMissingTopicsFatal(false);
+        containerProperties.setMessageListener((AcknowledgingMessageListener<String, String>) (message, acknowledgment) -> {
+            try {
+                subscriber.onMessage(JsonUtils.parseObject(message.value(), type));
+                acknowledgment.acknowledge();
+            } catch (Exception ex) {
+                log.error("[onMessage][topic({}/{}) message({}) 消费者({}) 处理异常]",
+                        subscriber.getTopic(), subscriber.getGroup(), message, subscriber.getClass().getName(), ex);
+                throw ex;
+            }
+        });
+        ConcurrentMessageListenerContainer<String, String> container = new ConcurrentMessageListenerContainer<>(
+                new DefaultKafkaConsumerFactory<>(buildConsumerProperties(kafkaProperties, subscriber.getGroup())),
+                containerProperties);
+        container.start();
+
+        // 2. 保存消费者引用
+        containers.add(container);
+        subscribers.add(subscriber);
+    }
+
+    @PreDestroy
+    public void destroy() {
+        for (ConcurrentMessageListenerContainer<String, String> container : containers) {
+            try {
+                container.stop();
+                log.info("[destroy][关闭 Kafka 消费者容器成功]");
+            } catch (Exception e) {
+                log.error("[destroy][关闭 Kafka 消费者容器异常]", e);
+            }
+        }
+        kafkaTemplate.destroy();
+    }
+
+    private static Map<String, Object> buildProducerProperties(KafkaProperties kafkaProperties) {
+        Map<String, Object> properties = new HashMap<>(kafkaProperties.buildProducerProperties());
+        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+        return properties;
+    }
+
+    private static Map<String, Object> buildConsumerProperties(KafkaProperties kafkaProperties, String group) {
+        Map<String, Object> properties = new HashMap<>(kafkaProperties.buildConsumerProperties());
+        properties.put(ConsumerConfig.GROUP_ID_CONFIG, group);
+        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
+        properties.putIfAbsent(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+        return properties;
+    }
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/IotLocalMessage.java

@@ -0,0 +1,14 @@
+package cn.iocoder.yudao.module.iot.core.messagebus.core.local;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+
+@Data
+@AllArgsConstructor
+public class IotLocalMessage {
+
+    private String topic;
+
+    private Object message;
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/IotLocalMessageBus.java

@@ -0,0 +1,67 @@
+package cn.iocoder.yudao.module.iot.core.messagebus.core.local;
+
+import cn.hutool.core.collection.CollUtil;
+import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;
+import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageSubscriber;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.event.EventListener;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * 本地的 {@link IotMessageBus} 实现类
+ *
+ * 注意：仅适用于单机场景！！！
+ *
+ * @author 芋道源码
+ */
+@RequiredArgsConstructor
+@Slf4j
+public class IotLocalMessageBus implements IotMessageBus {
+
+    private final ApplicationContext applicationContext;
+
+    /**
+     * 订阅者映射表
+     * Key: topic
+     */
+    private final Map<String, List<IotMessageSubscriber<?>>> subscribers = new HashMap<>();
+
+    @Override
+    public void post(String topic, Object message) {
+        applicationContext.publishEvent(new IotLocalMessage(topic, message));
+    }
+
+    @Override
+    public void register(IotMessageSubscriber<?> subscriber) {
+        String topic = subscriber.getTopic();
+        List<IotMessageSubscriber<?>> topicSubscribers = subscribers.computeIfAbsent(topic, k -> new ArrayList<>());
+        topicSubscribers.add(subscriber);
+        log.info("[register][topic({}/{}) 注册消费者({})成功]",
+                topic, subscriber.getGroup(), subscriber.getClass().getName());
+    }
+
+    @EventListener
+    @SuppressWarnings({"unchecked", "rawtypes"})
+    public void onMessage(IotLocalMessage message) {
+        String topic = message.getTopic();
+        List<IotMessageSubscriber<?>> topicSubscribers = subscribers.get(topic);
+        if (CollUtil.isEmpty(topicSubscribers)) {
+            return;
+        }
+        for (IotMessageSubscriber subscriber : topicSubscribers) {
+            try {
+                subscriber.onMessage(message.getMessage());
+            } catch (Exception ex) {
+                log.error("[onMessage][topic({}/{}) message({}) 消费者({}) 处理异常]",
+                        subscriber.getTopic(), subscriber.getGroup(), message.getMessage(), subscriber.getClass().getName(), ex);
+            }
+        }
+    }
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/rabbitmq/IotRabbitMQMessageBus.java

@@ -0,0 +1,104 @@
+package cn.iocoder.yudao.module.iot.core.messagebus.core.rabbitmq;
+
+import cn.hutool.core.util.TypeUtil;
+import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
+import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;
+import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageSubscriber;
+import jakarta.annotation.PreDestroy;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.amqp.core.AcknowledgeMode;
+import org.springframework.amqp.core.Binding;
+import org.springframework.amqp.core.BindingBuilder;
+import org.springframework.amqp.core.MessageBuilder;
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.core.TopicExchange;
+import org.springframework.amqp.rabbit.core.RabbitAdmin;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
+import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
+
+import java.lang.reflect.Type;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * 基于 RabbitMQ 的 {@link IotMessageBus} 实现类
+ *
+ * @author ywc
+ */
+@RequiredArgsConstructor
+@Slf4j
+public class IotRabbitMQMessageBus implements IotMessageBus {
+
+    private static final String ROUTING_KEY = "#";
+
+    private final RabbitTemplate rabbitTemplate;
+
+    private final RabbitAdmin rabbitAdmin;
+
+    @Getter
+    private final List<IotMessageSubscriber<?>> subscribers = new ArrayList<>();
+
+    private final List<SimpleMessageListenerContainer> containers = new ArrayList<>();
+
+    @Override
+    public void post(String topic, Object message) {
+        rabbitTemplate.send(topic, ROUTING_KEY, MessageBuilder.withBody(JsonUtils.toJsonByte(message)).build());
+        log.info("[post][topic({}) 发送消息({})]", topic, message);
+    }
+
+    @Override
+    @SuppressWarnings("DataFlowIssue")
+    public void register(IotMessageSubscriber<?> subscriber) {
+        Type type = TypeUtil.getTypeArgument(subscriber.getClass(), 0);
+        if (type == null) {
+            throw new IllegalStateException(String.format("类型(%s) 需要设置消息类型", getClass().getName()));
+        }
+
+        // 1.1 声明交换机、队列和绑定关系
+        Queue queue = new Queue(subscriber.getGroup(), true, false, false);
+        rabbitAdmin.declareQueue(queue);
+        TopicExchange exchange = new TopicExchange(subscriber.getTopic());
+        rabbitAdmin.declareExchange(exchange);
+        Binding binding = BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY);
+        rabbitAdmin.declareBinding(binding);
+
+        // 1.2 创建消费容器
+        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(rabbitTemplate.getConnectionFactory());
+        container.setQueues(queue);
+        container.setConcurrentConsumers(1);
+        container.setMaxConcurrentConsumers(10);
+        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
+        container.setMessageListener((ChannelAwareMessageListener) (message, channel) -> {
+            try {
+                subscriber.onMessage(JsonUtils.parseObject(message.getBody(), type));
+                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
+            } catch (Exception ex) {
+                log.error("[onMessage][topic({}/{}) message({}) 消费者({}) 处理异常]",
+                        subscriber.getTopic(), subscriber.getGroup(), message, subscriber.getClass().getName(), ex);
+                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
+            }
+        });
+        container.start();
+
+        // 2. 保存消费者引用
+        containers.add(container);
+        subscribers.add(subscriber);
+    }
+
+    @PreDestroy
+    public void destroy() {
+        for (SimpleMessageListenerContainer container : containers) {
+            try {
+                container.stop();
+                container.destroy();
+                log.info("[destroy][关闭 RabbitMQ 消费者容器成功]");
+            } catch (Exception e) {
+                log.error("[destroy][关闭 RabbitMQ 消费者容器异常]", e);
+            }
+        }
+    }
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/redis/IotRedisMessageBus.java

@@ -0,0 +1,99 @@
+package cn.iocoder.yudao.module.iot.core.messagebus.core.redis;
+
+import cn.hutool.core.util.TypeUtil;
+import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
+import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;
+import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageSubscriber;
+import jakarta.annotation.PostConstruct;
+import jakarta.annotation.PreDestroy;
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.data.redis.connection.stream.*;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.stream.StreamMessageListenerContainer;
+
+import java.lang.reflect.Type;
+import java.util.ArrayList;
+import java.util.List;
+
+import static cn.iocoder.yudao.framework.mq.redis.config.YudaoRedisMQConsumerAutoConfiguration.buildConsumerName;
+import static cn.iocoder.yudao.framework.mq.redis.config.YudaoRedisMQConsumerAutoConfiguration.checkRedisVersion;
+
+/**
+ * Redis 的 {@link IotMessageBus} 实现类
+ *
+ * @author 芋道源码
+ */
+@Slf4j
+public class IotRedisMessageBus implements IotMessageBus {
+
+    private final RedisTemplate<String, ?> redisTemplate;
+
+    private final StreamMessageListenerContainer<String, ObjectRecord<String, String>> redisStreamMessageListenerContainer;
+
+    @Getter
+    private final List<IotMessageSubscriber<?>> subscribers = new ArrayList<>();
+
+    public IotRedisMessageBus(RedisTemplate<String, ?> redisTemplate) {
+        this.redisTemplate = redisTemplate;
+        checkRedisVersion(redisTemplate);
+        // 创建 options 配置
+        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, ObjectRecord<String, String>> containerOptions =
+                StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()
+                        .batchSize(10) // 一次性最多拉取多少条消息
+                        .targetType(String.class) // 目标类型。统一使用 String，通过自己封装的 AbstractStreamMessageListener 去反序列化
+                        .build();
+        // 创建 container 对象
+        this.redisStreamMessageListenerContainer =
+                StreamMessageListenerContainer.create(redisTemplate.getRequiredConnectionFactory(), containerOptions);
+    }
+
+    @PostConstruct
+    public void init() {
+        this.redisStreamMessageListenerContainer.start();
+    }
+
+    @PreDestroy
+    public void destroy() {
+        this.redisStreamMessageListenerContainer.stop();
+    }
+
+    @Override
+    public void post(String topic, Object message) {
+        redisTemplate.opsForStream().add(StreamRecords.newRecord()
+                .ofObject(JsonUtils.toJsonString(message)) // 设置内容
+                .withStreamKey(topic)); // 设置 stream key
+    }
+
+    @Override
+    public void register(IotMessageSubscriber<?> subscriber) {
+        Type type = TypeUtil.getTypeArgument(subscriber.getClass(), 0);
+        if (type == null) {
+            throw new IllegalStateException(String.format("类型(%s) 需要设置消息类型", getClass().getName()));
+        }
+
+        // 创建 listener 对应的消费者分组
+        try {
+            redisTemplate.opsForStream().createGroup(subscriber.getTopic(), subscriber.getGroup());
+        } catch (Exception ignore) {
+        }
+        // 创建 Consumer 对象
+        String consumerName = buildConsumerName();
+        Consumer consumer = Consumer.from(subscriber.getGroup(), consumerName);
+        // 设置 Consumer 消费进度，以最小消费进度为准
+        StreamOffset<String> streamOffset = StreamOffset.create(subscriber.getTopic(), ReadOffset.lastConsumed());
+        // 设置 Consumer 监听
+        StreamMessageListenerContainer.StreamReadRequestBuilder<String> builder = StreamMessageListenerContainer.StreamReadRequest
+                .builder(streamOffset).consumer(consumer)
+                .autoAcknowledge(false) // 不自动 ack
+                .cancelOnError(throwable -> false); // 默认配置，发生异常就取消消费，显然不符合预期；因此，我们设置为 false
+        redisStreamMessageListenerContainer.register(builder.build(), message -> {
+            // 消费消息
+            subscriber.onMessage(JsonUtils.parseObject(message.getValue(), type));
+            // ack 消息消费完成
+            redisTemplate.opsForStream().acknowledge(subscriber.getGroup(), message);
+        });
+        this.subscribers.add(subscriber);
+    }
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/core/rocketmq/IotRocketMQMessageBus.java

@@ -0,0 +1,98 @@
+package cn.iocoder.yudao.module.iot.core.messagebus.core.rocketmq;
+
+import cn.hutool.core.util.TypeUtil;
+import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
+import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;
+import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageSubscriber;
+import jakarta.annotation.PreDestroy;
+import lombok.RequiredArgsConstructor;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
+import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
+import org.apache.rocketmq.spring.core.RocketMQTemplate;
+
+import java.lang.reflect.Type;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * 基于 RocketMQ 的 {@link IotMessageBus} 实现类
+ *
+ * @author 芋道源码
+ */
+@RequiredArgsConstructor
+@Slf4j
+public class IotRocketMQMessageBus implements IotMessageBus {
+
+    private final RocketMQProperties rocketMQProperties;
+
+    private final RocketMQTemplate rocketMQTemplate;
+
+    /**
+     * 主题对应的消费者映射
+     */
+    private final List<DefaultMQPushConsumer> topicConsumers = new ArrayList<>();
+
+    /**
+     * 销毁时关闭所有消费者
+     */
+    @PreDestroy
+    public void destroy() {
+        for (DefaultMQPushConsumer consumer : topicConsumers) {
+            try {
+                consumer.shutdown();
+                log.info("[destroy][关闭 group({}) 的消费者成功]", consumer.getConsumerGroup());
+            } catch (Exception e) {
+                log.error("[destroy]关闭 group({}) 的消费者异常]", consumer.getConsumerGroup(), e);
+            }
+        }
+    }
+
+    @Override
+    public void post(String topic, Object message) {
+        // TODO @芋艿：需要 orderly！
+        SendResult result = rocketMQTemplate.syncSend(topic, JsonUtils.toJsonString(message));
+        log.info("[post][topic({}) 发送消息({}) result({})]", topic, message, result);
+    }
+
+    @Override
+    @SneakyThrows
+    public void register(IotMessageSubscriber<?> subscriber) {
+        Type type = TypeUtil.getTypeArgument(subscriber.getClass(), 0);
+        if (type == null) {
+            throw new IllegalStateException(String.format("类型(%s) 需要设置消息类型", getClass().getName()));
+        }
+
+        // 1.1 创建 DefaultMQPushConsumer
+        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer();
+        consumer.setNamesrvAddr(rocketMQProperties.getNameServer());
+        consumer.setConsumerGroup(subscriber.getGroup());
+        // 1.2 订阅主题
+        consumer.subscribe(subscriber.getTopic(), "*");
+        // 1.3 设置消息监听器
+        consumer.setMessageListener((MessageListenerConcurrently) (messages, context) -> {
+            for (MessageExt messageExt : messages) {
+                try {
+                    byte[] body = messageExt.getBody();
+                    subscriber.onMessage(JsonUtils.parseObject(body, type));
+                } catch (Exception ex) {
+                    log.error("[onMessage][topic({}/{}) message({}) 消费者({}) 处理异常]",
+                            subscriber.getTopic(), subscriber.getGroup(), messageExt, subscriber.getClass().getName(), ex);
+                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
+                }
+            }
+            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
+        });
+        // 1.4 启动消费者
+        consumer.start();
+
+        // 2. 保存消费者引用
+        topicConsumers.add(consumer);
+    }
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/mq/message/IotDeviceMessage.java

@@ -0,0 +1,160 @@
+package cn.iocoder.yudao.module.iot.core.mq.message;
+
+import cn.iocoder.yudao.framework.common.exception.enums.GlobalErrorCodeConstants;
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
+import cn.iocoder.yudao.module.iot.core.enums.device.IotDeviceStateEnum;
+import cn.iocoder.yudao.module.iot.core.topic.state.IotDeviceStateUpdateReqDTO;
+import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.time.LocalDateTime;
+
+/**
+ * IoT 设备消息
+ */
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+@Builder
+public class IotDeviceMessage {
+
+    /**
+     * 【消息总线】应用的设备消息 Topic，由 iot-gateway 发给 iot-biz 进行消费
+     */
+    public static final String MESSAGE_BUS_DEVICE_MESSAGE_TOPIC = "iot_device_message";
+
+    /**
+     * 【消息总线】设备消息 Topic，由 iot-biz 发送给 iot-gateway 的某个 "server"(protocol) 进行消费
+     *
+     * 其中，%s 就是该"server"(protocol) 的标识
+     */
+    public static final String MESSAGE_BUS_GATEWAY_DEVICE_MESSAGE_TOPIC = MESSAGE_BUS_DEVICE_MESSAGE_TOPIC + "_%s";
+
+    /**
+     * 消息编号
+     *
+     * 由后端生成，通过 {@link IotDeviceMessageUtils#generateMessageId()}
+     */
+    private String id;
+    /**
+     * 上报时间
+     *
+     * 由后端生成，当前时间
+     */
+    private LocalDateTime reportTime;
+
+    /**
+     * 设备编号
+     */
+    private Long deviceId;
+    /**
+     * 租户编号
+     */
+    private Long tenantId;
+
+    /**
+     * 服务编号，该消息由哪个 server 发送
+     */
+    private String serverId;
+
+    // ========== serialize（序列化）相关字段 ==========
+
+    /**
+     * 请求编号
+     *
+     * 由设备生成，对应阿里云 IoT 的 Alink 协议中的 id、华为云 IoTDA 协议的 request_id
+     */
+    private String requestId;
+    /**
+     * 请求方法
+     *
+     * 枚举 {@link IotDeviceMessageMethodEnum}
+     * 例如说：thing.property.post 属性上报
+     */
+    private String method;
+    /**
+     * 请求参数
+     *
+     * 例如说：属性上报的 properties、事件上报的 params
+     */
+    private Object params;
+    /**
+     * 响应结果
+     */
+    private Object data;
+    /**
+     * 响应错误码
+     */
+    private Integer code;
+    /**
+     * 返回结果信息
+     */
+    private String msg;
+
+    // ========== 基础方法：只传递"serialize（序列化）相关字段" ==========
+
+    public static IotDeviceMessage requestOf(String method) {
+        return requestOf(null, method, null);
+    }
+
+    public static IotDeviceMessage requestOf(String method, Object params) {
+        return requestOf(null, method, params);
+    }
+
+    public static IotDeviceMessage requestOf(String requestId, String method, Object params) {
+        return of(requestId, method, params, null, null, null);
+    }
+
+    /**
+     * 创建设备请求消息（包含设备信息）
+     *
+     * @param deviceId 设备编号
+     * @param tenantId 租户编号
+     * @param serverId 服务标识
+     * @param method   消息方法
+     * @param params   消息参数
+     * @return 消息对象
+     */
+    public static IotDeviceMessage requestOf(Long deviceId, Long tenantId, String serverId,
+                                             String method, Object params) {
+        IotDeviceMessage message = of(null, method, params, null, null, null);
+        return message.setId(IotDeviceMessageUtils.generateMessageId())
+                .setDeviceId(deviceId).setTenantId(tenantId).setServerId(serverId);
+    }
+
+    public static IotDeviceMessage replyOf(String requestId, String method,
+                                           Object data, Integer code, String msg) {
+        if (code == null) {
+            code = GlobalErrorCodeConstants.SUCCESS.getCode();
+            msg = GlobalErrorCodeConstants.SUCCESS.getMsg();
+        }
+        return of(requestId, method, null, data, code, msg);
+    }
+
+    public static IotDeviceMessage of(String requestId, String method,
+                                      Object params, Object data, Integer code, String msg) {
+        // 通用参数
+        IotDeviceMessage message = new IotDeviceMessage()
+                .setId(IotDeviceMessageUtils.generateMessageId()).setReportTime(LocalDateTime.now());
+        // 当前参数
+        message.setRequestId(requestId).setMethod(method).setParams(params)
+                .setData(data).setCode(code).setMsg(msg);
+        return message;
+    }
+
+    // ========== 核心方法：在 of 基础方法之上，添加对应 method ==========
+
+    public static IotDeviceMessage buildStateUpdateOnline() {
+        return requestOf(IotDeviceMessageMethodEnum.STATE_UPDATE.getMethod(),
+                new IotDeviceStateUpdateReqDTO(IotDeviceStateEnum.ONLINE.getState()));
+    }
+
+    public static IotDeviceMessage buildStateOffline() {
+        return requestOf(IotDeviceMessageMethodEnum.STATE_UPDATE.getMethod(),
+                new IotDeviceStateUpdateReqDTO(IotDeviceStateEnum.OFFLINE.getState()));
+    }
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/mq/producer/IotDeviceMessageProducer.java

@@ -0,0 +1,37 @@
+package cn.iocoder.yudao.module.iot.core.mq.producer;
+
+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.core.util.IotDeviceMessageUtils;
+import lombok.RequiredArgsConstructor;
+
+/**
+ * IoT 设备消息生产者
+ *
+ * @author 芋道源码
+ */
+@RequiredArgsConstructor
+public class IotDeviceMessageProducer {
+
+    private final IotMessageBus messageBus;
+
+    /**
+     * 发送设备消息
+     *
+     * @param message 设备消息
+     */
+    public void sendDeviceMessage(IotDeviceMessage message) {
+        messageBus.post(IotDeviceMessage.MESSAGE_BUS_DEVICE_MESSAGE_TOPIC, message);
+    }
+
+    /**
+     * 发送网关设备消息
+     *
+     * @param serverId 网关的 serverId 标识
+     * @param message 设备消息
+     */
+    public void sendDeviceMessageToGateway(String serverId, IotDeviceMessage message) {
+        messageBus.post(IotDeviceMessageUtils.buildMessageBusGatewayDeviceMessageTopic(serverId), message);
+    }
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/IotDeviceIdentity.java

@@ -0,0 +1,32 @@
+package cn.iocoder.yudao.module.iot.core.topic;
+
+import jakarta.validation.constraints.NotEmpty;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * IoT 设备标识
+ *
+ * 用于标识一个设备的基本信息（productKey + deviceName）
+ *
+ * @author 芋道源码
+ */
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class IotDeviceIdentity {
+
+    /**
+     * 产品标识
+     */
+    @NotEmpty(message = "产品标识不能为空")
+    private String productKey;
+
+    /**
+     * 设备名称
+     */
+    @NotEmpty(message = "设备名称不能为空")
+    private String deviceName;
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/auth/IotDeviceRegisterReqDTO.java

@@ -0,0 +1,40 @@
+package cn.iocoder.yudao.module.iot.core.topic.auth;
+
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
+import jakarta.validation.constraints.NotEmpty;
+import lombok.Data;
+
+/**
+ * IoT 设备动态注册 Request DTO
+ * <p>
+ * 用于 {@link IotDeviceMessageMethodEnum#DEVICE_REGISTER} 消息的 params 参数
+ * <p>
+ * 直连设备/网关的一型一密动态注册：使用 productSecret 验证，返回 deviceSecret
+ *
+ * @author 芋道源码
+ * @see <a href="https://help.aliyun.com/zh/iot/user-guide/unique-certificate-per-product-verification">阿里云 - 一型一密</a>
+ */
+@Data
+public class IotDeviceRegisterReqDTO {
+
+    /**
+     * 产品标识
+     */
+    @NotEmpty(message = "产品标识不能为空")
+    private String productKey;
+
+    /**
+     * 设备名称
+     */
+    @NotEmpty(message = "设备名称不能为空")
+    private String deviceName;
+
+    /**
+     * 注册签名
+     *
+     * @see cn.iocoder.yudao.module.iot.core.util.IotProductAuthUtils#buildSign(String, String, String)
+     */
+    @NotEmpty(message = "签名不能为空")
+    private String sign;
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/auth/IotDeviceRegisterRespDTO.java

@@ -0,0 +1,36 @@
+package cn.iocoder.yudao.module.iot.core.topic.auth;
+
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * IoT 设备动态注册 Response DTO
+ * <p>
+ * 用于 {@link IotDeviceMessageMethodEnum#DEVICE_REGISTER} 响应的设备信息
+ *
+ * @author 芋道源码
+ * @see <a href="https://help.aliyun.com/zh/iot/user-guide/unique-certificate-per-product-verification">阿里云 - 一型一密</a>
+ */
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class IotDeviceRegisterRespDTO {
+
+    /**
+     * 产品标识
+     */
+    private String productKey;
+
+    /**
+     * 设备名称
+     */
+    private String deviceName;
+
+    /**
+     * 设备密钥
+     */
+    private String deviceSecret;
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/auth/IotSubDeviceRegisterReqDTO.java

@@ -0,0 +1,32 @@
+package cn.iocoder.yudao.module.iot.core.topic.auth;
+
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
+import jakarta.validation.constraints.NotEmpty;
+import lombok.Data;
+
+/**
+ * IoT 子设备动态注册 Request DTO
+ * <p>
+ * 用于 {@link IotDeviceMessageMethodEnum#SUB_DEVICE_REGISTER} 消息的 params 数组元素
+ * <p>
+ * 特殊：网关子设备的动态注册，必须已经创建好该网关子设备（不然哪来的 {@link #deviceName} 字段）。更多的好处，是设备不用提前烧录 deviceSecret 密钥。
+ *
+ * @author 芋道源码
+ * @see <a href="https://help.aliyun.com/zh/iot/user-guide/register-devices">阿里云 - 动态注册子设备</a>
+ */
+@Data
+public class IotSubDeviceRegisterReqDTO {
+
+    /**
+     * 子设备 ProductKey
+     */
+    @NotEmpty(message = "产品标识不能为空")
+    private String productKey;
+
+    /**
+     * 子设备 DeviceName
+     */
+    @NotEmpty(message = "设备名称不能为空")
+    private String deviceName;
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/auth/IotSubDeviceRegisterRespDTO.java

@@ -0,0 +1,36 @@
+package cn.iocoder.yudao.module.iot.core.topic.auth;
+
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * IoT 子设备动态注册 Response DTO
+ * <p>
+ * 用于 {@link IotDeviceMessageMethodEnum#SUB_DEVICE_REGISTER} 响应的设备信息
+ *
+ * @author 芋道源码
+ * @see <a href="https://help.aliyun.com/zh/iot/user-guide/register-devices">阿里云 - 动态注册子设备</a>
+ */
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class IotSubDeviceRegisterRespDTO {
+
+    /**
+     * 子设备 ProductKey
+     */
+    private String productKey;
+
+    /**
+     * 子设备 DeviceName
+     */
+    private String deviceName;
+
+    /**
+     * 分配的 DeviceSecret
+     */
+    private String deviceSecret;
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/config/IotDeviceConfigPushReqDTO.java

@@ -0,0 +1,54 @@
+package cn.iocoder.yudao.module.iot.core.topic.config;
+
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * IoT 设备配置推送 Request DTO
+ * <p>
+ * 用于 {@link IotDeviceMessageMethodEnum#CONFIG_PUSH} 下行消息的 params 参数
+ *
+ * @author 芋道源码
+ * @see <a href="https://help.aliyun.com/zh/iot/user-guide/remote-configuration-1">阿里云 - 远程配置</a>
+ */
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class IotDeviceConfigPushReqDTO {
+
+    /**
+     * 配置编号
+     */
+    private String configId;
+
+    /**
+     * 配置文件大小（字节）
+     */
+    private Long configSize;
+
+    /**
+     * 签名方法
+     */
+    private String signMethod;
+
+    /**
+     * 签名
+     */
+    private String sign;
+
+    /**
+     * 配置文件下载地址
+     */
+    private String url;
+
+    /**
+     * 获取类型
+     * <p>
+     * file: 文件
+     * content: 内容
+     */
+    private String getType;
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/event/IotDeviceEventPostReqDTO.java

@@ -0,0 +1,55 @@
+package cn.iocoder.yudao.module.iot.core.topic.event;
+
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
+import lombok.Data;
+
+/**
+ * IoT 设备事件上报 Request DTO
+ * <p>
+ * 用于 {@link IotDeviceMessageMethodEnum#EVENT_POST} 消息的 params 参数
+ *
+ * @author 芋道源码
+ * @see <a href="http://help.aliyun.com/zh/marketplace/device-reporting-events">阿里云 - 设备上报事件</a>
+ */
+@Data
+public class IotDeviceEventPostReqDTO {
+
+    /**
+     * 事件标识符
+     */
+    private String identifier;
+
+    /**
+     * 事件输出参数
+     */
+    private Object value;
+
+    /**
+     * 上报时间（毫秒时间戳，可选）
+     */
+    private Long time;
+
+    /**
+     * 创建事件上报 DTO
+     *
+     * @param identifier 事件标识符
+     * @param value   事件值
+     * @return DTO 对象
+     */
+    public static IotDeviceEventPostReqDTO of(String identifier, Object value) {
+        return of(identifier, value, null);
+    }
+
+    /**
+     * 创建事件上报 DTO（带时间）
+     *
+     * @param identifier 事件标识符
+     * @param value   事件值
+     * @param time    上报时间
+     * @return DTO 对象
+     */
+    public static IotDeviceEventPostReqDTO of(String identifier, Object value, Long time) {
+        return new IotDeviceEventPostReqDTO().setIdentifier(identifier).setValue(value).setTime(time);
+    }
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/ota/IotDeviceOtaProgressReqDTO.java

@@ -0,0 +1,41 @@
+package cn.iocoder.yudao.module.iot.core.topic.ota;
+
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * IoT 设备 OTA 升级进度上报 Request DTO
+ * <p>
+ * 用于 {@link IotDeviceMessageMethodEnum#OTA_PROGRESS} 上行消息的 params 参数
+ *
+ * @author 芋道源码
+ * @see <a href="https://help.aliyun.com/zh/iot/user-guide/perform-ota-updates">阿里云 - OTA 升级</a>
+ */
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class IotDeviceOtaProgressReqDTO {
+
+    /**
+     * 固件版本号
+     */
+    private String version;
+
+    /**
+     * 升级状态
+     */
+    private Integer status;
+
+    /**
+     * 描述信息
+     */
+    private String description;
+
+    /**
+     * 升级进度（0-100）
+     */
+    private Integer progress;
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/ota/IotDeviceOtaUpgradeReqDTO.java

@@ -0,0 +1,46 @@
+package cn.iocoder.yudao.module.iot.core.topic.ota;
+
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * IoT 设备 OTA 固件升级推送 Request DTO
+ * <p>
+ * 用于 {@link IotDeviceMessageMethodEnum#OTA_UPGRADE} 下行消息的 params 参数
+ *
+ * @author 芋道源码
+ * @see <a href="https://help.aliyun.com/zh/iot/user-guide/perform-ota-updates">阿里云 - OTA 升级</a>
+ */
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class IotDeviceOtaUpgradeReqDTO {
+
+    /**
+     * 固件版本号
+     */
+    private String version;
+
+    /**
+     * 固件文件下载地址
+     */
+    private String fileUrl;
+
+    /**
+     * 固件文件大小（字节）
+     */
+    private Long fileSize;
+
+    /**
+     * 固件文件摘要算法
+     */
+    private String fileDigestAlgorithm;
+
+    /**
+     * 固件文件摘要值
+     */
+    private String fileDigestValue;
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/package-info.java

@@ -0,0 +1,8 @@
+/**
+ * IoT Topic 消息体 DTO 定义
+ * <p>
+ * 定义设备与平台通信的消息体结构，遵循（参考）阿里云 Alink 协议规范
+ *
+ * @see <a href="https://help.aliyun.com/zh/iot/user-guide/alink-protocol-1">阿里云 Alink 协议</a>
+ */
+package cn.iocoder.yudao.module.iot.core.topic;

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/property/IotDevicePropertyPackPostReqDTO.java

@@ -0,0 +1,89 @@
+package cn.iocoder.yudao.module.iot.core.topic.property;
+
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
+import cn.iocoder.yudao.module.iot.core.topic.IotDeviceIdentity;
+import lombok.Data;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * IoT 设备属性批量上报 Request DTO
+ * <p>
+ * 用于 {@link IotDeviceMessageMethodEnum#PROPERTY_PACK_POST} 消息的 params 参数
+ *
+ * @author 芋道源码
+ * @see <a href="http://help.aliyun.com/zh/marketplace/gateway-reports-data-in-batches">阿里云 - 网关批量上报数据</a>
+ */
+@Data
+public class IotDevicePropertyPackPostReqDTO {
+
+    /**
+     * 网关自身属性
+     * <p>
+     * key: 属性标识符
+     * value: 属性值
+     */
+    private Map<String, Object> properties;
+
+    /**
+     * 网关自身事件
+     * <p>
+     * key: 事件标识符
+     * value: 事件值对象（包含 value 和 time）
+     */
+    private Map<String, EventValue> events;
+
+    /**
+     * 子设备数据列表
+     */
+    private List<SubDeviceData> subDevices;
+
+    /**
+     * 事件值对象
+     */
+    @Data
+    public static class EventValue {
+
+        /**
+         * 事件参数
+         */
+        private Object value;
+
+        /**
+         * 上报时间（毫秒时间戳）
+         */
+        private Long time;
+
+    }
+
+    /**
+     * 子设备数据
+     */
+    @Data
+    public static class SubDeviceData {
+
+        /**
+         * 子设备标识
+         */
+        private IotDeviceIdentity identity;
+
+        /**
+         * 子设备属性
+         * <p>
+         * key: 属性标识符
+         * value: 属性值
+         */
+        private Map<String, Object> properties;
+
+        /**
+         * 子设备事件
+         * <p>
+         * key: 事件标识符
+         * value: 事件值对象（包含 value 和 time）
+         */
+        private Map<String, EventValue> events;
+
+    }
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/property/IotDevicePropertyPostReqDTO.java

@@ -0,0 +1,38 @@
+package cn.iocoder.yudao.module.iot.core.topic.property;
+
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * IoT 设备属性上报 Request DTO
+ * <p>
+ * 用于 {@link IotDeviceMessageMethodEnum#PROPERTY_POST} 消息的 params 参数
+ * <p>
+ * 本质是一个 Map，key 为属性标识符，value 为属性值
+ *
+ * @author 芋道源码
+ * @see <a href="http://help.aliyun.com/zh/marketplace/device-reporting-attributes">阿里云 - 设备上报属性</a>
+ */
+public class IotDevicePropertyPostReqDTO extends HashMap<String, Object> {
+
+    public IotDevicePropertyPostReqDTO() {
+        super();
+    }
+
+    public IotDevicePropertyPostReqDTO(Map<String, Object> properties) {
+        super(properties);
+    }
+
+    /**
+     * 创建属性上报 DTO
+     *
+     * @param properties 属性数据
+     * @return DTO 对象
+     */
+    public static IotDevicePropertyPostReqDTO of(Map<String, Object> properties) {
+        return new IotDevicePropertyPostReqDTO(properties);
+    }
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/property/IotDevicePropertySetReqDTO.java

@@ -0,0 +1,37 @@
+package cn.iocoder.yudao.module.iot.core.topic.property;
+
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * IoT 设备属性设置 Request DTO
+ * <p>
+ * 用于 {@link IotDeviceMessageMethodEnum#PROPERTY_SET} 下行消息的 params 参数
+ * <p>
+ * 本质是一个 Map，key 为属性标识符，value 为属性值
+ *
+ * @author 芋道源码
+ */
+public class IotDevicePropertySetReqDTO extends HashMap<String, Object> {
+
+    public IotDevicePropertySetReqDTO() {
+        super();
+    }
+
+    public IotDevicePropertySetReqDTO(Map<String, Object> properties) {
+        super(properties);
+    }
+
+    /**
+     * 创建属性设置 DTO
+     *
+     * @param properties 属性数据
+     * @return DTO 对象
+     */
+    public static IotDevicePropertySetReqDTO of(Map<String, Object> properties) {
+        return new IotDevicePropertySetReqDTO(properties);
+    }
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/service/IotDeviceServiceInvokeReqDTO.java

@@ -0,0 +1,32 @@
+package cn.iocoder.yudao.module.iot.core.topic.service;
+
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.Map;
+
+/**
+ * IoT 设备服务调用 Request DTO
+ * <p>
+ * 用于 {@link IotDeviceMessageMethodEnum#SERVICE_INVOKE} 下行消息的 params 参数
+ *
+ * @author 芋道源码
+ */
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class IotDeviceServiceInvokeReqDTO {
+
+    /**
+     * 服务标识符
+     */
+    private String identifier;
+
+    /**
+     * 服务输入参数
+     */
+    private Map<String, Object> inputParams;
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/state/IotDeviceStateUpdateReqDTO.java

@@ -0,0 +1,25 @@
+package cn.iocoder.yudao.module.iot.core.topic.state;
+
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * IoT 设备状态更新 Request DTO
+ * <p>
+ * 用于 {@link IotDeviceMessageMethodEnum#STATE_UPDATE} 消息的 params 参数
+ *
+ * @author 芋道源码
+ */
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class IotDeviceStateUpdateReqDTO {
+
+    /**
+     * 设备状态
+     */
+    private Integer state;
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/topo/IotDeviceTopoAddReqDTO.java

@@ -0,0 +1,29 @@
+package cn.iocoder.yudao.module.iot.core.topic.topo;
+
+import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceAuthReqDTO;
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
+import jakarta.validation.constraints.NotEmpty;
+import lombok.Data;
+
+import java.util.List;
+
+/**
+ * IoT 设备拓扑添加 Request DTO
+ * <p>
+ * 用于 {@link IotDeviceMessageMethodEnum#TOPO_ADD} 消息的 params 参数
+ *
+ * @author 芋道源码
+ * @see <a href="http://help.aliyun.com/zh/marketplace/add-topological-relationship">阿里云 - 添加拓扑关系</a>
+ */
+@Data
+public class IotDeviceTopoAddReqDTO {
+
+    /**
+     * 子设备认证信息列表
+     * <p>
+     * 复用 {@link IotDeviceAuthReqDTO}，包含 clientId、username、password
+     */
+    @NotEmpty(message = "子设备认证信息列表不能为空")
+    private List<IotDeviceAuthReqDTO> subDevices;
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/topo/IotDeviceTopoChangeReqDTO.java

@@ -0,0 +1,45 @@
+package cn.iocoder.yudao.module.iot.core.topic.topo;
+
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
+import cn.iocoder.yudao.module.iot.core.topic.IotDeviceIdentity;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.List;
+
+/**
+ * IoT 设备拓扑关系变更通知 Request DTO
+ * <p>
+ * 用于 {@link IotDeviceMessageMethodEnum#TOPO_CHANGE} 下行消息的 params 参数
+ *
+ * @author 芋道源码
+ * @see <a href="https://help.aliyun.com/zh/marketplace/notify-gateway-topology-changes">阿里云 - 通知网关拓扑关系变化</a>
+ */
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class IotDeviceTopoChangeReqDTO {
+
+    public static final Integer STATUS_CREATE = 0;
+    public static final Integer STATUS_DELETE = 1;
+
+    /**
+     * 拓扑关系状态
+     */
+    private Integer status;
+
+    /**
+     * 子设备列表
+     */
+    private List<IotDeviceIdentity> subList;
+
+    public static IotDeviceTopoChangeReqDTO ofCreate(List<IotDeviceIdentity> subList) {
+        return new IotDeviceTopoChangeReqDTO(STATUS_CREATE, subList);
+    }
+
+    public static IotDeviceTopoChangeReqDTO ofDelete(List<IotDeviceIdentity> subList) {
+        return new IotDeviceTopoChangeReqDTO(STATUS_DELETE, subList);
+    }
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/topo/IotDeviceTopoDeleteReqDTO.java

@@ -0,0 +1,29 @@
+package cn.iocoder.yudao.module.iot.core.topic.topo;
+
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
+import cn.iocoder.yudao.module.iot.core.topic.IotDeviceIdentity;
+import jakarta.validation.Valid;
+import jakarta.validation.constraints.NotEmpty;
+import lombok.Data;
+
+import java.util.List;
+
+/**
+ * IoT 设备拓扑删除 Request DTO
+ * <p>
+ * 用于 {@link IotDeviceMessageMethodEnum#TOPO_DELETE} 消息的 params 参数
+ *
+ * @author 芋道源码
+ * @see <a href="https://help.aliyun.com/zh/marketplace/delete-a-topological-relationship">阿里云 - 删除拓扑关系</a>
+ */
+@Data
+public class IotDeviceTopoDeleteReqDTO {
+
+    /**
+     * 子设备标识列表
+     */
+    @Valid
+    @NotEmpty(message = "子设备标识列表不能为空")
+    private List<IotDeviceIdentity> subDevices;
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/topo/IotDeviceTopoGetReqDTO.java

@@ -0,0 +1,17 @@
+package cn.iocoder.yudao.module.iot.core.topic.topo;
+
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
+import lombok.Data;
+
+/**
+ * IoT 设备拓扑关系获取 Request DTO
+ * <p>
+ * 用于 {@link IotDeviceMessageMethodEnum#TOPO_GET} 请求的 params 参数（目前为空，预留扩展）
+ *
+ * @author 芋道源码
+ * @see <a href="https://help.aliyun.com/zh/marketplace/obtain-topological-relationship">阿里云 - 获取拓扑关系</a>
+ */
+@Data
+public class IotDeviceTopoGetReqDTO {
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/topic/topo/IotDeviceTopoGetRespDTO.java

@@ -0,0 +1,25 @@
+package cn.iocoder.yudao.module.iot.core.topic.topo;
+
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
+import cn.iocoder.yudao.module.iot.core.topic.IotDeviceIdentity;
+import lombok.Data;
+
+import java.util.List;
+
+/**
+ * IoT 设备拓扑关系获取 Response DTO
+ * <p>
+ * 用于 {@link IotDeviceMessageMethodEnum#TOPO_GET} 响应
+ *
+ * @author 芋道源码
+ * @see <a href="https://help.aliyun.com/zh/marketplace/obtain-topological-relationship">阿里云 - 获取拓扑关系</a>
+ */
+@Data
+public class IotDeviceTopoGetRespDTO {
+
+    /**
+     * 子设备列表
+     */
+    private List<IotDeviceIdentity> subDevices;
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/util/IotDeviceAuthUtils.java

@@ -0,0 +1,60 @@
+package cn.iocoder.yudao.module.iot.core.util;
+
+import cn.hutool.core.util.StrUtil;
+import cn.hutool.crypto.digest.DigestUtil;
+import cn.hutool.crypto.digest.HmacAlgorithm;
+import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceAuthReqDTO;
+import cn.iocoder.yudao.module.iot.core.topic.IotDeviceIdentity;
+
+/**
+ * IoT 设备【认证】的工具类，参考阿里云
+ *
+ * @see <a href="https://help.aliyun.com/zh/iot/user-guide/how-do-i-obtain-mqtt-parameters-for-authentication">如何计算 MQTT 签名参数</a>
+ */
+public class IotDeviceAuthUtils {
+
+    public static IotDeviceAuthReqDTO getAuthInfo(String productKey, String deviceName, String deviceSecret) {
+        String clientId = buildClientId(productKey, deviceName);
+        String username = buildUsername(productKey, deviceName);
+        String password = buildPassword(deviceSecret,
+                buildContent(clientId, productKey, deviceName, deviceSecret));
+        return new IotDeviceAuthReqDTO(clientId, username, password);
+    }
+
+    public static String buildClientId(String productKey, String deviceName) {
+        return String.format("%s.%s", productKey, deviceName);
+    }
+
+    public static String buildClientIdFromUsername(String username) {
+        IotDeviceIdentity identity = parseUsername(username);
+        if (identity == null) {
+            return null;
+        }
+        return buildClientId(identity.getProductKey(), identity.getDeviceName());
+    }
+
+    public static String buildUsername(String productKey, String deviceName) {
+        return String.format("%s&%s", deviceName, productKey);
+    }
+
+    public static String buildPassword(String deviceSecret, String content) {
+        return DigestUtil.hmac(HmacAlgorithm.HmacSHA256, StrUtil.utf8Bytes(deviceSecret))
+                .digestHex(content);
+    }
+
+    private static String buildContent(String clientId, String productKey, String deviceName, String deviceSecret) {
+        return "clientId" + clientId +
+                "deviceName" + deviceName +
+                "deviceSecret" + deviceSecret +
+                "productKey" + productKey;
+    }
+
+    public static IotDeviceIdentity parseUsername(String username) {
+        String[] usernameParts = username.split("&");
+        if (usernameParts.length != 2) {
+            return null;
+        }
+        return new IotDeviceIdentity(usernameParts[1], usernameParts[0]);
+    }
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/util/IotDeviceMessageUtils.java

@@ -0,0 +1,319 @@
+package cn.iocoder.yudao.module.iot.core.util;
+
+import cn.hutool.core.lang.Assert;
+import cn.hutool.core.util.ClassUtil;
+import cn.hutool.core.util.IdUtil;
+import cn.hutool.core.util.ReflectUtil;
+import cn.hutool.core.util.StrUtil;
+import cn.hutool.system.SystemUtil;
+import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
+import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
+import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
+
+import java.util.LinkedHashSet;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * IoT 设备【消息】的工具类
+ *
+ * @author 芋道源码
+ */
+public class IotDeviceMessageUtils {
+
+    // ========== Message 相关 ==========
+
+    public static String generateMessageId() {
+        return IdUtil.fastSimpleUUID();
+    }
+
+    /**
+     * 是否是上行消息：由设备发送
+     *
+     * @param message 消息
+     * @return 是否
+     */
+    @SuppressWarnings("SimplifiableConditionalExpression")
+    public static boolean isUpstreamMessage(IotDeviceMessage message) {
+        IotDeviceMessageMethodEnum methodEnum = IotDeviceMessageMethodEnum.of(message.getMethod());
+        Assert.notNull(methodEnum, "无法识别的消息方法：" + message.getMethod());
+        // 注意：回复消息时，需要取反
+        return !isReplyMessage(message) ? methodEnum.getUpstream() : !methodEnum.getUpstream();
+    }
+
+    /**
+     * 是否是回复消息，通过 {@link IotDeviceMessage#getCode()} 非空进行识别
+     *
+     * @param message 消息
+     * @return 是否
+     */
+    public static boolean isReplyMessage(IotDeviceMessage message) {
+        return message.getCode() != null;
+    }
+
+    /**
+     * 提取消息中的标识符
+     *
+     * @param message 消息
+     * @return 标识符
+     */
+    public static String getIdentifier(IotDeviceMessage message) {
+        if (message == null || message.getParams() == null) {
+            return null;
+        }
+        Object params = message.getParams();
+        if (StrUtil.equalsAny(message.getMethod(), IotDeviceMessageMethodEnum.EVENT_POST.getMethod(),
+                IotDeviceMessageMethodEnum.SERVICE_INVOKE.getMethod())) {
+            return StrUtil.toStringOrNull(readField(params, "identifier"));
+        } else if (StrUtil.equalsAny(message.getMethod(), IotDeviceMessageMethodEnum.STATE_UPDATE.getMethod())) {
+            return StrUtil.toStringOrNull(readField(params, "state"));
+        }
+        return null;
+    }
+
+    /**
+     * 从 params 中读取字段值，兼容 Map 和 POJO（DTO）两种形态
+     *
+     * Why：MQ 消息经 JSON 反序列化后 params 是 Map，但本地总线场景 producer 可能直接传 DTO 对象（如 IotDeviceEventPostReqDTO），
+     * matcher 必须同时支持两种形态，避免事件触发器在同 JVM 内部消息总线下匹配失败
+     */
+    private static Object readField(Object params, String fieldName) {
+        if (params == null) {
+            return null;
+        }
+        if (params instanceof Map) {
+            return ((Map<?, ?>) params).get(fieldName);
+        }
+        // 跳过 JDK 内置类型，避免反射读取到内部字段（例如 JDK8 下 String#value 会返回 char[]）
+        if (ClassUtil.isJdkClass(params.getClass())) {
+            return null;
+        }
+        try {
+            return ReflectUtil.getFieldValue(params, fieldName);
+        } catch (Exception ignored) {
+            return null;
+        }
+    }
+
+    /**
+     * 获取属性上报消息中包含的所有属性标识符
+     *
+     * 仅支持扁平结构：{ temperature: 25.5, humidity: 60 }，顶层 key 即属性标识符
+     *
+     * @param message 设备消息
+     * @return 属性标识符集合，不为 null
+     */
+    public static Set<String> getPropertyIdentifiers(IotDeviceMessage message) {
+        if (message == null) {
+            return new LinkedHashSet<>();
+        }
+        Map<String, Object> params = parseParamsToMap(message.getParams());
+        if (params == null) {
+            return new LinkedHashSet<>();
+        }
+        return new LinkedHashSet<>(params.keySet());
+    }
+
+    /**
+     * 判断消息中是否包含指定的标识符
+     * <p>
+     * 对于不同消息类型的处理：
+     * - EVENT_POST/SERVICE_INVOKE：检查 params.identifier 是否匹配
+     * - STATE_UPDATE：检查 params.state 是否匹配
+     * - PROPERTY_POST：检查 params 中是否包含该属性 key
+     *
+     * @param message    消息
+     * @param identifier 要检查的标识符
+     * @return 是否包含
+     */
+    @SuppressWarnings("unchecked")
+    public static boolean containsIdentifier(IotDeviceMessage message, String identifier) {
+        if (message == null || message.getParams() == null || StrUtil.isBlank(identifier)) {
+            return false;
+        }
+        // EVENT_POST / SERVICE_INVOKE / STATE_UPDATE：使用原有逻辑
+        String messageIdentifier = getIdentifier(message);
+        if (messageIdentifier != null) {
+            return identifier.equals(messageIdentifier);
+        }
+        // PROPERTY_POST：检查 params 中是否包含该属性 key（支持扁平和嵌套 properties 结构）
+        if (StrUtil.equals(message.getMethod(), IotDeviceMessageMethodEnum.PROPERTY_POST.getMethod())) {
+            Map<String, Object> params = parseParamsToMap(message.getParams());
+            if (params == null) {
+                return false;
+            }
+            if (params.containsKey(identifier)) {
+                return true;
+            }
+            Object properties = params.get("properties");
+            return properties instanceof Map && ((Map<String, Object>) properties).containsKey(identifier);
+        }
+        return false;
+    }
+
+    /**
+     * 判断消息中是否不包含指定的标识符
+     *
+     * @param message    消息
+     * @param identifier 要检查的标识符
+     * @return 是否不包含
+     */
+    public static boolean notContainsIdentifier(IotDeviceMessage message, String identifier) {
+        return !containsIdentifier(message, identifier);
+    }
+
+    /**
+     * 将 params 解析为 Map
+     *
+     * @param params 参数（可能是 Map 或 JSON 字符串）
+     * @return Map，解析失败返回 null
+     */
+    @SuppressWarnings("unchecked")
+    private static Map<String, Object> parseParamsToMap(Object params) {
+        if (params instanceof Map) {
+            return (Map<String, Object>) params;
+        }
+        if (params instanceof String) {
+            try {
+                return JsonUtils.parseObject((String) params, Map.class);
+            } catch (Exception ignored) {
+            }
+        }
+        return null;
+    }
+
+    /**
+     * 从设备消息中提取指定标识符的属性值
+     * <p>
+     * 支持的提取策略（按优先级顺序）：
+     * 1. 直接值：如果 params 不是 Map，直接返回该值（适用于简单消息）
+     * 2. 标识符字段：从 params[identifier] 获取
+     * 3. properties 结构：从 params.properties[identifier] 获取（标准属性上报）
+     * 4. data 结构：从 params.data[identifier] 获取
+     * 5. value 字段：从 params.value 获取（单值消息）
+     * 6. 单值 Map：如果 Map 只包含 identifier 和一个值，返回该值
+     *
+     * @param message    设备消息
+     * @param identifier 属性标识符
+     * @return 属性值，如果未找到则返回 null
+     */
+    @SuppressWarnings("unchecked")
+    public static Object extractPropertyValue(IotDeviceMessage message, String identifier) {
+        Object params = message != null ? message.getParams() : null;
+        if (params == null) {
+            return null;
+        }
+
+        // 策略 1：如果 params 不是 Map，直接返回该值（适用于简单的单属性消息）
+        if (!(params instanceof Map)) {
+            return params;
+        }
+
+        // 策略 2：直接通过标识符获取属性值
+        Map<String, Object> paramsMap = (Map<String, Object>) params;
+        Object directValue = paramsMap.get(identifier);
+        if (directValue != null) {
+            return directValue;
+        }
+
+        // 策略 3：从 properties 字段中获取（适用于标准属性上报消息）
+        Object properties = paramsMap.get("properties");
+        if (properties instanceof Map) {
+            Map<String, Object> propertiesMap = (Map<String, Object>) properties;
+            Object propertyValue = propertiesMap.get(identifier);
+            if (propertyValue != null) {
+                return propertyValue;
+            }
+        }
+
+        // 策略 4：从 data 字段中获取（适用于某些消息格式）
+        Object data = paramsMap.get("data");
+        if (data instanceof Map) {
+            Map<String, Object> dataMap = (Map<String, Object>) data;
+            Object dataValue = dataMap.get(identifier);
+            if (dataValue != null) {
+                return dataValue;
+            }
+        }
+
+        // 策略 5：从 value 字段中获取（适用于单值消息）
+        Object value = paramsMap.get("value");
+        if (value != null) {
+            return value;
+        }
+
+        // 策略 6：如果 Map 只有两个字段且包含 identifier，返回另一个字段的值
+        if (paramsMap.size() == 2 && paramsMap.containsKey("identifier")) {
+            for (Map.Entry<String, Object> entry : paramsMap.entrySet()) {
+                if (!"identifier".equals(entry.getKey())) {
+                    return entry.getValue();
+                }
+            }
+        }
+
+        // 未找到对应的属性值
+        return null;
+    }
+
+    /**
+     * 从设备事件上报消息中提取事件值
+     * <p>
+     * 事件上报的 params 结构为：{"identifier": "xxx", "value": ...}，事件值即 value 字段。
+     * value 可能是标量（字符串/数字/布尔），也可能是结构体（如告警事件 {level, message}）
+     *
+     * @param message 设备消息
+     * @return 事件值，如果未找到则返回 null
+     */
+    public static Object extractEventValue(IotDeviceMessage message) {
+        return readField(message != null ? message.getParams() : null, "value");
+    }
+
+    /**
+     * 从服务调用消息中提取输入参数
+     * <p>
+     * 服务调用消息的 params 结构通常为：
+     * {
+     *     "identifier": "serviceIdentifier",
+     *     "inputData": { ... } 或 "inputParams": { ... }
+     * }
+     *
+     * @param message 设备消息
+     * @return 输入参数 Map，如果未找到则返回 null
+     */
+    @SuppressWarnings("unchecked")
+    public static Map<String, Object> extractServiceInputParams(IotDeviceMessage message) {
+        if (message == null || message.getParams() == null) {
+            return null;
+        }
+        Object params = message.getParams();
+        // 兼容 Map 和 POJO（如 IotDeviceServiceInvokeReqDTO）两种 params 形态
+        Object inputData = readField(params, "inputData");
+        if (inputData instanceof Map) {
+            return (Map<String, Object>) inputData;
+        }
+        Object inputParams = readField(params, "inputParams");
+        if (inputParams instanceof Map) {
+            return (Map<String, Object>) inputParams;
+        }
+        return null;
+    }
+
+    // ========== Topic 相关 ==========
+
+    public static String buildMessageBusGatewayDeviceMessageTopic(String serverId) {
+        return String.format(IotDeviceMessage.MESSAGE_BUS_GATEWAY_DEVICE_MESSAGE_TOPIC, serverId);
+    }
+
+    /**
+     * 生成服务器编号
+     *
+     * @param serverPort 服务器端口
+     * @return 服务器编号
+     */
+    public static String generateServerId(Integer serverPort) {
+        String serverId = String.format("%s.%d", SystemUtil.getHostInfo().getAddress(), serverPort);
+        // 避免一些场景无法使用 . 符号，例如说 RocketMQ Topic
+        return serverId.replaceAll("\\.", "_");
+    }
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/util/IotProductAuthUtils.java

@@ -0,0 +1,55 @@
+package cn.iocoder.yudao.module.iot.core.util;
+
+import cn.hutool.core.util.StrUtil;
+import cn.hutool.crypto.digest.DigestUtil;
+import cn.hutool.crypto.digest.HmacAlgorithm;
+
+/**
+ * IoT 产品【动态注册】认证工具类
+ * <p>
+ * 用于一型一密场景，使用 productSecret 生成签名
+ *
+ * @author 芋道源码
+ */
+public class IotProductAuthUtils {
+
+    /**
+     * 生成设备动态注册签名
+     *
+     * @param productKey    产品标识
+     * @param deviceName    设备名称
+     * @param productSecret 产品密钥
+     * @return 签名
+     */
+    public static String buildSign(String productKey, String deviceName, String productSecret) {
+        String content = buildContent(productKey, deviceName);
+        return DigestUtil.hmac(HmacAlgorithm.HmacSHA256, StrUtil.utf8Bytes(productSecret))
+                .digestHex(content);
+    }
+
+    /**
+     * 验证设备动态注册签名
+     *
+     * @param productKey    产品标识
+     * @param deviceName    设备名称
+     * @param productSecret 产品密钥
+     * @param sign          待验证的签名
+     * @return 是否验证通过
+     */
+    public static boolean verifySign(String productKey, String deviceName, String productSecret, String sign) {
+        String expectedSign = buildSign(productKey, deviceName, productSecret);
+        return expectedSign.equals(sign);
+    }
+
+    /**
+     * 构建签名内容
+     *
+     * @param productKey 产品标识
+     * @param deviceName 设备名称
+     * @return 签名内容
+     */
+    private static String buildContent(String productKey, String deviceName) {
+        return "deviceName" + deviceName + "productKey" + productKey;
+    }
+
+}

yudao-module-iot/yudao-module-iot-core/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports

@@ -0,0 +1 @@
+cn.iocoder.yudao.module.iot.core.messagebus.config.IotMessageBusAutoConfiguration

yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/messagebus/core/TestMessage.java

@@ -0,0 +1,12 @@
+package cn.iocoder.yudao.module.iot.core.messagebus.core;
+
+import lombok.Data;
+
+@Data
+public class TestMessage {
+
+    private String nickname;
+
+    private Integer age;
+
+}

yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/messagebus/core/local/LocalIotMessageBusIntegrationTest.java

@@ -0,0 +1,177 @@
+package cn.iocoder.yudao.module.iot.core.messagebus.core.local;
+
+import cn.iocoder.yudao.module.iot.core.messagebus.config.IotMessageBusAutoConfiguration;
+import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;
+import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageSubscriber;
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.context.annotation.Import;
+import org.springframework.test.context.TestPropertySource;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * {@link IotLocalMessageBus} 集成测试
+ *
+ * @author 芋道源码
+ */
+@SpringBootTest(classes = LocalIotMessageBusIntegrationTest.class)
+@Import(IotMessageBusAutoConfiguration.class)
+@TestPropertySource(properties = {
+    "yudao.iot.message-bus.type=local"
+})
+@Slf4j
+public class LocalIotMessageBusIntegrationTest {
+
+    @Resource
+    private IotMessageBus messageBus;
+
+    /**
+     * 1 topic 2 subscriber
+     */
+    @Test
+    public void testSendMessageWithTwoSubscribers() throws InterruptedException {
+        // 准备
+        String topic = "test-topic";
+        String testMessage = "Hello IoT Message Bus!";
+        // 用于等待消息处理完成
+        CountDownLatch latch = new CountDownLatch(2);
+        // 用于记录接收到的消息
+        AtomicInteger subscriber1Count = new AtomicInteger(0);
+        AtomicInteger subscriber2Count = new AtomicInteger(0);
+
+        // 创建第一个订阅者
+        IotMessageSubscriber<String> subscriber1 = new IotMessageSubscriber<>() {
+
+            @Override
+            public String getTopic() {
+                return topic;
+            }
+
+            @Override
+            public String getGroup() {
+                return "group1";
+            }
+
+            @Override
+            public void onMessage(String message) {
+                log.info("[订阅者1] 收到消息 - Topic: {}, Message: {}", getTopic(), message);
+                subscriber1Count.incrementAndGet();
+                assertEquals(testMessage, message);
+                latch.countDown();
+            }
+
+        };
+        // 创建第二个订阅者
+        IotMessageSubscriber<String> subscriber2 = new IotMessageSubscriber<>() {
+
+            @Override
+            public String getTopic() {
+                return topic;
+            }
+
+            @Override
+            public String getGroup() {
+                return "group2";
+            }
+
+            @Override
+            public void onMessage(String message) {
+                log.info("[订阅者2] 收到消息 - Topic: {}, Message: {}", getTopic(), message);
+                subscriber2Count.incrementAndGet();
+                assertEquals(testMessage, message);
+                latch.countDown();
+            }
+
+        };
+        // 注册订阅者
+        messageBus.register(subscriber1);
+        messageBus.register(subscriber2);
+
+        // 发送消息
+        log.info("[测试] 发送消息 - Topic: {}, Message: {}", topic, testMessage);
+        messageBus.post(topic, testMessage);
+        // 等待消息处理完成（最多等待 10 秒）
+        boolean completed = latch.await(10, TimeUnit.SECONDS);
+
+        // 验证结果
+        assertTrue(completed, "消息处理超时");
+        assertEquals(1, subscriber1Count.get(), "订阅者 1 应该收到 1 条消息");
+        assertEquals(1, subscriber2Count.get(), "订阅者 2 应该收到 1 条消息");
+        log.info("[测试] 测试完成 - 订阅者 1 收到{}条消息，订阅者 2 收到{}条消息", subscriber1Count.get(), subscriber2Count.get());
+    }
+
+    /**
+     * 2 topic 2 subscriber
+     */
+    @Test
+    public void testMultipleTopics() throws InterruptedException {
+        // 准备
+        String topic1 = "device-status";
+        String topic2 = "device-data";
+        String message1 = "设备在线";
+        String message2 = "温度:25°C";
+        CountDownLatch latch = new CountDownLatch(2);
+
+        // 创建订阅者 1 - 只订阅设备状态
+        IotMessageSubscriber<String> statusSubscriber = new IotMessageSubscriber<>() {
+
+            @Override
+            public String getTopic() {
+                return topic1;
+            }
+
+            @Override
+            public String getGroup() {
+                return "status-group";
+            }
+
+            @Override
+            public void onMessage(String message) {
+                log.info("[状态订阅者] 收到消息 - Topic: {}, Message: {}", getTopic(), message);
+                assertEquals(message1, message);
+                latch.countDown();
+            }
+
+        };
+        // 创建订阅者 2 - 只订阅设备数据
+        IotMessageSubscriber<String> dataSubscriber = new IotMessageSubscriber<>() {
+
+            @Override
+            public String getTopic() {
+                return topic2;
+            }
+
+            @Override
+            public String getGroup() {
+                return "data-group";
+            }
+
+            @Override
+            public void onMessage(String message) {
+                log.info("[数据订阅者] 收到消息 - Topic: {}, Message: {}", getTopic(), message);
+                assertEquals(message2, message);
+                latch.countDown();
+            }
+
+        };
+        // 注册订阅者到不同主题
+        messageBus.register(statusSubscriber);
+        messageBus.register(dataSubscriber);
+
+        // 发送消息到不同主题
+        messageBus.post(topic1, message1);
+        messageBus.post(topic2, message2);
+        // 等待消息处理完成
+        boolean completed = latch.await(10, TimeUnit.SECONDS);
+        assertTrue(completed, "消息处理超时");
+        log.info("[测试] 多主题测试完成");
+    }
+
+}

yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/messagebus/core/rocketmq/RocketMQIotMessageBusTest.java

@@ -0,0 +1,270 @@
+package cn.iocoder.yudao.module.iot.core.messagebus.core.rocketmq;
+
+import cn.hutool.core.util.IdUtil;
+import cn.iocoder.yudao.module.iot.core.messagebus.config.IotMessageBusAutoConfiguration;
+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.messagebus.core.TestMessage;
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.spring.autoconfigure.RocketMQAutoConfiguration;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.context.annotation.Import;
+import org.springframework.test.context.TestPropertySource;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * {@link IotRocketMQMessageBus} 集成测试
+ *
+ * @author 芋道源码
+ */
+@SpringBootTest(classes = RocketMQIotMessageBusTest.class)
+@Import({RocketMQAutoConfiguration.class, IotMessageBusAutoConfiguration.class})
+@TestPropertySource(properties = {
+        "yudao.iot.message-bus.type=rocketmq",
+        "rocketmq.name-server=127.0.0.1:9876",
+        "rocketmq.producer.group=test-rocketmq-group",
+        "rocketmq.producer.send-message-timeout=10000"
+})
+@Slf4j
+@Disabled
+public class RocketMQIotMessageBusTest {
+
+    @Resource
+    private IotMessageBus messageBus;
+
+    /**
+     * 1 topic 1 subscriber（string）
+     */
+    @Test
+    public void testSendMessageWithOneSubscriber() throws InterruptedException {
+        // 准备
+        String topic = "test-topic-" + IdUtil.simpleUUID();
+//        String topic = "test-topic-pojo";
+        String testMessage = "Hello IoT Message Bus!";
+        // 用于等待消息处理完成
+        CountDownLatch latch = new CountDownLatch(1);
+        // 用于记录接收到的消息
+        AtomicInteger subscriberCount = new AtomicInteger(0);
+        AtomicReference<String> subscriberMessageRef = new AtomicReference<>();
+
+        // 发送消息（需要提前发，保证 RocketMQ 路由的创建）
+        log.info("[测试] 发送消息 - Topic: {}, Message: {}", topic, testMessage);
+        messageBus.post(topic, testMessage);
+
+        // 创建订阅者
+        IotMessageSubscriber<String> subscriber1 = new IotMessageSubscriber<>() {
+
+            @Override
+            public String getTopic() {
+                return topic;
+            }
+
+            @Override
+            public String getGroup() {
+                return "test-topic-" + IdUtil.simpleUUID() + "-consumer";
+//                return "test-topic-consumer-01";
+            }
+
+            @Override
+            public void onMessage(String message) {
+                log.info("[订阅者] 收到消息 - Topic: {}, Message: {}", getTopic(), message);
+                subscriberCount.incrementAndGet();
+                subscriberMessageRef.set(message);
+                assertEquals(testMessage, message);
+                latch.countDown();
+            }
+
+        };
+        // 注册订阅者
+        messageBus.register(subscriber1);
+
+        // 等待消息处理完成（最多等待 5 秒）
+        boolean completed = latch.await(10, TimeUnit.SECONDS);
+
+        // 验证结果
+        assertTrue(completed, "消息处理超时");
+        assertEquals(1, subscriberCount.get(), "订阅者应该收到 1 条消息");
+        log.info("[测试] 测试完成 - 订阅者收到{}条消息", subscriberCount.get());
+        assertEquals(testMessage, subscriberMessageRef.get(), "接收到的消息内容不匹配");
+    }
+
+    /**
+     * 1 topic 2 subscriber（pojo）
+     */
+    @Test
+    public void testSendMessageWithTwoSubscribers() throws InterruptedException {
+        // 准备
+        String topic = "test-topic-" + IdUtil.simpleUUID();
+//        String topic = "test-topic-pojo";
+        TestMessage testMessage = new TestMessage().setNickname("yunai").setAge(18);
+        // 用于等待消息处理完成
+        CountDownLatch latch = new CountDownLatch(2);
+        // 用于记录接收到的消息
+        AtomicInteger subscriber1Count = new AtomicInteger(0);
+        AtomicReference<TestMessage> subscriber1MessageRef = new AtomicReference<>();
+        AtomicInteger subscriber2Count = new AtomicInteger(0);
+        AtomicReference<TestMessage> subscriber2MessageRef = new AtomicReference<>();
+
+        // 发送消息（需要提前发，保证 RocketMQ 路由的创建）
+        log.info("[测试] 发送消息 - Topic: {}, Message: {}", topic, testMessage);
+        messageBus.post(topic, testMessage);
+
+        // 创建第一个订阅者
+        IotMessageSubscriber<TestMessage> subscriber1 = new IotMessageSubscriber<>() {
+
+            @Override
+            public String getTopic() {
+                return topic;
+            }
+
+            @Override
+            public String getGroup() {
+                return "test-topic-" + IdUtil.simpleUUID() + "-consumer";
+//                return "test-topic-consumer-01";
+            }
+
+            @Override
+            public void onMessage(TestMessage message) {
+                log.info("[订阅者1] 收到消息 - Topic: {}, Message: {}", getTopic(), message);
+                subscriber1Count.incrementAndGet();
+                subscriber1MessageRef.set(message);
+                assertEquals(testMessage, message);
+                latch.countDown();
+            }
+
+        };
+        // 创建第二个订阅者
+        IotMessageSubscriber<TestMessage> subscriber2 = new IotMessageSubscriber<>() {
+
+            @Override
+            public String getTopic() {
+                return topic;
+            }
+
+            @Override
+            public String getGroup() {
+                return "test-topic-" + IdUtil.simpleUUID() + "-consumer";
+//                return "test-topic-consumer-02";
+            }
+
+            @Override
+            public void onMessage(TestMessage message) {
+                log.info("[订阅者2] 收到消息 - Topic: {}, Message: {}", getTopic(), message);
+                subscriber2Count.incrementAndGet();
+                subscriber2MessageRef.set(message);
+                assertEquals(testMessage, message);
+                latch.countDown();
+            }
+
+        };
+        // 注册订阅者
+        messageBus.register(subscriber1);
+        messageBus.register(subscriber2);
+
+        // 等待消息处理完成（最多等待 5 秒）
+        boolean completed = latch.await(10, TimeUnit.SECONDS);
+
+        // 验证结果
+        assertTrue(completed, "消息处理超时");
+        assertEquals(1, subscriber1Count.get(), "订阅者 1 应该收到 1 条消息");
+        assertEquals(1, subscriber2Count.get(), "订阅者 2 应该收到 1 条消息");
+        log.info("[测试] 测试完成 - 订阅者 1 收到{}条消息，订阅者2收到{}条消息", subscriber1Count.get(), subscriber2Count.get());
+        assertEquals(testMessage, subscriber1MessageRef.get(), "接收到的消息内容不匹配");
+        assertEquals(testMessage, subscriber2MessageRef.get(), "接收到的消息内容不匹配");
+    }
+
+    /**
+     * 2 topic 2 subscriber
+     */
+    @Test
+    public void testMultipleTopics() throws InterruptedException {
+        // 准备
+        String topic1 = "device-status-" + IdUtil.simpleUUID();
+        String topic2 = "device-data-" + IdUtil.simpleUUID();
+        String message1 = "设备在线";
+        String message2 = "温度:25°C";
+        CountDownLatch latch = new CountDownLatch(2);
+        AtomicInteger subscriber1Count = new AtomicInteger(0);
+        AtomicReference<String> subscriber1MessageRef = new AtomicReference<>();
+        AtomicInteger subscriber2Count = new AtomicInteger(0);
+        AtomicReference<String> subscriber2MessageRef = new AtomicReference<>();
+
+
+        // 发送消息到不同主题（需要提前发，保证 RocketMQ 路由的创建）
+        log.info("[测试] 发送消息 - Topic1: {}, Message1: {}", topic1, message1);
+        messageBus.post(topic1, message1);
+        log.info("[测试] 发送消息 - Topic2: {}, Message2: {}", topic2, message2);
+        messageBus.post(topic2, message2);
+
+        // 创建订阅者 1 - 只订阅设备状态
+        IotMessageSubscriber<String> statusSubscriber = new IotMessageSubscriber<>() {
+
+            @Override
+            public String getTopic() {
+                return topic1;
+            }
+
+            @Override
+            public String getGroup() {
+                return "status-group-" + IdUtil.simpleUUID();
+            }
+
+            @Override
+            public void onMessage(String message) {
+                log.info("[状态订阅者] 收到消息 - Topic: {}, Message: {}", getTopic(), message);
+                subscriber1Count.incrementAndGet();
+                subscriber1MessageRef.set(message);
+                assertEquals(message1, message);
+                latch.countDown();
+            }
+
+        };
+        // 创建订阅者 2 - 只订阅设备数据
+        IotMessageSubscriber<String> dataSubscriber = new IotMessageSubscriber<>() {
+
+            @Override
+            public String getTopic() {
+                return topic2;
+            }
+
+            @Override
+            public String getGroup() {
+                return "data-group-" + IdUtil.simpleUUID();
+            }
+
+            @Override
+            public void onMessage(String message) {
+                log.info("[数据订阅者] 收到消息 - Topic: {}, Message: {}", getTopic(), message);
+                subscriber2Count.incrementAndGet();
+                subscriber2MessageRef.set(message);
+                assertEquals(message2, message);
+                latch.countDown();
+            }
+
+        };
+        // 注册订阅者到不同主题
+        messageBus.register(statusSubscriber);
+        messageBus.register(dataSubscriber);
+
+        // 等待消息处理完成
+        boolean completed = latch.await(10, TimeUnit.SECONDS);
+
+        // 验证结果
+        assertTrue(completed, "消息处理超时");
+        assertEquals(1, subscriber1Count.get(), "状态订阅者应该收到 1 条消息");
+        assertEquals(message1, subscriber1MessageRef.get(), "状态订阅者接收到的消息内容不匹配");
+        assertEquals(1, subscriber2Count.get(), "数据订阅者应该收到 1 条消息");
+        assertEquals(message2, subscriber2MessageRef.get(), "数据订阅者接收到的消息内容不匹配");
+        log.info("[测试] 多主题测试完成 - 状态订阅者收到{}条消息，数据订阅者收到{}条消息", subscriber1Count.get(), subscriber2Count.get());
+    }
+
+}

yudao-module-iot/yudao-module-iot-core/src/test/java/cn/iocoder/yudao/module/iot/core/util/IotDeviceMessageUtilsTest.java

@@ -0,0 +1,351 @@
+package cn.iocoder.yudao.module.iot.core.util;
+
+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.core.topic.event.IotDeviceEventPostReqDTO;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * {@link IotDeviceMessageUtils} 的单元测试
+ *
+ * @author HUIHUI
+ */
+public class IotDeviceMessageUtilsTest {
+
+    @Test
+    public void testExtractPropertyValue_directValue() {
+        // 测试直接值（非 Map）
+        IotDeviceMessage message = new IotDeviceMessage();
+        message.setParams(25.5);
+
+        Object result = IotDeviceMessageUtils.extractPropertyValue(message, "temperature");
+        assertEquals(25.5, result);
+    }
+
+    @Test
+    public void testExtractPropertyValue_directIdentifier() {
+        // 测试直接通过标识符获取
+        IotDeviceMessage message = new IotDeviceMessage();
+        Map<String, Object> params = new HashMap<>();
+        params.put("temperature", 25.5);
+        message.setParams(params);
+
+        Object result = IotDeviceMessageUtils.extractPropertyValue(message, "temperature");
+        assertEquals(25.5, result);
+    }
+
+    @Test
+    public void testExtractPropertyValue_propertiesStructure() {
+        // 测试 properties 结构
+        IotDeviceMessage message = new IotDeviceMessage();
+        Map<String, Object> properties = new HashMap<>();
+        properties.put("temperature", 25.5);
+        properties.put("humidity", 60);
+
+        Map<String, Object> params = new HashMap<>();
+        params.put("properties", properties);
+        message.setParams(params);
+
+        Object result = IotDeviceMessageUtils.extractPropertyValue(message, "temperature");
+        assertEquals(25.5, result);
+    }
+
+    @Test
+    public void testExtractPropertyValue_dataStructure() {
+        // 测试 data 结构
+        IotDeviceMessage message = new IotDeviceMessage();
+        Map<String, Object> data = new HashMap<>();
+        data.put("temperature", 25.5);
+
+        Map<String, Object> params = new HashMap<>();
+        params.put("data", data);
+        message.setParams(params);
+
+        Object result = IotDeviceMessageUtils.extractPropertyValue(message, "temperature");
+        assertEquals(25.5, result);
+    }
+
+    @Test
+    public void testExtractPropertyValue_valueField() {
+        // 测试 value 字段（策略 5）
+        IotDeviceMessage message = new IotDeviceMessage();
+        Map<String, Object> params = new HashMap<>();
+        params.put("identifier", "temperature");
+        params.put("value", 25.5);
+        message.setParams(params);
+
+        Object result = IotDeviceMessageUtils.extractPropertyValue(message, "temperature");
+        assertEquals(25.5, result);
+    }
+
+    @Test
+    public void testExtractPropertyValue_singleValueMap() {
+        // 测试单值 Map（策略 6：包含 identifier 和一个其他字段）
+        IotDeviceMessage message = new IotDeviceMessage();
+        Map<String, Object> params = new HashMap<>();
+        params.put("identifier", "temperature");
+        params.put("actualValue", 25.5);
+        message.setParams(params);
+
+        Object result = IotDeviceMessageUtils.extractPropertyValue(message, "temperature");
+        assertEquals(25.5, result);
+    }
+
+    @Test
+    public void testExtractPropertyValue_notFound() {
+        // 测试未找到属性值
+        IotDeviceMessage message = new IotDeviceMessage();
+        Map<String, Object> params = new HashMap<>();
+        params.put("humidity", 60);
+        message.setParams(params);
+
+        Object result = IotDeviceMessageUtils.extractPropertyValue(message, "temperature");
+        assertNull(result);
+    }
+
+    @Test
+    public void testExtractPropertyValue_nullParams() {
+        // 测试 params 为 null
+        IotDeviceMessage message = new IotDeviceMessage();
+        message.setParams(null);
+
+        Object result = IotDeviceMessageUtils.extractPropertyValue(message, "temperature");
+        assertNull(result);
+    }
+
+    @Test
+    public void testExtractPropertyValue_priorityOrder() {
+        // 测试优先级顺序：直接标识符 > properties > data > value
+        IotDeviceMessage message = new IotDeviceMessage();
+
+        Map<String, Object> properties = new HashMap<>();
+        properties.put("temperature", 20.0);
+
+        Map<String, Object> data = new HashMap<>();
+        data.put("temperature", 30.0);
+
+        Map<String, Object> params = new HashMap<>();
+        params.put("temperature", 25.5); // 最高优先级
+        params.put("properties", properties);
+        params.put("data", data);
+        params.put("value", 40.0);
+        message.setParams(params);
+
+        Object result = IotDeviceMessageUtils.extractPropertyValue(message, "temperature");
+        assertEquals(25.5, result); // 应该返回直接标识符的值
+    }
+
+    // ========== extractEventValue 测试 ==========
+
+    @Test
+    public void testExtractEventValue_scalar() {
+        // 标量事件值：{identifier: "gzzt", value: "normal"}
+        IotDeviceMessage message = new IotDeviceMessage();
+        Map<String, Object> params = new HashMap<>();
+        params.put("identifier", "gzzt");
+        params.put("value", "normal");
+        message.setParams(params);
+
+        Object result = IotDeviceMessageUtils.extractEventValue(message);
+        assertEquals("normal", result);
+    }
+
+    @Test
+    public void testExtractEventValue_struct() {
+        // 结构体事件值：{identifier: "alarm", value: {level: "high", message: "..."}}
+        IotDeviceMessage message = new IotDeviceMessage();
+        Map<String, Object> eventValue = new HashMap<>();
+        eventValue.put("level", "high");
+        eventValue.put("message", "over temperature");
+
+        Map<String, Object> params = new HashMap<>();
+        params.put("identifier", "alarm");
+        params.put("value", eventValue);
+        message.setParams(params);
+
+        Object result = IotDeviceMessageUtils.extractEventValue(message);
+        assertEquals(eventValue, result);
+    }
+
+    @Test
+    public void testExtractEventValue_nullParams() {
+        IotDeviceMessage message = new IotDeviceMessage();
+        message.setParams(null);
+
+        Object result = IotDeviceMessageUtils.extractEventValue(message);
+        assertNull(result);
+    }
+
+    @Test
+    public void testExtractEventValue_paramsWithoutValueField() {
+        // params 是字符串等非结构化对象，无 value 字段，应返回 null
+        IotDeviceMessage message = new IotDeviceMessage();
+        message.setParams("not a map");
+
+        Object result = IotDeviceMessageUtils.extractEventValue(message);
+        assertNull(result);
+    }
+
+    @Test
+    public void testExtractEventValue_missingValueField() {
+        IotDeviceMessage message = new IotDeviceMessage();
+        Map<String, Object> params = new HashMap<>();
+        params.put("identifier", "gzzt");
+        message.setParams(params);
+
+        Object result = IotDeviceMessageUtils.extractEventValue(message);
+        assertNull(result);
+    }
+
+    @Test
+    public void testExtractEventValue_pojoParams() {
+        // 本地总线场景：params 是 IotDeviceEventPostReqDTO POJO（未经 JSON 反序列化），应能反射取到 value
+        IotDeviceMessage message = new IotDeviceMessage();
+        message.setParams(IotDeviceEventPostReqDTO.of("gzzt", "normal"));
+
+        Object result = IotDeviceMessageUtils.extractEventValue(message);
+        assertEquals("normal", result);
+    }
+
+    @Test
+    public void testGetIdentifier_eventPostPojoParams() {
+        // 本地总线场景：EVENT_POST 消息 params 是 DTO POJO，仍应能解析出 identifier
+        IotDeviceMessage message = new IotDeviceMessage();
+        message.setMethod(IotDeviceMessageMethodEnum.EVENT_POST.getMethod());
+        message.setParams(IotDeviceEventPostReqDTO.of("gzzt", "normal"));
+
+        assertEquals("gzzt", IotDeviceMessageUtils.getIdentifier(message));
+    }
+
+    // ========== notContainsIdentifier 测试 ==========
+
+    /**
+     * 测试 notContainsIdentifier 与 containsIdentifier 的互补性
+     * **Property 2: notContainsIdentifier 与 containsIdentifier 互补性**
+     * **Validates: Requirements 4.1**
+     */
+    @Test
+    public void testNotContainsIdentifier_complementary_whenContains() {
+        // 准备参数：消息包含指定标识符
+        IotDeviceMessage message = new IotDeviceMessage();
+        message.setMethod(IotDeviceMessageMethodEnum.PROPERTY_POST.getMethod());
+        Map<String, Object> params = new HashMap<>();
+        params.put("temperature", 25);
+        message.setParams(params);
+        String identifier = "temperature";
+
+        // 调用 & 断言：验证互补性
+        boolean containsResult = IotDeviceMessageUtils.containsIdentifier(message, identifier);
+        boolean notContainsResult = IotDeviceMessageUtils.notContainsIdentifier(message, identifier);
+        assertTrue(containsResult);
+        assertFalse(notContainsResult);
+        assertEquals(!containsResult, notContainsResult);
+    }
+
+    /**
+     * 测试 notContainsIdentifier 与 containsIdentifier 的互补性
+     * **Property 2: notContainsIdentifier 与 containsIdentifier 互补性**
+     * **Validates: Requirements 4.1**
+     */
+    @Test
+    public void testNotContainsIdentifier_complementary_whenNotContains() {
+        // 准备参数：消息不包含指定标识符
+        IotDeviceMessage message = new IotDeviceMessage();
+        message.setMethod(IotDeviceMessageMethodEnum.PROPERTY_POST.getMethod());
+        Map<String, Object> params = new HashMap<>();
+        params.put("temperature", 25);
+        message.setParams(params);
+        String identifier = "humidity";
+
+        // 调用 & 断言：验证互补性
+        boolean containsResult = IotDeviceMessageUtils.containsIdentifier(message, identifier);
+        boolean notContainsResult = IotDeviceMessageUtils.notContainsIdentifier(message, identifier);
+        assertFalse(containsResult);
+        assertTrue(notContainsResult);
+        assertEquals(!containsResult, notContainsResult);
+    }
+
+    /**
+     * 测试 notContainsIdentifier 与 containsIdentifier 的互补性 - 空参数场景
+     * **Property 2: notContainsIdentifier 与 containsIdentifier 互补性**
+     * **Validates: Requirements 4.1**
+     */
+    @Test
+    public void testNotContainsIdentifier_complementary_nullParams() {
+        // 准备参数：params 为 null
+        IotDeviceMessage message = new IotDeviceMessage();
+        message.setParams(null);
+        String identifier = "temperature";
+
+        // 调用 & 断言：验证互补性
+        boolean containsResult = IotDeviceMessageUtils.containsIdentifier(message, identifier);
+        boolean notContainsResult = IotDeviceMessageUtils.notContainsIdentifier(message, identifier);
+        assertFalse(containsResult);
+        assertTrue(notContainsResult);
+        assertEquals(!containsResult, notContainsResult);
+    }
+
+    // ========== getPropertyIdentifiers 测试 ==========
+
+    @Test
+    public void testGetPropertyIdentifiers_flatStructure() {
+        // 扁平结构：顶层 key 即标识符
+        IotDeviceMessage message = new IotDeviceMessage();
+        Map<String, Object> params = new HashMap<>();
+        params.put("temperature", 25.5);
+        params.put("humidity", 60);
+        message.setParams(params);
+
+        Set<String> identifiers = IotDeviceMessageUtils.getPropertyIdentifiers(message);
+        assertEquals(2, identifiers.size());
+        assertTrue(identifiers.contains("temperature"));
+        assertTrue(identifiers.contains("humidity"));
+    }
+
+    @Test
+    public void testGetPropertyIdentifiers_nullMessage() {
+        // 入参为 null：返回空集合
+        Set<String> identifiers = IotDeviceMessageUtils.getPropertyIdentifiers(null);
+        assertNotNull(identifiers);
+        assertTrue(identifiers.isEmpty());
+    }
+
+    @Test
+    public void testGetPropertyIdentifiers_nullParams() {
+        // params 为 null：返回空集合
+        IotDeviceMessage message = new IotDeviceMessage();
+        message.setParams(null);
+
+        Set<String> identifiers = IotDeviceMessageUtils.getPropertyIdentifiers(message);
+        assertTrue(identifiers.isEmpty());
+    }
+
+    @Test
+    public void testGetPropertyIdentifiers_emptyParams() {
+        // params 为空 Map：返回空集合
+        IotDeviceMessage message = new IotDeviceMessage();
+        message.setParams(new HashMap<>());
+
+        Set<String> identifiers = IotDeviceMessageUtils.getPropertyIdentifiers(message);
+        assertTrue(identifiers.isEmpty());
+    }
+
+    @Test
+    public void testGetPropertyIdentifiers_jsonStringParams() {
+        // params 为 JSON 字符串：parseParamsToMap 解析后正常提取顶层标识符
+        IotDeviceMessage message = new IotDeviceMessage();
+        message.setParams("{\"temperature\":25.5,\"humidity\":60}");
+
+        Set<String> identifiers = IotDeviceMessageUtils.getPropertyIdentifiers(message);
+        assertEquals(2, identifiers.size());
+        assertTrue(identifiers.contains("temperature"));
+        assertTrue(identifiers.contains("humidity"));
+    }
+
+}