feat：【IoT 物联网】新增 MQTT HTTP 认证协议，优化设备认证逻辑，更新配置文件以支持 HTTP 认证服务

yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/config/IotGatewayConfiguration.java

@@ -3,6 +3,9 @@ package cn.iocoder.yudao.module.iot.gateway.config;
 import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;
 import cn.iocoder.yudao.module.iot.gateway.protocol.http.IotHttpDownstreamSubscriber;
 import cn.iocoder.yudao.module.iot.gateway.protocol.http.IotHttpUpstreamProtocol;
+import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttDownstreamSubscriber;
+import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttHttpAuthProtocol;
+import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttUpstreamProtocol;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
@@ -42,17 +45,24 @@ public class IotGatewayConfiguration {
     @Slf4j
     public static class MqttProtocolConfiguration {
 
-        // TODO @haohao：临时注释，避免报错
-//        @Bean
-//        public IotMqttUpstreamProtocol iotMqttUpstreamProtocol(IotGatewayProperties gatewayProperties) {
-//            return new IotMqttUpstreamProtocol(gatewayProperties.getProtocol().getEmqx());
-//        }
-//
-//        @Bean
-//        public IotMqttDownstreamSubscriber iotMqttDownstreamSubscriber(IotMqttUpstreamProtocol mqttUpstreamProtocol,
-//                                                                       IotMessageBus messageBus) {
-//            return new IotMqttDownstreamSubscriber(mqttUpstreamProtocol, messageBus);
-//        }
+        @Bean
+        public IotMqttUpstreamProtocol iotMqttUpstreamProtocol(IotGatewayProperties gatewayProperties) {
+            return new IotMqttUpstreamProtocol(gatewayProperties.getProtocol().getEmqx());
+        }
+
+        @Bean
+        public IotMqttDownstreamSubscriber iotMqttDownstreamSubscriber(IotMqttUpstreamProtocol mqttUpstreamProtocol,
+                                                                       IotMessageBus messageBus) {
+            return new IotMqttDownstreamSubscriber(mqttUpstreamProtocol, messageBus);
+        }
+
+        /**
+         * MQTT HTTP 认证协议：提供 HTTP 认证接口供 EMQX 调用
+         */
+        @Bean
+        public IotMqttHttpAuthProtocol mqttHttpAuthProtocol(IotGatewayProperties gatewayProperties) {
+            return new IotMqttHttpAuthProtocol(gatewayProperties.getProtocol().getEmqx());
+        }
     }
 
 }

yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/config/IotGatewayProperties.java

@@ -103,6 +103,10 @@ public class IotGatewayProperties {
          */
         @NotNull(message = "是否开启不能为空")
         private Boolean enabled;
+        /**
+         * HTTP 认证端口
+         */
+        private Integer httpAuthPort = 8090;
         /**
          * MQTT 服务器地址
          */
@@ -127,10 +131,6 @@ public class IotGatewayProperties {
          * MQTT 主题
          */
         private List<String> mqttTopics;
-        /**
-         * 认证端口
-         */
-        private Integer authPort;
 
     }
 

yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/IotMqttDownstreamSubscriber.java

@@ -1,12 +1,14 @@
 package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt;
 
 import cn.hutool.json.JSONObject;
+import cn.iocoder.yudao.module.iot.core.biz.IotDeviceCommonApi;
+import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceGetReqDTO;
+import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceRespDTO;
 import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;
 import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageSubscriber;
 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
 import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils;
 import cn.iocoder.yudao.module.iot.gateway.enums.IotDeviceTopicEnum;
-import cn.iocoder.yudao.module.iot.gateway.service.device.IotDeviceCacheService;
 import jakarta.annotation.PostConstruct;
 import jakarta.annotation.Resource;
 import lombok.RequiredArgsConstructor;
@@ -25,7 +27,7 @@ public class IotMqttDownstreamSubscriber implements IotMessageSubscriber<IotDevi
     private final IotMessageBus messageBus;
 
     @Resource
-    private IotDeviceCacheService deviceCacheService;
+    private IotDeviceCommonApi deviceApi;
 
     @PostConstruct
     public void init() {
@@ -81,7 +83,9 @@ public class IotMqttDownstreamSubscriber implements IotMessageSubscriber<IotDevi
         String method = message.getMethod();
 
         // 通过 deviceId 获取设备信息
-        IotDeviceCacheService.DeviceInfo deviceInfo = deviceCacheService.getDeviceInfoById(message.getDeviceId());
+        IotDeviceGetReqDTO reqDTO = new IotDeviceGetReqDTO();
+        reqDTO.setId(message.getDeviceId());
+        IotDeviceRespDTO deviceInfo = deviceApi.getDevice(reqDTO).getData();
         if (deviceInfo == null) {
             log.warn("[handlePropertyMessage][设备信息不存在][deviceId: {}]", message.getDeviceId());
             return;
@@ -116,7 +120,9 @@ public class IotMqttDownstreamSubscriber implements IotMessageSubscriber<IotDevi
         String method = message.getMethod();
 
         // 通过 deviceId 获取设备信息
-        IotDeviceCacheService.DeviceInfo deviceInfo = deviceCacheService.getDeviceInfoById(message.getDeviceId());
+        IotDeviceGetReqDTO reqDTO = new IotDeviceGetReqDTO();
+        reqDTO.setId(message.getDeviceId());
+        IotDeviceRespDTO deviceInfo = deviceApi.getDevice(reqDTO).getData();
         if (deviceInfo == null) {
             log.warn("[handleServiceMessage][设备信息不存在][deviceId: {}]", message.getDeviceId());
             return;
@@ -141,7 +147,9 @@ public class IotMqttDownstreamSubscriber implements IotMessageSubscriber<IotDevi
      */
     private void handleConfigMessage(IotDeviceMessage message) {
         // 通过 deviceId 获取设备信息
-        IotDeviceCacheService.DeviceInfo deviceInfo = deviceCacheService.getDeviceInfoById(message.getDeviceId());
+        IotDeviceGetReqDTO reqDTO = new IotDeviceGetReqDTO();
+        reqDTO.setId(message.getDeviceId());
+        IotDeviceRespDTO deviceInfo = deviceApi.getDevice(reqDTO).getData();
         if (deviceInfo == null) {
             log.warn("[handleConfigMessage][设备信息不存在][deviceId: {}]", message.getDeviceId());
             return;
@@ -163,7 +171,9 @@ public class IotMqttDownstreamSubscriber implements IotMessageSubscriber<IotDevi
      */
     private void handleOtaMessage(IotDeviceMessage message) {
         // 通过 deviceId 获取设备信息
-        IotDeviceCacheService.DeviceInfo deviceInfo = deviceCacheService.getDeviceInfoById(message.getDeviceId());
+        IotDeviceGetReqDTO reqDTO = new IotDeviceGetReqDTO();
+        reqDTO.setId(message.getDeviceId());
+        IotDeviceRespDTO deviceInfo = deviceApi.getDevice(reqDTO).getData();
         if (deviceInfo == null) {
             log.warn("[handleOtaMessage][设备信息不存在][deviceId: {}]", message.getDeviceId());
             return;

yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/IotMqttHttpAuthProtocol.java

@@ -0,0 +1,96 @@
+package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt;
+
+import cn.iocoder.yudao.module.iot.gateway.config.IotGatewayProperties;
+import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router.IotMqttHttpAuthHandler;
+import io.vertx.core.AbstractVerticle;
+import io.vertx.core.Vertx;
+import io.vertx.core.http.HttpServer;
+import io.vertx.ext.web.Router;
+import io.vertx.ext.web.handler.BodyHandler;
+import jakarta.annotation.PostConstruct;
+import jakarta.annotation.PreDestroy;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * IoT 网关 MQTT HTTP 认证协议：提供 HTTP 认证接口供 EMQX 调用
+ *
+ * @author 芋道源码
+ */
+@RequiredArgsConstructor
+@Slf4j
+public class IotMqttHttpAuthProtocol extends AbstractVerticle {
+
+    private final IotGatewayProperties.EmqxProperties emqxProperties;
+
+    private HttpServer httpServer;
+    private Vertx vertx;
+
+    @Override
+    @PostConstruct
+    public void start() {
+        log.info("[start][开始启动 MQTT HTTP 认证协议服务]");
+
+        // 创建 Vertx 实例
+        this.vertx = Vertx.vertx();
+
+        // 创建路由
+        Router router = Router.router(vertx);
+        router.route().handler(BodyHandler.create());
+
+        // 创建 MQTT 认证处理器
+        IotMqttHttpAuthHandler authHandler = new IotMqttHttpAuthHandler();
+
+        // 添加 MQTT 认证路由
+        router.post("/mqtt/auth/authenticate").handler(authHandler::authenticate);
+        router.post("/mqtt/auth/connected").handler(authHandler::connected);
+        router.post("/mqtt/auth/disconnected").handler(authHandler::disconnected);
+
+        // 启动 HTTP 服务器，使用独立端口
+        int authPort = getAuthPort();
+        try {
+            httpServer = vertx.createHttpServer()
+                    .requestHandler(router)
+                    .listen(authPort)
+                    .result();
+            log.info("[start][MQTT HTTP 认证协议启动成功，端口：{}]", authPort);
+        } catch (Exception e) {
+            log.error("[start][MQTT HTTP 认证协议启动失败]", e);
+            throw e;
+        }
+    }
+
+    @Override
+    @PreDestroy
+    public void stop() {
+        if (httpServer != null) {
+            try {
+                httpServer.close().result();
+                log.info("[stop][MQTT HTTP 认证协议已停止]");
+            } catch (Exception e) {
+                log.error("[stop][MQTT HTTP 认证协议停止失败]", e);
+            }
+        }
+
+        if (vertx != null) {
+            try {
+                vertx.close();
+                log.info("[stop][Vertx 已关闭]");
+            } catch (Exception e) {
+                log.warn("[stop][关闭 Vertx 异常]", e);
+            }
+        }
+    }
+
+    /**
+     * 获取认证服务端口
+     * 从配置中获取，默认使用 8090 端口
+     */
+    private int getAuthPort() {
+        return emqxProperties.getHttpAuthPort() != null ? emqxProperties.getHttpAuthPort() : 8090;
+    }
+
+    public String getServerId() {
+        return "mqtt_auth_gateway_" + getAuthPort();
+    }
+}

yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/IotMqttUpstreamProtocol.http

@@ -0,0 +1,139 @@
+### IoT MQTT HTTP 认证服务测试文件
+### 用于测试独立的 MQTT HTTP 认证服务接口
+### 服务由 IotMqttHttpAuthProtocol 启动，运行在配置的端口上（默认 8090）
+### 注意：这些接口通常由 EMQX 调用，也可以用于手动测试
+
+### 配置说明：
+### 在 application-local.yaml 中可以配置端口：
+### yudao:
+###   iot:
+###     gateway:
+###       protocol:
+###         emqx:
+###           http-auth-port: 8090  # 可以修改为其他端口
+
+### 1. MQTT 设备认证接口 => 成功认证
+POST http://localhost:8090/mqtt/auth/authenticate
+Content-Type: application/json
+
+{
+  "clientid": "test_product.test_device",
+  "username": "test_device&test_product",
+  "password": "test_device_secret_hmac_password"
+}
+
+### 2. MQTT 设备认证接口 => 参数不完整（clientid 为空）
+POST http://localhost:8090/mqtt/auth/authenticate
+Content-Type: application/json
+
+{
+  "clientid": "",
+  "username": "test_device&test_product",
+  "password": "test_device_secret"
+}
+
+### 3. MQTT 设备认证接口 => 参数不完整（username 为空）
+POST http://localhost:8090/mqtt/auth/authenticate
+Content-Type: application/json
+
+{
+  "clientid": "test_product.test_device",
+  "username": "",
+  "password": "test_device_secret"
+}
+
+### 4. MQTT 设备认证接口 => 参数不完整（password 为空）
+POST http://localhost:8090/mqtt/auth/authenticate
+Content-Type: application/json
+
+{
+  "clientid": "test_product.test_device",
+  "username": "test_device&test_product",
+  "password": ""
+}
+
+### 5. MQTT 设备认证接口 => 认证失败（错误的设备密钥）
+POST http://localhost:8090/mqtt/auth/authenticate
+Content-Type: application/json
+
+{
+  "clientid": "test_product.test_device",
+  "username": "test_device&test_product",
+  "password": "wrong_password"
+}
+
+### 6. EMQX 客户端连接事件钩子 => 设备上线
+POST http://localhost:8090/mqtt/auth/connected
+Content-Type: application/json
+
+{
+  "clientid": "test_product.test_device",
+  "username": "test_device&test_product",
+  "timestamp": 1703808000000,
+  "peername": "192.168.1.100:52036",
+  "sockname": "127.0.0.1:1883"
+}
+
+### 7. EMQX 客户端连接事件钩子 => 用户名为空
+POST http://localhost:8090/mqtt/auth/connected
+Content-Type: application/json
+
+{
+  "clientid": "test_product.test_device",
+  "username": "",
+  "timestamp": 1703808000000,
+  "peername": "192.168.1.100:52036",
+  "sockname": "127.0.0.1:1883"
+}
+
+### 8. EMQX 客户端断开连接事件钩子 => 设备下线（正常断开）
+POST http://localhost:8090/mqtt/auth/disconnected
+Content-Type: application/json
+
+{
+  "clientid": "test_product.test_device",
+  "username": "test_device&test_product",
+  "reason": "normal",
+  "timestamp": 1703808060000
+}
+
+### 9. 使用自定义端口测试（如果配置了不同端口）
+### 假设配置了端口 8091：
+### POST http://localhost:8091/mqtt/auth/authenticate
+### Content-Type: application/json
+### 
+### {
+###   "clientid": "test_product.test_device",
+###   "username": "test_device&test_product",
+###   "password": "test_device_secret_hmac_password"
+### }
+
+### EMQX 配置参考：
+### authentication = [
+###   {
+###     mechanism = http
+###     method = post
+###     url = "http://localhost:8090/mqtt/auth/authenticate"  # 使用配置的端口
+###     body = {
+###       clientid = "${clientid}"
+###       username = "${username}"
+###       password = "${password}"
+###     }
+###     headers = {
+###       "content-type" = "application/json"
+###     }
+###   }
+### ]
+###
+### webhooks = [
+###   {
+###     name = "client_connected"
+###     url = "http://localhost:8090/mqtt/auth/connected"    # 使用配置的端口
+###     events = ["client.connected"]
+###   },
+###   {
+###     name = "client_disconnected"
+###     url = "http://localhost:8090/mqtt/auth/disconnected" # 使用配置的端口
+###     events = ["client.disconnected"]
+###   }
+### ] 

yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/IotMqttUpstreamProtocol.java

@@ -1,15 +1,11 @@
 package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt;
 
 import cn.hutool.core.collection.CollUtil;
-import cn.hutool.core.util.ArrayUtil;
 import cn.hutool.core.util.IdUtil;
 import cn.hutool.core.util.ObjUtil;
 import cn.hutool.core.util.StrUtil;
-import cn.hutool.extra.spring.SpringUtil;
-import cn.iocoder.yudao.module.iot.core.mq.producer.IotDeviceMessageProducer;
 import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils;
 import cn.iocoder.yudao.module.iot.gateway.config.IotGatewayProperties;
-import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router.IotMqttAuthRouter;
 import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router.IotMqttUpstreamRouter;
 import io.netty.handler.codec.mqtt.MqttQoS;
 import io.vertx.core.Vertx;
@@ -43,8 +39,6 @@ public class IotMqttUpstreamProtocol {
     private Vertx vertx;
     private MqttClient mqttClient;
     private IotMqttUpstreamRouter messageRouter;
-    private IotMqttAuthRouter authRouter;
-        private IotDeviceMessageProducer deviceMessageProducer;
 
     /**
      * 服务运行状态标志
@@ -61,9 +55,7 @@ public class IotMqttUpstreamProtocol {
 
         // 初始化组件
         this.vertx = Vertx.vertx();
-        this.deviceMessageProducer = SpringUtil.getBean(IotDeviceMessageProducer.class);
         this.messageRouter = new IotMqttUpstreamRouter(this);
-        this.authRouter = new IotMqttAuthRouter(this);
 
         // 创建 MQTT 客户端
         MqttClientOptions options = new MqttClientOptions()
@@ -255,13 +247,4 @@ public class IotMqttUpstreamProtocol {
         return mqttClient;
     }
 
-    /**
-     * 获取认证路由器
-     *
-     * @return 认证路由器
-     */
-    public IotMqttAuthRouter getAuthRouter() {
-        return authRouter;
-    }
-
 }

yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttAuthRouter.java

@@ -1,149 +0,0 @@
-package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router;
-
-import cn.hutool.core.util.StrUtil;
-import cn.hutool.extra.spring.SpringUtil;
-import cn.iocoder.yudao.framework.common.pojo.CommonResult;
-import cn.iocoder.yudao.module.iot.core.biz.IotDeviceCommonApi;
-import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceAuthReqDTO;
-import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
-import cn.iocoder.yudao.module.iot.core.mq.producer.IotDeviceMessageProducer;
-import cn.iocoder.yudao.module.iot.core.util.IotDeviceAuthUtils;
-import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttUpstreamProtocol;
-import cn.iocoder.yudao.module.iot.gateway.service.auth.IotDeviceTokenService;
-import cn.iocoder.yudao.module.iot.gateway.service.message.IotDeviceMessageService;
-import lombok.RequiredArgsConstructor;
-import lombok.extern.slf4j.Slf4j;
-
-/**
- * IoT 网关 MQTT 认证路由器
- * <p>
- * 处理设备的 MQTT 连接认证和连接状态管理
- *
- * @author 芋道源码
- */
-@RequiredArgsConstructor
-@Slf4j
-public class IotMqttAuthRouter {
-
-    private final IotMqttUpstreamProtocol protocol;
-    private final IotDeviceMessageProducer deviceMessageProducer;
-    private final IotDeviceTokenService deviceTokenService;
-    private final IotDeviceCommonApi deviceCommonApi;
-    private final IotDeviceMessageService deviceMessageService;
-
-    public IotMqttAuthRouter(IotMqttUpstreamProtocol protocol) {
-        this.protocol = protocol;
-        this.deviceMessageProducer = SpringUtil.getBean(IotDeviceMessageProducer.class);
-        this.deviceTokenService = SpringUtil.getBean(IotDeviceTokenService.class);
-        this.deviceCommonApi = SpringUtil.getBean(IotDeviceCommonApi.class);
-        this.deviceMessageService = SpringUtil.getBean(IotDeviceMessageService.class);
-    }
-
-    /**
-     * 处理设备认证
-     *
-     * @param clientId 客户端 ID
-     * @param username 用户名
-     * @param password 密码
-     * @return 认证结果
-     */
-    public boolean authenticate(String clientId, String username, String password) {
-        try {
-            log.info("[authenticate][开始认证设备][clientId: {}][username: {}]", clientId, username);
-
-            // 1. 参数校验
-            if (StrUtil.isEmpty(clientId) || StrUtil.isEmpty(username) || StrUtil.isEmpty(password)) {
-                log.warn("[authenticate][认证参数不完整][clientId: {}][username: {}]", clientId, username);
-                return false;
-            }
-
-            // 2. 执行认证
-            CommonResult<Boolean> result = deviceCommonApi.authDevice(new IotDeviceAuthReqDTO()
-                    .setClientId(clientId).setUsername(username).setPassword(password));
-            result.checkError();
-            if (!Boolean.TRUE.equals(result.getData())) {
-                log.warn("[authenticate][设备认证失败][clientId: {}][username: {}]", clientId, username);
-                return false;
-            }
-
-            log.info("[authenticate][设备认证成功][clientId: {}][username: {}]", clientId, username);
-            return true;
-        } catch (Exception e) {
-            log.error("[authenticate][设备认证异常][clientId: {}][username: {}]", clientId, username, e);
-            return false;
-        }
-    }
-
-    /**
-     * 处理设备连接事件
-     *
-     * @param clientId 客户端 ID
-     * @param username 用户名
-     */
-    public void handleClientConnected(String clientId, String username) {
-        try {
-            log.info("[handleClientConnected][设备连接][clientId: {}][username: {}]", clientId, username);
-
-            // 解析设备信息并发送上线消息
-            handleDeviceStateChange(username, true);
-        } catch (Exception e) {
-            log.error("[handleClientConnected][处理设备连接事件异常][clientId: {}][username: {}]", clientId, username, e);
-        }
-    }
-
-    /**
-     * 处理设备断开连接事件
-     *
-     * @param clientId 客户端 ID
-     * @param username 用户名
-     */
-    public void handleClientDisconnected(String clientId, String username) {
-        try {
-            log.info("[handleClientDisconnected][设备断开连接][clientId: {}][username: {}]", clientId, username);
-
-            // 解析设备信息并发送下线消息
-            handleDeviceStateChange(username, false);
-        } catch (Exception e) {
-            log.error("[handleClientDisconnected][处理设备断开连接事件异常][clientId: {}][username: {}]", clientId, username, e);
-        }
-    }
-
-    /**
-     * 处理设备状态变化
-     *
-     * @param username 用户名
-     * @param online   是否在线
-     */
-    private void handleDeviceStateChange(String username, boolean online) {
-        // 解析设备信息
-        if (StrUtil.isEmpty(username) || "undefined".equals(username)) {
-            log.warn("[handleDeviceStateChange][用户名为空或未定义][username: {}]", username);
-            return;
-        }
-
-        IotDeviceAuthUtils.DeviceInfo deviceInfo = deviceTokenService.parseUsername(username);
-        if (deviceInfo == null) {
-            log.warn("[handleDeviceStateChange][无法解析设备信息][username: {}]", username);
-            return;
-        }
-
-        try {
-            // 使用 IotDeviceMessageService 构建设备状态消息
-            IotDeviceMessage message;
-            if (online) {
-                message = deviceMessageService.buildDeviceMessageOfStateOnline(
-                        deviceInfo.getProductKey(), deviceInfo.getDeviceName(), protocol.getServerId());
-                log.info("[handleDeviceStateChange][发送设备上线消息成功][username: {}]", username);
-            } else {
-                message = deviceMessageService.buildDeviceMessageOfStateOffline(
-                        deviceInfo.getProductKey(), deviceInfo.getDeviceName(), protocol.getServerId());
-                log.info("[handleDeviceStateChange][发送设备下线消息成功][username: {}]", username);
-            }
-
-            deviceMessageProducer.sendDeviceMessage(message);
-        } catch (Exception e) {
-            log.error("[handleDeviceStateChange][发送设备状态消息失败][username: {}][online: {}]", username, online, e);
-        }
-    }
-
-}

yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttEventHandler.java

@@ -6,7 +6,7 @@ import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
 import cn.iocoder.yudao.module.iot.core.mq.producer.IotDeviceMessageProducer;
 import cn.iocoder.yudao.module.iot.gateway.enums.IotDeviceTopicEnum;
 import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttUpstreamProtocol;
-import cn.iocoder.yudao.module.iot.gateway.service.message.IotDeviceMessageService;
+import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 
@@ -49,7 +49,7 @@ public class IotMqttEventHandler extends IotMqttAbstractHandler {
             // 使用 IotDeviceMessageService 解码消息
             byte[] messageBytes = payload.getBytes(StandardCharsets.UTF_8);
             IotDeviceMessage message = deviceMessageService.decodeDeviceMessage(
-                    messageBytes, productKey, deviceName, protocol.getServerId());
+                    messageBytes, productKey, deviceName);
 
             // 发送消息
             deviceMessageProducer.sendDeviceMessage(message);

yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttHttpAuthHandler.java

@@ -0,0 +1,192 @@
+package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router;
+
+import cn.hutool.core.util.BooleanUtil;
+import cn.hutool.core.util.StrUtil;
+import cn.hutool.extra.spring.SpringUtil;
+import cn.iocoder.yudao.framework.common.pojo.CommonResult;
+import cn.iocoder.yudao.module.iot.core.biz.IotDeviceCommonApi;
+import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceAuthReqDTO;
+import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
+import cn.iocoder.yudao.module.iot.core.util.IotDeviceAuthUtils;
+import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
+import io.vertx.core.json.JsonObject;
+import io.vertx.ext.web.RoutingContext;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * IoT 网关 MQTT HTTP 认证处理器
+ * <p>
+ * 处理 EMQX 的认证请求和事件钩子
+ *
+ * @author 芋道源码
+ */
+@Slf4j
+public class IotMqttHttpAuthHandler {
+
+    /**
+     * EMQX 认证接口
+     */
+    public void authenticate(RoutingContext context) {
+        try {
+            // 解析请求体
+            JsonObject body = context.body().asJsonObject();
+            if (body == null) {
+                sendErrorResponse(context, 400, "请求体不能为空");
+                return;
+            }
+
+            String clientid = body.getString("clientid");
+            String username = body.getString("username");
+            String password = body.getString("password");
+
+            log.info("[authenticate][EMQX 设备认证请求][clientId: {}][username: {}]", clientid, username);
+
+            // 参数校验
+            if (StrUtil.isEmpty(clientid) || StrUtil.isEmpty(username) || StrUtil.isEmpty(password)) {
+                log.warn("[authenticate][认证参数不完整][clientId: {}][username: {}]", clientid, username);
+                sendErrorResponse(context, 400, "认证参数不完整");
+                return;
+            }
+
+            // 执行设备认证
+            IotDeviceCommonApi deviceApi = SpringUtil.getBean(IotDeviceCommonApi.class);
+            CommonResult<Boolean> result = deviceApi.authDevice(new IotDeviceAuthReqDTO()
+                    .setClientId(clientid)
+                    .setUsername(username)
+                    .setPassword(password));
+
+            result.checkError();
+            if (!BooleanUtil.isTrue(result.getData())) {
+                log.warn("[authenticate][设备认证失败][clientId: {}][username: {}]", clientid, username);
+                sendErrorResponse(context, 401, "设备认证失败");
+                return;
+            }
+
+            log.info("[authenticate][设备认证成功][clientId: {}][username: {}]", clientid, username);
+            sendSuccessResponse(context, "认证成功");
+
+        } catch (Exception e) {
+            log.error("[authenticate][设备认证异常]", e);
+            sendErrorResponse(context, 500, "认证服务异常");
+        }
+    }
+
+    /**
+     * EMQX 客户端连接事件钩子
+     */
+    public void connected(RoutingContext context) {
+        try {
+            // 解析请求体
+            JsonObject body = context.body().asJsonObject();
+            if (body == null) {
+                sendErrorResponse(context, 400, "请求体不能为空");
+                return;
+            }
+
+            String clientid = body.getString("clientid");
+            String username = body.getString("username");
+            Long timestamp = body.getLong("timestamp");
+
+            log.info("[connected][设备连接事件][clientId: {}][username: {}]", clientid, username);
+
+            handleDeviceStateChange(username, true);
+            sendSuccessResponse(context, "处理成功");
+
+        } catch (Exception e) {
+            log.error("[connected][处理设备连接事件失败]", e);
+            sendErrorResponse(context, 500, "处理失败");
+        }
+    }
+
+    /**
+     * EMQX 客户端断开连接事件钩子
+     */
+    public void disconnected(RoutingContext context) {
+        try {
+            // 解析请求体
+            JsonObject body = context.body().asJsonObject();
+            if (body == null) {
+                sendErrorResponse(context, 400, "请求体不能为空");
+                return;
+            }
+
+            String clientid = body.getString("clientid");
+            String username = body.getString("username");
+            String reason = body.getString("reason");
+            Long timestamp = body.getLong("timestamp");
+
+            log.info("[disconnected][设备断开连接事件][clientId: {}][username: {}][reason: {}]",
+                    clientid, username, reason);
+
+            handleDeviceStateChange(username, false);
+            sendSuccessResponse(context, "处理成功");
+
+        } catch (Exception e) {
+            log.error("[disconnected][处理设备断开连接事件失败]", e);
+            sendErrorResponse(context, 500, "处理失败");
+        }
+    }
+
+    /**
+     * 处理设备状态变化
+     *
+     * @param username 用户名
+     * @param online   是否在线
+     */
+    private void handleDeviceStateChange(String username, boolean online) {
+        if (StrUtil.isEmpty(username) || "undefined".equals(username)) {
+            log.warn("[handleDeviceStateChange][用户名为空或未定义][username: {}]", username);
+            return;
+        }
+
+        // 解析设备信息
+        IotDeviceAuthUtils.DeviceInfo deviceInfo = IotDeviceAuthUtils.parseUsername(username);
+        if (deviceInfo == null) {
+            log.warn("[handleDeviceStateChange][无法解析设备信息][username: {}]", username);
+            return;
+        }
+
+        try {
+            // 获取服务器 ID
+            String serverId = "mqtt_auth_gateway";
+
+            // 构建设备状态消息
+            IotDeviceMessageService deviceMessageService = SpringUtil.getBean(IotDeviceMessageService.class);
+            IotDeviceMessage message;
+            if (online) {
+                message = IotDeviceMessage.buildStateOnline();
+                log.info("[handleDeviceStateChange][发送设备上线消息成功][username: {}]", username);
+            } else {
+                message = IotDeviceMessage.buildStateOffline();
+                log.info("[handleDeviceStateChange][发送设备下线消息成功][username: {}]", username);
+            }
+
+            // 发送消息到消息总线
+            deviceMessageService.sendDeviceMessage(message,
+                    deviceInfo.getProductKey(), deviceInfo.getDeviceName(), serverId);
+        } catch (Exception e) {
+            log.error("[handleDeviceStateChange][发送设备状态消息失败][username: {}][online: {}]",
+                    username, online, e);
+        }
+    }
+
+    /**
+     * 发送成功响应
+     */
+    private void sendSuccessResponse(RoutingContext context, String message) {
+        context.response()
+                .setStatusCode(200)
+                .putHeader("Content-Type", "text/plain; charset=utf-8")
+                .end(message);
+    }
+
+    /**
+     * 发送错误响应
+     */
+    private void sendErrorResponse(RoutingContext context, int statusCode, String message) {
+        context.response()
+                .setStatusCode(statusCode)
+                .putHeader("Content-Type", "text/plain; charset=utf-8")
+                .end(message);
+    }
+}

yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttPropertyHandler.java

@@ -6,7 +6,7 @@ import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
 import cn.iocoder.yudao.module.iot.core.mq.producer.IotDeviceMessageProducer;
 import cn.iocoder.yudao.module.iot.gateway.enums.IotDeviceTopicEnum;
 import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttUpstreamProtocol;
-import cn.iocoder.yudao.module.iot.gateway.service.message.IotDeviceMessageService;
+import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 
@@ -65,7 +65,7 @@ public class IotMqttPropertyHandler extends IotMqttAbstractHandler {
             // 使用 IotDeviceMessageService 解码消息
             byte[] messageBytes = payload.getBytes(StandardCharsets.UTF_8);
             IotDeviceMessage message = deviceMessageService.decodeDeviceMessage(
-                    messageBytes, productKey, deviceName, protocol.getServerId());
+                    messageBytes, productKey, deviceName);
 
             // 发送消息
             deviceMessageProducer.sendDeviceMessage(message);
@@ -100,7 +100,7 @@ public class IotMqttPropertyHandler extends IotMqttAbstractHandler {
             // 使用 IotDeviceMessageService 解码消息
             byte[] messageBytes = payload.getBytes(StandardCharsets.UTF_8);
             IotDeviceMessage message = deviceMessageService.decodeDeviceMessage(
-                    messageBytes, productKey, deviceName, protocol.getServerId());
+                    messageBytes, productKey, deviceName);
 
             // 发送消息
             deviceMessageProducer.sendDeviceMessage(message);
@@ -132,7 +132,7 @@ public class IotMqttPropertyHandler extends IotMqttAbstractHandler {
             // 使用 IotDeviceMessageService 解码消息
             byte[] messageBytes = payload.getBytes(StandardCharsets.UTF_8);
             IotDeviceMessage message = deviceMessageService.decodeDeviceMessage(
-                    messageBytes, productKey, deviceName, protocol.getServerId());
+                    messageBytes, productKey, deviceName);
 
             // 发送消息
             deviceMessageProducer.sendDeviceMessage(message);

yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttServiceHandler.java

@@ -6,7 +6,7 @@ import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
 import cn.iocoder.yudao.module.iot.core.mq.producer.IotDeviceMessageProducer;
 import cn.iocoder.yudao.module.iot.gateway.enums.IotDeviceTopicEnum;
 import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttUpstreamProtocol;
-import cn.iocoder.yudao.module.iot.gateway.service.message.IotDeviceMessageService;
+import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 
@@ -49,7 +49,7 @@ public class IotMqttServiceHandler extends IotMqttAbstractHandler {
             // 使用 IotDeviceMessageService 解码消息
             byte[] messageBytes = payload.getBytes(StandardCharsets.UTF_8);
             IotDeviceMessage message = deviceMessageService.decodeDeviceMessage(
-                    messageBytes, productKey, deviceName, protocol.getServerId());
+                    messageBytes, productKey, deviceName);
 
             // 发送消息
             deviceMessageProducer.sendDeviceMessage(message);

yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttUpstreamRouter.java

@@ -5,7 +5,7 @@ import cn.hutool.extra.spring.SpringUtil;
 import cn.iocoder.yudao.module.iot.core.mq.producer.IotDeviceMessageProducer;
 import cn.iocoder.yudao.module.iot.gateway.enums.IotDeviceTopicEnum;
 import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttUpstreamProtocol;
-import cn.iocoder.yudao.module.iot.gateway.service.message.IotDeviceMessageService;
+import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
 import io.vertx.mqtt.messages.MqttPublishMessage;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;

yudao-module-iot/yudao-module-iot-gateway/src/main/resources/application-local.yaml

@@ -1,5 +1,4 @@
-server:
-  port: 8090  # IoT 网关服务端口
+# ==================== IoT 网关本地开发环境配置 ====================
 
 --- #################### 消息队列相关 ####################
 
@@ -26,16 +25,23 @@ yudao:
         # 针对引入的 HTTP 组件的配置
         # ====================================
         http:
+          enabled: true
           server-port: 8092
         # ====================================
         # 针对引入的 EMQX 组件的配置
         # ====================================
         emqx:
+          enabled: true
+          http-auth-port: 8090  # MQTT HTTP 认证服务端口
           mqtt-host: 127.0.0.1                # MQTT Broker 地址
           mqtt-port: 1883                     # MQTT Broker 端口
           mqtt-username: admin                 # MQTT 用户名
           mqtt-password: public                # MQTT 密码
-          auth-port: 8101                      # 认证端口
+          mqtt-ssl: false                      # 是否开启 SSL
+          mqtt-topics:
+            - "/sys/#"                         # 系统主题（设备上报）
+            - "/ota/#"                         # OTA 升级主题
+            - "/config/#"                      # 配置主题
 
     # 消息总线配置
     message-bus:
@@ -50,4 +56,4 @@ logging:
     cn.iocoder.yudao.module.iot.gateway.protocol.mqtt: DEBUG
     cn.iocoder.yudao.module.iot.gateway.protocol.http: DEBUG
     # MQTT 客户端日志
-    io.vertx.mqtt: DEBUG
+    io.vertx.mqtt: DEBUG