diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/api/device/IoTDeviceApiImpl.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/api/device/IoTDeviceApiImpl.java
index 6bfe0e4..7922eb1 100644
--- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/api/device/IoTDeviceApiImpl.java
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/api/device/IoTDeviceApiImpl.java
@@ -22,6 +22,7 @@ import cn.iocoder.yudao.module.iot.service.device.IotDeviceService;
import cn.iocoder.yudao.module.iot.service.product.IotProductService;
import jakarta.annotation.Resource;
import jakarta.annotation.security.PermitAll;
+import org.springframework.context.annotation.Lazy;
import org.springframework.context.annotation.Primary;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.PostMapping;
@@ -51,8 +52,10 @@ public class IoTDeviceApiImpl implements IotDeviceCommonApi {
@Resource
private IotProductService productService;
@Resource
+ @Lazy // 延迟加载，解决循环依赖
private IotDeviceModbusConfigService modbusConfigService;
@Resource
+ @Lazy // 延迟加载，解决循环依赖
private IotDeviceModbusPointService modbusPointService;
@Override
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/product/IotProductServiceImpl.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/product/IotProductServiceImpl.java
index 9686f42..49d7f36 100644
--- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/product/IotProductServiceImpl.java
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/product/IotProductServiceImpl.java
@@ -17,6 +17,7 @@ import com.baomidou.dynamic.datasource.annotation.DSTransactional;
import jakarta.annotation.Resource;
import org.springframework.cache.annotation.CacheEvict;
import org.springframework.cache.annotation.Cacheable;
+import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;
import org.springframework.validation.annotation.Validated;
@@ -41,9 +42,10 @@ public class IotProductServiceImpl implements IotProductService {
private IotProductMapper productMapper;
@Resource
- private IotDevicePropertyService devicePropertyDataService;
- @Resource
private IotDeviceService deviceService;
+ @Resource
+ @Lazy // 延迟加载，避免循环依赖
+ private IotDevicePropertyService devicePropertyDataService;
@Override
public Long createProduct(IotProductSaveReqVO createReqVO) {
diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/thingmodel/IotThingModelServiceImpl.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/thingmodel/IotThingModelServiceImpl.java
index 18a5b62..4a8b974 100644
--- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/thingmodel/IotThingModelServiceImpl.java
+++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/thingmodel/IotThingModelServiceImpl.java
@@ -51,8 +51,8 @@ public class IotThingModelServiceImpl implements IotThingModelService {
@Resource
@Lazy // 延迟加载，解决循环依赖
private IotProductService productService;
-
@Resource
+ @Lazy // 延迟加载，解决循环依赖
private IotDeviceModbusPointService deviceModbusPointService;
@Override
diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/upstream/IotEmqxUpstreamHandler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/upstream/IotEmqxUpstreamHandler.java
index 17d5f85..45c9481 100644
--- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/upstream/IotEmqxUpstreamHandler.java
+++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/handler/upstream/IotEmqxUpstreamHandler.java
@@ -5,6 +5,7 @@ import cn.hutool.core.util.StrUtil;
import cn.hutool.extra.spring.SpringUtil;
import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
+import cn.iocoder.yudao.module.iot.gateway.util.IotMqttTopicUtils;
import io.vertx.mqtt.messages.MqttPublishMessage;
import lombok.extern.slf4j.Slf4j;
@@ -42,12 +43,14 @@ public class IotEmqxUpstreamHandler {
return;
}
- // 2. 反序列化消息
+ // 2.1 反序列化消息
IotDeviceMessage message = deviceMessageService.deserializeDeviceMessage(payload, productKey, deviceName);
if (message == null) {
log.warn("[handle][topic({}) payload({}) 消息解码失败]", topic, new String(payload));
return;
}
+ // 2.2 标准化回复消息的 method（MQTT 协议中，设备回复消息的 method 会携带 _reply 后缀）
+ IotMqttTopicUtils.normalizeReplyMethod(message);
// 3. 发送消息到队列
deviceMessageService.sendDeviceMessage(message, productKey, deviceName, serverId);
diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/handler/upstream/IotMqttUpstreamHandler.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/handler/upstream/IotMqttUpstreamHandler.java
index 6b452e4..81a6ece 100644
--- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/handler/upstream/IotMqttUpstreamHandler.java
+++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/handler/upstream/IotMqttUpstreamHandler.java
@@ -2,7 +2,6 @@ package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.handler.upstream;
import cn.hutool.core.lang.Assert;
import cn.hutool.core.util.ArrayUtil;
-import cn.iocoder.yudao.framework.common.exception.ServiceException;
import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.manager.IotMqttConnectionManager;
import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
@@ -10,8 +9,6 @@ import cn.iocoder.yudao.module.iot.gateway.util.IotMqttTopicUtils;
import io.vertx.mqtt.MqttEndpoint;
import lombok.extern.slf4j.Slf4j;
-import static cn.iocoder.yudao.framework.common.exception.enums.GlobalErrorCodeConstants.BAD_REQUEST;
-import static cn.iocoder.yudao.framework.common.exception.enums.GlobalErrorCodeConstants.INTERNAL_SERVER_ERROR;
/**
* IoT 网关 MQTT 上行消息处理器：处理业务消息（属性上报、事件上报等）
@@ -39,10 +36,6 @@ public class IotMqttUpstreamHandler extends IotMqttAbstractHandler {
*/
public void handleBusinessRequest(MqttEndpoint endpoint, String topic, byte[] payload) {
String clientId = endpoint.clientIdentifier();
- IotDeviceMessage message = null;
- String productKey = null;
- String deviceName = null;
-
try {
// 1.1 基础检查
if (ArrayUtil.isEmpty(payload)) {
@@ -50,8 +43,8 @@ public class IotMqttUpstreamHandler extends IotMqttAbstractHandler {
}
// 1.2 解析主题，获取 productKey 和 deviceName
String[] topicParts = topic.split("/");
- productKey = ArrayUtil.get(topicParts, 2);
- deviceName = ArrayUtil.get(topicParts, 3);
+ String productKey = ArrayUtil.get(topicParts, 2);
+ String deviceName = ArrayUtil.get(topicParts, 3);
Assert.notBlank(productKey, "产品 Key 不能为空");
Assert.notBlank(deviceName, "设备名称不能为空");
// 1.3 校验设备信息，防止伪造设备消息
@@ -65,38 +58,21 @@ public class IotMqttUpstreamHandler extends IotMqttAbstractHandler {
return;
}
- // 2. 反序列化消息
- message = deviceMessageService.deserializeDeviceMessage(payload, productKey, deviceName);
+ // 2.1 反序列化消息
+ IotDeviceMessage message = deviceMessageService.deserializeDeviceMessage(payload, productKey, deviceName);
if (message == null) {
log.warn("[handleBusinessRequest][消息解码失败，客户端 ID: {}，主题: {}]", clientId, topic);
- sendErrorResponse(endpoint, productKey, deviceName, null, null,
- BAD_REQUEST.getCode(), "消息解码失败");
return;
}
+ // 2.2 标准化回复消息的 method（MQTT 协议中，设备回复消息的 method 会携带 _reply 后缀）
+ IotMqttTopicUtils.normalizeReplyMethod(message);
// 3. 处理业务消息
deviceMessageService.sendDeviceMessage(message, productKey, deviceName, serverId);
log.debug("[handleBusinessRequest][消息处理成功，客户端 ID: {}，主题: {}]", clientId, topic);
- } catch (ServiceException e) {
- log.warn("[handleBusinessRequest][业务异常，客户端 ID: {}，主题: {}，错误: {}]",
- clientId, topic, e.getMessage());
- String requestId = message != null ? message.getRequestId() : null;
- String method = message != null ? message.getMethod() : null;
- sendErrorResponse(endpoint, productKey, deviceName, requestId, method, e.getCode(), e.getMessage());
- } catch (IllegalArgumentException e) {
- log.warn("[handleBusinessRequest][参数校验失败，客户端 ID: {}，主题: {}，错误: {}]",
- clientId, topic, e.getMessage());
- String requestId = message != null ? message.getRequestId() : null;
- String method = message != null ? message.getMethod() : null;
- sendErrorResponse(endpoint, productKey, deviceName, requestId, method,
- BAD_REQUEST.getCode(), e.getMessage());
} catch (Exception e) {
log.error("[handleBusinessRequest][消息处理异常，客户端 ID: {}，主题: {}，错误: {}]",
clientId, topic, e.getMessage(), e);
- String requestId = message != null ? message.getRequestId() : null;
- String method = message != null ? message.getMethod() : null;
- sendErrorResponse(endpoint, productKey, deviceName, requestId, method,
- INTERNAL_SERVER_ERROR.getCode(), INTERNAL_SERVER_ERROR.getMsg());
}
}
diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/util/IotMqttTopicUtils.java b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/util/IotMqttTopicUtils.java
index e72597d..60e8d6c 100644
--- a/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/util/IotMqttTopicUtils.java
+++ b/yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/util/IotMqttTopicUtils.java
@@ -1,6 +1,8 @@
package cn.iocoder.yudao.module.iot.gateway.util;
import cn.hutool.core.util.StrUtil;
+import cn.iocoder.yudao.framework.common.exception.enums.GlobalErrorCodeConstants;
+import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
/**
* IoT 网关 MQTT 主题工具类
@@ -44,6 +46,32 @@ public final class IotMqttTopicUtils {
*/
public static final String MQTT_ACL_PATH = "/mqtt/acl";
+ // ========== 消息方法标准化 ==========
+
+ /**
+ * 标准化设备回复消息的 method
+ *
+ * MQTT 协议中，设备回复下行指令时，topic 和 method 会携带 _reply 后缀
+ * （如 thing.service.invoke_reply）。平台内部统一使用基础 method（如 thing.service.invoke），
+ * 通过 {@link IotDeviceMessage#getCode()} 非空来识别回复消息。
+ *
+ * 此方法剥离 _reply 后缀，并确保 code 字段被设置。
+ *
+ * @param message 设备消息
+ */
+ public static void normalizeReplyMethod(IotDeviceMessage message) {
+ String method = message.getMethod();
+ if (!StrUtil.endWith(method, REPLY_TOPIC_SUFFIX)) {
+ return;
+ }
+ // 1. 剥离 _reply 后缀
+ message.setMethod(method.substring(0, method.length() - REPLY_TOPIC_SUFFIX.length()));
+ // 2. 确保 code 被设置，使 isReplyMessage() 能正确识别
+ if (message.getCode() == null) {
+ message.setCode(GlobalErrorCodeConstants.SUCCESS.getCode());
+ }
+ }
+
// ========== 工具方法 ==========
/**
diff --git a/yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/IotDirectDeviceMqttProtocolIntegrationTest.java b/yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/IotDirectDeviceMqttProtocolIntegrationTest.java
index 45cb7ca..3333af6 100644
--- a/yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/IotDirectDeviceMqttProtocolIntegrationTest.java
+++ b/yudao-module-iot/yudao-module-iot-gateway/src/test/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/IotDirectDeviceMqttProtocolIntegrationTest.java
@@ -239,12 +239,28 @@ public class IotDirectDeviceMqttProtocolIntegrationTest {
log.info("[testSubscribe][连接认证成功]");
try {
- // 2. 设置消息处理器
- client.publishHandler(message -> log.info("[testSubscribe][收到消息: topic={}, payload={}]",
- message.topicName(), message.payload().toString()));
+ // 2. 设置消息处理器：收到属性设置时，回复 _reply 消息
+ client.publishHandler(message -> {
+ log.info("[testSubscribe][收到消息: topic={}, payload={}]",
+ message.topicName(), message.payload().toString());
+ // 收到属性设置消息时，回复 _reply
+ if (message.topicName().endsWith("/thing/property/set")) {
+ try {
+ IotDeviceMessage received = SERIALIZER.deserialize(message.payload().getBytes());
+ IotDeviceMessage reply = IotDeviceMessage.replyOf(
+ received.getRequestId(), "thing.property.set_reply", null, 0, null);
+ String replyTopic = String.format("/sys/%s/%s/thing/property/set_reply", PRODUCT_KEY, DEVICE_NAME);
+ byte[] replyPayload = SERIALIZER.serialize(reply);
+ client.publish(replyTopic, Buffer.buffer(replyPayload), MqttQoS.AT_LEAST_ONCE, false, false);
+ log.info("[testSubscribe][已回复属性设置: topic={}]", replyTopic);
+ } catch (Exception e) {
+ log.error("[testSubscribe][回复属性设置异常]", e);
+ }
+ }
+ });
- // 3. 订阅下行主题
- String topic = String.format("/sys/%s/%s/thing/service/#", PRODUCT_KEY, DEVICE_NAME);
+ // 3. 订阅下行主题（属性设置 + 服务调用）
+ String topic = String.format("/sys/%s/%s/#", PRODUCT_KEY, DEVICE_NAME);
log.info("[testSubscribe][订阅主题: {}]", topic);
subscribe(client, topic);
log.info("[testSubscribe][订阅成功，等待下行消息... (30秒后自动断开)]");
diff --git a/yudao-module-mp/src/main/java/cn/iocoder/yudao/module/mp/service/account/MpAccountServiceImpl.java b/yudao-module-mp/src/main/java/cn/iocoder/yudao/module/mp/service/account/MpAccountServiceImpl.java
index 14e91b6..6f59f8d 100644
--- a/yudao-module-mp/src/main/java/cn/iocoder/yudao/module/mp/service/account/MpAccountServiceImpl.java
+++ b/yudao-module-mp/src/main/java/cn/iocoder/yudao/module/mp/service/account/MpAccountServiceImpl.java
@@ -60,7 +60,7 @@ public class MpAccountServiceImpl implements MpAccountService {
private MpAccountMapper mpAccountMapper;
@Resource
- @Lazy // 延迟加载，解决循环依赖的问题
+ @Lazy // 延迟加载，解决循环依赖
private MpServiceFactory mpServiceFactory;
@Override