topicList = emqxProperties.getMqttTopics();
+ if (CollUtil.isEmpty(topicList)) {
+ log.warn("[subscribeToTopics][未配置 MQTT 主题，使用默认主题]");
+ topicList = List.of("/sys/#"); // 默认订阅所有系统主题
+ }
+
+ for (String topic : topicList) {
+ if (StrUtil.isBlank(topic)) {
+ log.warn("[subscribeToTopics][跳过空主题]");
+ continue;
+ }
+
+ mqttClient.subscribe(topic, DEFAULT_QOS.value())
+ .onSuccess(ack -> log.info("[subscribeToTopics][订阅主题成功: {}]", topic))
+ .onFailure(err -> log.error("[subscribeToTopics][订阅主题失败: {}]", topic, err));
+ }
+ }
+
+ /**
+ * 重连 MQTT 客户端
+ */
+ private void reconnectWithDelay() {
+ if (!isRunning) {
+ log.info("[reconnectWithDelay][服务已停止，不再尝试重连]");
+ return;
+ }
+
+ // 默认重连延迟 5 秒
+ int reconnectDelayMs = 5000;
+ vertx.setTimer(reconnectDelayMs, id -> {
+ log.info("[reconnectWithDelay][开始重新连接 MQTT]");
+ connectMqtt();
+ });
+ }
+
+ /**
+ * 发布消息到 MQTT
+ *
+ * @param topic 主题
+ * @param payload 消息内容
+ */
+ public void publishMessage(String topic, String payload) {
+ if (mqttClient == null || !mqttClient.isConnected()) {
+ log.warn("[publishMessage][MQTT 客户端未连接，无法发送消息][topic: {}]", topic);
+ return;
+ }
+
+ mqttClient.publish(topic, io.vertx.core.buffer.Buffer.buffer(payload), DEFAULT_QOS, false, false)
+ .onSuccess(v -> log.debug("[publishMessage][发送消息成功][topic: {}]", topic))
+ .onFailure(err -> log.error("[publishMessage][发送消息失败][topic: {}]", topic, err));
+ }
+
+ /**
+ * 获取服务器 ID
+ *
+ * @return 服务器 ID
+ */
+ public String getServerId() {
+ return IotDeviceMessageUtils.generateServerId(emqxProperties.getMqttPort());
+ }
+
+ /**
+ * 获取 MQTT 客户端
+ *
+ * @return MQTT 客户端
+ */
+ public MqttClient getMqttClient() {
+ return mqttClient;
+ }
+
+ /**
+ * 获取认证路由器
+ *
+ * @return 认证路由器
+ */
+ public IotMqttAuthRouter getAuthRouter() {
+ return authRouter;
+ }
+
+}
\ No newline at end of file
diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttAbstractHandler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttAbstractHandler.java
new file mode 100644
index 0000000000..66b835bf03
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttAbstractHandler.java
@@ -0,0 +1,38 @@
+package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router;
+
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * IoT 网关 MQTT 协议的路由处理器抽象基类
+ *
+ * 提供通用的处理方法，所有 MQTT 消息处理器都应继承此类
+ *
+ * @author 芋道源码
+ */
+@Slf4j
+public abstract class IotMqttAbstractHandler {
+
+ /**
+ * 处理 MQTT 消息
+ *
+ * @param topic 主题
+ * @param payload 消息内容
+ */
+ public abstract void handle(String topic, String payload);
+
+ /**
+ * 解析主题，获取主题各部分
+ *
+ * @param topic 主题
+ * @return 主题各部分数组，如果解析失败返回 null
+ */
+ protected String[] parseTopic(String topic) {
+ String[] topicParts = topic.split("/");
+ if (topicParts.length < 7) {
+ log.warn("[parseTopic][主题格式不正确][topic: {}]", topic);
+ return null;
+ }
+ return topicParts;
+ }
+
+}
\ No newline at end of file
diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttAuthRouter.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttAuthRouter.java
new file mode 100644
index 0000000000..d1dda7e563
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttAuthRouter.java
@@ -0,0 +1,146 @@
+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 lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * IoT 网关 MQTT 认证路由器
+ *
+ * 处理设备的 MQTT 连接认证和连接状态管理
+ *
+ * @author 芋道源码
+ */
+@RequiredArgsConstructor
+@Slf4j
+public class IotMqttAuthRouter {
+
+ private final IotMqttUpstreamProtocol protocol;
+ private final IotDeviceMessageProducer deviceMessageProducer;
+ private final IotDeviceTokenService deviceTokenService;
+ private final IotDeviceCommonApi deviceCommonApi;
+
+ 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);
+ }
+
+ /**
+ * 处理设备认证
+ *
+ * @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 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 {
+ // 发送设备状态消息
+ IotDeviceMessage message = IotDeviceMessage.of(
+ deviceInfo.getProductKey(), deviceInfo.getDeviceName(), protocol.getServerId());
+
+ if (online) {
+ message = message.ofStateOnline();
+ log.info("[handleDeviceStateChange][发送设备上线消息成功][username: {}]", username);
+ } else {
+ message = message.ofStateOffline();
+ log.info("[handleDeviceStateChange][发送设备下线消息成功][username: {}]", username);
+ }
+
+ deviceMessageProducer.sendDeviceMessage(message);
+ } catch (Exception e) {
+ log.error("[handleDeviceStateChange][发送设备状态消息失败][username: {}][online: {}]", username, online, e);
+ }
+ }
+
+}
\ No newline at end of file
diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttEventHandler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttEventHandler.java
new file mode 100644
index 0000000000..49a77c0778
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttEventHandler.java
@@ -0,0 +1,120 @@
+package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router;
+
+import cn.hutool.json.JSONObject;
+import cn.hutool.json.JSONUtil;
+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 lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.Map;
+
+/**
+ * IoT 网关 MQTT 事件处理器
+ *
+ * 处理设备事件相关的 MQTT 消息
+ *
+ * @author 芋道源码
+ */
+@RequiredArgsConstructor
+@Slf4j
+public class IotMqttEventHandler extends IotMqttAbstractHandler {
+
+ private final IotMqttUpstreamProtocol protocol;
+ private final IotDeviceMessageProducer deviceMessageProducer;
+
+ @Override
+ public void handle(String topic, String payload) {
+ try {
+ log.info("[handle][接收到设备事件上报][topic: {}]", topic);
+
+ // 解析消息内容
+ JSONObject jsonObject = JSONUtil.parseObj(payload);
+ String[] topicParts = parseTopic(topic);
+ if (topicParts == null) {
+ return;
+ }
+
+ // 构建设备消息
+ String productKey = topicParts[2];
+ String deviceName = topicParts[3];
+ String eventIdentifier = getEventIdentifier(topicParts, topic);
+ if (eventIdentifier == null) {
+ return;
+ }
+
+ Map eventData = parseEventDataFromPayload(jsonObject);
+ IotDeviceMessage message = IotDeviceMessage.of(productKey, deviceName, protocol.getServerId());
+ // 设置事件消息类型和标识符
+ message.setType("event");
+ message.setIdentifier(eventIdentifier);
+ message.setData(eventData);
+
+ // 发送消息
+ deviceMessageProducer.sendDeviceMessage(message);
+ log.info("[handle][处理设备事件上报成功][topic: {}]", topic);
+
+ // 发送响应消息
+ String method = "thing.event." + eventIdentifier + ".post";
+ sendResponse(topic, jsonObject, method);
+ } catch (Exception e) {
+ log.error("[handle][处理设备事件上报失败][topic: {}][payload: {}]", topic, payload, e);
+ }
+ }
+
+ /**
+ * 从主题部分中获取事件标识符
+ *
+ * @param topicParts 主题各部分
+ * @param topic 原始主题，用于日志
+ * @return 事件标识符，如果获取失败返回 null
+ */
+ private String getEventIdentifier(String[] topicParts, String topic) {
+ try {
+ return topicParts[6];
+ } catch (ArrayIndexOutOfBoundsException e) {
+ log.warn("[getEventIdentifier][无法从主题中获取事件标识符][topic: {}]", topic);
+ return null;
+ }
+ }
+
+ /**
+ * 从消息载荷解析事件数据
+ *
+ * @param jsonObject 消息 JSON 对象
+ * @return 事件数据映射
+ */
+ private Map parseEventDataFromPayload(JSONObject jsonObject) {
+ JSONObject params = jsonObject.getJSONObject("params");
+ if (params == null) {
+ log.warn("[parseEventDataFromPayload][消息格式不正确，缺少 params 字段][jsonObject: {}]", jsonObject);
+ return Map.of();
+ }
+ return params;
+ }
+
+ /**
+ * 发送响应消息
+ *
+ * @param topic 原始主题
+ * @param jsonObject 原始消息 JSON 对象
+ * @param method 响应方法
+ */
+ private void sendResponse(String topic, JSONObject jsonObject, String method) {
+ String replyTopic = IotDeviceTopicEnum.getReplyTopic(topic);
+
+ // 构建响应消息
+ JSONObject response = new JSONObject();
+ response.set("id", jsonObject.getStr("id"));
+ response.set("code", 200);
+ response.set("method", method);
+ response.set("data", new JSONObject());
+
+ // 发送响应
+ protocol.publishMessage(replyTopic, response.toString());
+ log.debug("[sendResponse][发送响应消息][topic: {}]", replyTopic);
+ }
+
+}
\ No newline at end of file
diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttPropertyHandler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttPropertyHandler.java
new file mode 100644
index 0000000000..95eccb1aae
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttPropertyHandler.java
@@ -0,0 +1,147 @@
+package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router;
+
+import cn.hutool.json.JSONObject;
+import cn.hutool.json.JSONUtil;
+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 lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.Map;
+
+/**
+ * IoT 网关 MQTT 属性处理器
+ *
+ * 处理设备属性相关的 MQTT 消息
+ *
+ * @author 芋道源码
+ */
+@RequiredArgsConstructor
+@Slf4j
+public class IotMqttPropertyHandler extends IotMqttAbstractHandler {
+
+ private final IotMqttUpstreamProtocol protocol;
+ private final IotDeviceMessageProducer deviceMessageProducer;
+
+ @Override
+ public void handle(String topic, String payload) {
+ if (topic.endsWith(IotDeviceTopicEnum.PROPERTY_POST_TOPIC.getTopic())) {
+ // 属性上报
+ handlePropertyPost(topic, payload);
+ } else if (topic.contains(IotDeviceTopicEnum.PROPERTY_SET_TOPIC.getTopic())) {
+ // 属性设置响应
+ handlePropertySetReply(topic, payload);
+ } else if (topic.contains(IotDeviceTopicEnum.PROPERTY_GET_TOPIC.getTopic())) {
+ // 属性获取响应
+ handlePropertyGetReply(topic, payload);
+ } else {
+ log.warn("[handle][未知的属性主题][topic: {}]", topic);
+ }
+ }
+
+ /**
+ * 处理设备属性上报消息
+ *
+ * @param topic 主题
+ * @param payload 消息内容
+ */
+ private void handlePropertyPost(String topic, String payload) {
+ try {
+ log.info("[handlePropertyPost][接收到设备属性上报][topic: {}]", topic);
+
+ // 解析消息内容
+ JSONObject jsonObject = JSONUtil.parseObj(payload);
+ String[] topicParts = parseTopic(topic);
+ if (topicParts == null) {
+ return;
+ }
+
+ // 构建设备消息
+ String productKey = topicParts[2];
+ String deviceName = topicParts[3];
+ Map properties = parsePropertiesFromPayload(jsonObject);
+
+ IotDeviceMessage message = IotDeviceMessage.of(productKey, deviceName, protocol.getServerId())
+ .ofPropertyReport(properties);
+
+ // 发送消息
+ deviceMessageProducer.sendDeviceMessage(message);
+ log.info("[handlePropertyPost][处理设备属性上报成功][topic: {}]", topic);
+
+ // 发送响应消息
+ sendResponse(topic, jsonObject, "thing.event.property.post");
+ } catch (Exception e) {
+ log.error("[handlePropertyPost][处理设备属性上报失败][topic: {}][payload: {}]", topic, payload, e);
+ }
+ }
+
+ /**
+ * 处理属性设置响应消息
+ *
+ * @param topic 主题
+ * @param payload 消息内容
+ */
+ private void handlePropertySetReply(String topic, String payload) {
+ try {
+ log.info("[handlePropertySetReply][接收到属性设置响应][topic: {}]", topic);
+ // TODO: 处理属性设置响应逻辑
+ } catch (Exception e) {
+ log.error("[handlePropertySetReply][处理属性设置响应失败][topic: {}][payload: {}]", topic, payload, e);
+ }
+ }
+
+ /**
+ * 处理属性获取响应消息
+ *
+ * @param topic 主题
+ * @param payload 消息内容
+ */
+ private void handlePropertyGetReply(String topic, String payload) {
+ try {
+ log.info("[handlePropertyGetReply][接收到属性获取响应][topic: {}]", topic);
+ // TODO: 处理属性获取响应逻辑
+ } catch (Exception e) {
+ log.error("[handlePropertyGetReply][处理属性获取响应失败][topic: {}][payload: {}]", topic, payload, e);
+ }
+ }
+
+ /**
+ * 从消息载荷解析属性
+ *
+ * @param jsonObject 消息 JSON 对象
+ * @return 属性映射
+ */
+ private Map parsePropertiesFromPayload(JSONObject jsonObject) {
+ JSONObject params = jsonObject.getJSONObject("params");
+ if (params == null) {
+ log.warn("[parsePropertiesFromPayload][消息格式不正确，缺少 params 字段][jsonObject: {}]", jsonObject);
+ return Map.of();
+ }
+ return params;
+ }
+
+ /**
+ * 发送响应消息
+ *
+ * @param topic 原始主题
+ * @param jsonObject 原始消息 JSON 对象
+ * @param method 响应方法
+ */
+ private void sendResponse(String topic, JSONObject jsonObject, String method) {
+ String replyTopic = IotDeviceTopicEnum.getReplyTopic(topic);
+
+ // 构建响应消息
+ JSONObject response = new JSONObject();
+ response.set("id", jsonObject.getStr("id"));
+ response.set("code", 200);
+ response.set("method", method);
+ response.set("data", new JSONObject());
+
+ // 发送响应
+ protocol.publishMessage(replyTopic, response.toString());
+ log.debug("[sendResponse][发送响应消息][topic: {}]", replyTopic);
+ }
+
+}
\ No newline at end of file
diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttServiceHandler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttServiceHandler.java
new file mode 100644
index 0000000000..0a08f0c9e5
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttServiceHandler.java
@@ -0,0 +1,121 @@
+package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router;
+
+import cn.hutool.json.JSONObject;
+import cn.hutool.json.JSONUtil;
+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 lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.Map;
+
+/**
+ * IoT 网关 MQTT 服务处理器
+ *
+ * 处理设备服务调用相关的 MQTT 消息
+ *
+ * @author 芋道源码
+ */
+@RequiredArgsConstructor
+@Slf4j
+public class IotMqttServiceHandler extends IotMqttAbstractHandler {
+
+ private final IotMqttUpstreamProtocol protocol;
+ private final IotDeviceMessageProducer deviceMessageProducer;
+
+ @Override
+ public void handle(String topic, String payload) {
+ try {
+ log.info("[handle][接收到设备服务调用响应][topic: {}]", topic);
+
+ // 解析消息内容
+ JSONObject jsonObject = JSONUtil.parseObj(payload);
+ String[] topicParts = parseTopic(topic);
+ if (topicParts == null) {
+ return;
+ }
+
+ // 构建设备消息
+ String productKey = topicParts[2];
+ String deviceName = topicParts[3];
+ String serviceIdentifier = getServiceIdentifier(topicParts, topic);
+ if (serviceIdentifier == null) {
+ return;
+ }
+
+ Map serviceData = parseServiceDataFromPayload(jsonObject);
+ IotDeviceMessage message = IotDeviceMessage.of(productKey, deviceName, protocol.getServerId());
+ // 设置服务消息类型和标识符
+ message.setType("service");
+ message.setIdentifier(serviceIdentifier);
+ message.setData(serviceData);
+
+ // 发送消息
+ deviceMessageProducer.sendDeviceMessage(message);
+ log.info("[handle][处理设备服务调用响应成功][topic: {}]", topic);
+
+ // 发送响应消息
+ String method = "thing.service." + serviceIdentifier;
+ sendResponse(topic, jsonObject, method);
+ } catch (Exception e) {
+ log.error("[handle][处理设备服务调用响应失败][topic: {}][payload: {}]", topic, payload, e);
+ }
+ }
+
+ /**
+ * 从主题部分中获取服务标识符
+ *
+ * @param topicParts 主题各部分
+ * @param topic 原始主题，用于日志
+ * @return 服务标识符，如果获取失败返回 null
+ */
+ private String getServiceIdentifier(String[] topicParts, String topic) {
+ try {
+ // 服务主题格式：/sys/{productKey}/{deviceName}/thing/service/{serviceIdentifier}
+ return topicParts[6];
+ } catch (ArrayIndexOutOfBoundsException e) {
+ log.warn("[getServiceIdentifier][无法从主题中获取服务标识符][topic: {}]", topic);
+ return null;
+ }
+ }
+
+ /**
+ * 从消息载荷解析服务数据
+ *
+ * @param jsonObject 消息 JSON 对象
+ * @return 服务数据映射
+ */
+ private Map parseServiceDataFromPayload(JSONObject jsonObject) {
+ JSONObject params = jsonObject.getJSONObject("params");
+ if (params == null) {
+ log.warn("[parseServiceDataFromPayload][消息格式不正确，缺少 params 字段][jsonObject: {}]", jsonObject);
+ return Map.of();
+ }
+ return params;
+ }
+
+ /**
+ * 发送响应消息
+ *
+ * @param topic 原始主题
+ * @param jsonObject 原始消息 JSON 对象
+ * @param method 响应方法
+ */
+ private void sendResponse(String topic, JSONObject jsonObject, String method) {
+ String replyTopic = IotDeviceTopicEnum.getReplyTopic(topic);
+
+ // 构建响应消息
+ JSONObject response = new JSONObject();
+ response.set("id", jsonObject.getStr("id"));
+ response.set("code", 200);
+ response.set("method", method);
+ response.set("data", new JSONObject());
+
+ // 发送响应
+ protocol.publishMessage(replyTopic, response.toString());
+ log.debug("[sendResponse][发送响应消息][topic: {}]", replyTopic);
+ }
+
+}
\ No newline at end of file
diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttUpstreamRouter.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttUpstreamRouter.java
new file mode 100644
index 0000000000..c4b37ad148
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttUpstreamRouter.java
@@ -0,0 +1,105 @@
+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.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 io.vertx.mqtt.messages.MqttPublishMessage;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * IoT 网关 MQTT 上行路由器
+ *
+ * 根据消息主题路由到不同的处理器
+ *
+ * @author 芋道源码
+ */
+@RequiredArgsConstructor
+@Slf4j
+public class IotMqttUpstreamRouter {
+
+ private final IotMqttUpstreamProtocol protocol;
+ private final IotDeviceMessageProducer deviceMessageProducer;
+
+ // 处理器实例
+ private IotMqttPropertyHandler propertyHandler;
+ private IotMqttEventHandler eventHandler;
+ private IotMqttServiceHandler serviceHandler;
+
+ public IotMqttUpstreamRouter(IotMqttUpstreamProtocol protocol) {
+ this.protocol = protocol;
+ this.deviceMessageProducer = SpringUtil.getBean(IotDeviceMessageProducer.class);
+ // 初始化处理器
+ this.propertyHandler = new IotMqttPropertyHandler(protocol, deviceMessageProducer);
+ this.eventHandler = new IotMqttEventHandler(protocol, deviceMessageProducer);
+ this.serviceHandler = new IotMqttServiceHandler(protocol, deviceMessageProducer);
+ }
+
+ /**
+ * 路由 MQTT 消息
+ *
+ * @param message MQTT 发布消息
+ */
+ public void route(MqttPublishMessage message) {
+ String topic = message.topicName();
+ String payload = message.payload().toString();
+ log.info("[route][接收到 MQTT 消息][topic: {}][payload: {}]", topic, payload);
+
+ try {
+ if (StrUtil.isEmpty(payload)) {
+ log.warn("[route][消息内容为空][topic: {}]", topic);
+ return;
+ }
+
+ // 根据主题路由到相应的处理器
+ if (isPropertyTopic(topic)) {
+ propertyHandler.handle(topic, payload);
+ } else if (isEventTopic(topic)) {
+ eventHandler.handle(topic, payload);
+ } else if (isServiceTopic(topic)) {
+ serviceHandler.handle(topic, payload);
+ } else {
+ log.warn("[route][未知的消息类型][topic: {}]", topic);
+ }
+ } catch (Exception e) {
+ log.error("[route][处理 MQTT 消息失败][topic: {}][payload: {}]", topic, payload, e);
+ }
+ }
+
+ /**
+ * 判断是否为属性相关主题
+ *
+ * @param topic 主题
+ * @return 是否为属性主题
+ */
+ private boolean isPropertyTopic(String topic) {
+ return topic.endsWith(IotDeviceTopicEnum.PROPERTY_POST_TOPIC.getTopic()) ||
+ topic.contains(IotDeviceTopicEnum.PROPERTY_SET_TOPIC.getTopic()) ||
+ topic.contains(IotDeviceTopicEnum.PROPERTY_GET_TOPIC.getTopic());
+ }
+
+ /**
+ * 判断是否为事件相关主题
+ *
+ * @param topic 主题
+ * @return 是否为事件主题
+ */
+ private boolean isEventTopic(String topic) {
+ return topic.contains(IotDeviceTopicEnum.EVENT_POST_TOPIC_PREFIX.getTopic()) &&
+ topic.endsWith(IotDeviceTopicEnum.EVENT_POST_TOPIC_SUFFIX.getTopic());
+ }
+
+ /**
+ * 判断是否为服务相关主题
+ *
+ * @param topic 主题
+ * @return 是否为服务主题
+ */
+ private boolean isServiceTopic(String topic) {
+ return topic.contains(IotDeviceTopicEnum.SERVICE_TOPIC_PREFIX.getTopic()) &&
+ !isPropertyTopic(topic); // 排除属性相关的服务调用
+ }
+
+}
\ No newline at end of file
diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/package-info.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/package-info.java
new file mode 100644
index 0000000000..57d68d7497
--- /dev/null
+++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/package-info.java
@@ -0,0 +1,22 @@
+/**
+ * MQTT 协议路由器包
+ *
+ * 包含 MQTT 协议的所有路由处理器和抽象基类：
+ *
+ * - {@link cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router.IotMqttAbstractHandler}
+ * - 抽象路由处理器基类
+ * - {@link cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router.IotMqttUpstreamRouter}
+ * - 上行消息路由器
+ * - {@link cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router.IotMqttAuthRouter}
+ * - 认证路由器
+ * - {@link cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router.IotMqttPropertyHandler}
+ * - 属性处理器
+ * - {@link cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router.IotMqttEventHandler}
+ * - 事件处理器
+ * - {@link cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router.IotMqttServiceHandler}
+ * - 服务处理器
+ *
+ *
+ * @author 芋道源码
+ */
+package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router;
\ No newline at end of file
diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/resources/application.yaml b/yudao-module-iot/yudao-module-iot-gateway/src/main/resources/application.yaml
index b57cc266d1..678569f807 100644
--- a/yudao-module-iot/yudao-module-iot-gateway/src/main/resources/application.yaml
+++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/resources/application.yaml
@@ -37,7 +37,7 @@ yudao:
# 针对引入的 EMQX 组件的配置
# ====================================
emqx:
- enabled: false
+ enabled: true
mqtt-ssl: false
mqtt-topics:
- "/sys/#" # 系统主题（设备上报）