BJ_ZTEITS-Developers / parking-cloud-yudao

Commit 569eef4a743610606762e3ffd9f077a4d320ef25

Authored by haohao
1 parent b06556da

feat：【IoT 物联网】更新设备消息处理逻辑，重构 MQTT 下行消息处理器，优化主题构建和消息发布流程

Inline Side-by-side
Showing 13 changed files with 344 additions and 749 deletions
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/mq/consumer/device/IotDeviceMessageSubscriber.java
  View file @569eef4
... ... @@ -22,7 +22,7 @@ import java.time.LocalDateTime;
22 22  
23 23 /**
24 24 * 针对 {@link IotDeviceMessage} 的业务处理器：调用 method 对应的逻辑。例如说：
25   - * 1. {@link IotDeviceMessageMethodEnum#PROPERTY_REPORT} 属性上报时，记录设备属性
  25 + * 1. {@link IotDeviceMessageMethodEnum#PROPERTY_POST} 属性上报时，记录设备属性
26 26 *
27 27 * @author alwayssuper
28 28 */
... ...
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/message/IotDeviceMessageServiceImpl.java
  View file @569eef4
... ... @@ -126,7 +126,7 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService {
126 126  
127 127 @Override
128 128 public void handleUpstreamDeviceMessage(IotDeviceMessage message, IotDeviceDO device) {
129   - // 1. 理消息
  129 + // 1. 处理消息
130 130 Object replyData = null;
131 131 ServiceException serviceException = null;
132 132 try {
... ... @@ -175,8 +175,7 @@ public class IotDeviceMessageServiceImpl implements IotDeviceMessageService {
175 175 }
176 176  
177 177 // 属性上报
178   - if (Objects.equal(message.getMethod(), IotDeviceMessageMethodEnum.PROPERTY_REPORT.getMethod()) ||
179   - Objects.equal(message.getMethod(), IotDeviceMessageMethodEnum.PROPERTY_POST.getMethod())) {
  178 + if (Objects.equal(message.getMethod(), IotDeviceMessageMethodEnum.PROPERTY_POST.getMethod())) {
180 179 devicePropertyService.saveDeviceProperty(device, message);
181 180 return null;
182 181 }
... ...
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/enums/IotDeviceMessageMethodEnum.java
  View file @569eef4
... ... @@ -23,15 +23,13 @@ public enum IotDeviceMessageMethodEnum implements ArrayValuable<String> {
23 23 STATE_OFFLINE("thing.state.offline", true),
24 24  
25 25 // ========== 设备属性 ==========
26   - // 可参考 https://help.aliyun.com/zh/iot/user-guide/device-properties-events-and-services
27   - // TODO @haohao：使用 report 哈；
  26 + // 可参考
  27 + // https://help.aliyun.com/zh/iot/user-guide/device-properties-events-and-services
28 28 PROPERTY_POST("thing.property.post", true),
29   - PROPERTY_REPORT("thing.property.report", true),
30   -
31 29 PROPERTY_SET("thing.property.set", false),
32   - PROPERTY_GET("thing.property.get", false),
33 30  
34   - //
  31 + // ========== 设备事件 ==========
  32 + EVENT_POST("thing.event.post", true),
35 33  
36 34 ;
37 35  
... ...
yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/mq/message/IotDeviceMessage.java
  View file @569eef4
... ... @@ -25,9 +25,9 @@ public class IotDeviceMessage {
25 25 public static final String MESSAGE_BUS_DEVICE_MESSAGE_TOPIC = "iot_device_message";
26 26  
27 27 /**
28   - * 【消息总线】设备消息 Topic，由 iot-biz 发送给 iot-gateway 的某个 “server”(protocol) 进行消费
  28 + * 【消息总线】设备消息 Topic，由 iot-biz 发送给 iot-gateway 的某个 "server"(protocol) 进行消费
29 29 *
30   - * 其中，%s 就是该“server”(protocol) 的标识
  30 + * 其中，%s 就是该"server"(protocol) 的标识
31 31 */
32 32 public static final String MESSAGE_BUS_GATEWAY_DEVICE_MESSAGE_TOPIC = MESSAGE_BUS_DEVICE_MESSAGE_TOPIC + "_%s";
33 33  
... ... @@ -92,7 +92,7 @@ public class IotDeviceMessage {
92 92 */
93 93 private String msg;
94 94  
95   - // ========== 基础方法：只传递“codec（编解码）字段” ==========
  95 + // ========== 基础方法：只传递"codec（编解码）字段" ==========
96 96  
97 97 public static IotDeviceMessage requestOf(String method) {
98 98 return requestOf(null, method, null);
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/config/IotGatewayConfiguration.java
  View file @569eef4
... ... @@ -45,7 +45,7 @@ public class IotGatewayConfiguration {
45 45 public static class MqttProtocolConfiguration {
46 46  
47 47 @Bean
48   - public IotMqttUpstreamProtocol iotMqttUnifiedProtocol(IotGatewayProperties gatewayProperties) {
  48 + public IotMqttUpstreamProtocol iotMqttUpstreamProtocol(IotGatewayProperties gatewayProperties) {
49 49 return new IotMqttUpstreamProtocol(gatewayProperties.getProtocol().getEmqx());
50 50 }
51 51  
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/config/IotGatewayProperties.java
  View file @569eef4
... ... @@ -134,7 +134,7 @@ public class IotGatewayProperties {
134 134 private String mqttPassword;
135 135  
136 136 /**
137   - * MQTT 是否开启 SSL（默认：false）
  137 + * MQTT 客户端的 SSL 开关
138 138 */
139 139 @NotNull(message = "MQTT 是否开启 SSL 不能为空")
140 140 private Boolean mqttSsl = false;
... ... @@ -145,12 +145,31 @@ public class IotGatewayProperties {
145 145 private String mqttClientId;
146 146  
147 147 /**
148   - * MQTT 主题列表
  148 + * MQTT 订阅的主题
149 149 */
150 150 @NotEmpty(message = "MQTT 主题不能为空")
151 151 private List<String> mqttTopics;
152 152  
153 153 /**
  154 + * 默认 QoS 级别
  155 + * <p>
  156 + * 0 - 最多一次
  157 + * 1 - 至少一次
  158 + * 2 - 刚好一次
  159 + */
  160 + private Integer mqttQos = 1;
  161 +
  162 + /**
  163 + * 连接超时时间（秒）
  164 + */
  165 + private Integer connectTimeoutSeconds = 10;
  166 +
  167 + /**
  168 + * 重连延迟时间（毫秒）
  169 + */
  170 + private Long reconnectDelayMs = 5000L;
  171 +
  172 + /**
154 173 * 获取 MQTT 客户端 ID，如果未配置则自动生成
155 174 *
156 175 * @return MQTT 客户端 ID
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/enums/IotDeviceTopicEnum.java deleted
View file @b06556d
1   -package cn.iocoder.yudao.module.iot.gateway.enums;
2   -
3   -import lombok.Getter;
4   -import lombok.RequiredArgsConstructor;
5   -
6   -// TODO @haohao：这个类，我们是不是可以删除了哈？
7   -/**
8   - * IoT 设备主题枚举
9   - * <p>
10   - * 用于统一管理 MQTT 协议中的主题常量，基于 Alink 协议规范
11   - *
12   - * @author haohao
13   - */
14   -@RequiredArgsConstructor
15   -@Getter
16   -public enum IotDeviceTopicEnum {
17   -
18   - /**
19   - * 设备属性设置主题
20   - * 请求 Topic：/sys/${productKey}/${deviceName}/thing/property/set
21   - * 响应 Topic：/sys/${productKey}/${deviceName}/thing/property/set_reply
22   - */
23   - PROPERTY_SET_TOPIC("/thing/property/set", "设备属性设置主题"),
24   -
25   - /**
26   - * 设备属性获取主题
27   - * 请求 Topic：/sys/${productKey}/${deviceName}/thing/property/get
28   - * 响应 Topic：/sys/${productKey}/${deviceName}/thing/property/get_reply
29   - */
30   - PROPERTY_GET_TOPIC("/thing/property/get", "设备属性获取主题"),
31   -
32   - /**
33   - * 设备配置设置主题
34   - * 请求 Topic：/sys/${productKey}/${deviceName}/thing/config/set
35   - * 响应 Topic：/sys/${productKey}/${deviceName}/thing/config/set_reply
36   - */
37   - CONFIG_SET_TOPIC("/thing/config/set", "设备配置设置主题"),
38   -
39   - /**
40   - * 设备 OTA 升级主题
41   - * 请求 Topic：/sys/${productKey}/${deviceName}/thing/ota/upgrade
42   - * 响应 Topic：/sys/${productKey}/${deviceName}/thing/ota/upgrade_reply
43   - */
44   - OTA_UPGRADE_TOPIC("/thing/ota/upgrade", "设备 OTA 升级主题"),
45   -
46   - /**
47   - * 设备属性上报主题
48   - * 请求 Topic：/sys/${productKey}/${deviceName}/thing/event/property/post
49   - * 响应 Topic：/sys/${productKey}/${deviceName}/thing/event/property/post_reply
50   - */
51   - PROPERTY_POST_TOPIC("/thing/property/post", "设备属性上报主题"),
52   -
53   - /**
54   - * 设备事件上报主题前缀
55   - */
56   - EVENT_POST_TOPIC_PREFIX("/thing/event/", "设备事件上报主题前缀"),
57   -
58   - /**
59   - * 设备事件上报主题后缀
60   - */
61   - EVENT_POST_TOPIC_SUFFIX("/post", "设备事件上报主题后缀");
62   -
63   - // ========== 静态常量 ==========
64   -
65   - /**
66   - * 系统主题前缀
67   - */
68   - public static final String SYS_TOPIC_PREFIX = "/sys/";
69   -
70   - /**
71   - * 服务调用主题前缀
72   - */
73   - public static final String SERVICE_TOPIC_PREFIX = "/thing/";
74   -
75   - /**
76   - * 响应主题后缀
77   - */
78   - public static final String REPLY_SUFFIX = "_reply";
79   -
80   - // ========== 方法常量 ==========
81   -
82   - /**
83   - * 服务方法前缀
84   - */
85   - public static final String SERVICE_METHOD_PREFIX = "thing.";
86   -
87   - /**
88   - * 属性服务方法前缀
89   - */
90   - public static final String PROPERTY_SERVICE_METHOD_PREFIX = "thing.property.";
91   -
92   - /**
93   - * 配置服务方法前缀
94   - */
95   - public static final String CONFIG_SERVICE_METHOD_PREFIX = "thing.config.";
96   -
97   - /**
98   - * OTA 服务方法前缀
99   - */
100   - public static final String OTA_SERVICE_METHOD_PREFIX = "thing.ota.";
101   -
102   - /**
103   - * 属性设置方法
104   - */
105   - public static final String PROPERTY_SET_METHOD = "thing.property.set";
106   -
107   - /**
108   - * 属性获取方法
109   - */
110   - public static final String PROPERTY_GET_METHOD = "thing.property.get";
111   -
112   - // ========== 主题匹配常量 ==========
113   -
114   - /**
115   - * 事件上报主题模式
116   - */
117   - public static final String EVENT_POST_TOPIC_PATTERN = "/thing/event/";
118   -
119   - /**
120   - * 主题后缀：post
121   - */
122   - public static final String POST_SUFFIX = "/post";
123   -
124   - /**
125   - * 属性上报主题后缀
126   - */
127   - public static final String PROPERTY_POST_SUFFIX = "/thing/property/post";
128   -
129   - /**
130   - * 属性设置响应主题包含
131   - */
132   - public static final String PROPERTY_SET_TOPIC_CONTAINS = "/thing/property/set";
133   -
134   - /**
135   - * 属性获取响应主题包含
136   - */
137   - public static final String PROPERTY_GET_TOPIC_CONTAINS = "/thing/property/get";
138   -
139   - // ========== MQTT 认证路径常量 ==========
140   -
141   - /**
142   - * MQTT 认证路径
143   - */
144   - public static final String MQTT_AUTH_AUTHENTICATE_PATH = "/mqtt/auth/authenticate";
145   -
146   - /**
147   - * MQTT 连接事件路径
148   - */
149   - public static final String MQTT_AUTH_CONNECTED_PATH = "/mqtt/auth/connected";
150   -
151   - /**
152   - * MQTT 断开事件路径
153   - */
154   - public static final String MQTT_AUTH_DISCONNECTED_PATH = "/mqtt/auth/disconnected";
155   -
156   - private final String topic;
157   - private final String description;
158   -
159   - // ========== 工具方法 ==========
160   -
161   - /**
162   - * 构建设备主题前缀
163   - *
164   - * @param productKey 产品 Key
165   - * @param deviceName 设备名称
166   - * @return 设备主题前缀：/sys/{productKey}/{deviceName}
167   - */
168   - private static String buildDeviceTopicPrefix(String productKey, String deviceName) {
169   - return SYS_TOPIC_PREFIX + productKey + "/" + deviceName;
170   - }
171   -
172   - /**
173   - * 构建设备服务调用主题
174   - *
175   - * @param productKey 产品 Key
176   - * @param deviceName 设备名称
177   - * @param serviceIdentifier 服务标识符
178   - * @return 完整的主题路径
179   - */
180   - public static String buildServiceTopic(String productKey, String deviceName, String serviceIdentifier) {
181   - return buildDeviceTopicPrefix(productKey, deviceName) + SERVICE_TOPIC_PREFIX + serviceIdentifier;
182   - }
183   -
184   - /**
185   - * 构建设备属性设置主题
186   - *
187   - * @param productKey 产品 Key
188   - * @param deviceName 设备名称
189   - * @return 完整的主题路径
190   - */
191   - public static String buildPropertySetTopic(String productKey, String deviceName) {
192   - return buildDeviceTopicPrefix(productKey, deviceName) + PROPERTY_SET_TOPIC.getTopic();
193   - }
194   -
195   - /**
196   - * 构建设备属性获取主题
197   - *
198   - * @param productKey 产品 Key
199   - * @param deviceName 设备名称
200   - * @return 完整的主题路径
201   - */
202   - public static String buildPropertyGetTopic(String productKey, String deviceName) {
203   - return buildDeviceTopicPrefix(productKey, deviceName) + PROPERTY_GET_TOPIC.getTopic();
204   - }
205   -
206   - /**
207   - * 构建设备配置设置主题
208   - *
209   - * @param productKey 产品 Key
210   - * @param deviceName 设备名称
211   - * @return 完整的主题路径
212   - */
213   - public static String buildConfigSetTopic(String productKey, String deviceName) {
214   - return buildDeviceTopicPrefix(productKey, deviceName) + CONFIG_SET_TOPIC.getTopic();
215   - }
216   -
217   - /**
218   - * 构建设备 OTA 升级主题
219   - *
220   - * @param productKey 产品 Key
221   - * @param deviceName 设备名称
222   - * @return 完整的主题路径
223   - */
224   - public static String buildOtaUpgradeTopic(String productKey, String deviceName) {
225   - return buildDeviceTopicPrefix(productKey, deviceName) + OTA_UPGRADE_TOPIC.getTopic();
226   - }
227   -
228   - /**
229   - * 构建设备属性上报主题
230   - *
231   - * @param productKey 产品 Key
232   - * @param deviceName 设备名称
233   - * @return 完整的主题路径
234   - */
235   - public static String buildPropertyPostTopic(String productKey, String deviceName) {
236   - return buildDeviceTopicPrefix(productKey, deviceName) + PROPERTY_POST_TOPIC.getTopic();
237   - }
238   -
239   - /**
240   - * 构建设备事件上报主题
241   - *
242   - * @param productKey 产品 Key
243   - * @param deviceName 设备名称
244   - * @param eventIdentifier 事件标识符
245   - * @return 完整的主题路径
246   - */
247   - public static String buildEventPostTopic(String productKey, String deviceName, String eventIdentifier) {
248   - return buildDeviceTopicPrefix(productKey, deviceName) +
249   - EVENT_POST_TOPIC_PREFIX.getTopic() + eventIdentifier + EVENT_POST_TOPIC_SUFFIX.getTopic();
250   - }
251   -
252   - /**
253   - * 获取响应主题
254   - *
255   - * @param requestTopic 请求主题
256   - * @return 响应主题
257   - */
258   - public static String getReplyTopic(String requestTopic) {
259   - return requestTopic + REPLY_SUFFIX;
260   - }
261   -
262   -}
263 0 \ No newline at end of file
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/IotMqttDownstreamSubscriber.java
  View file @569eef4
1 1 package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt;
2 2  
3   -import cn.hutool.core.util.StrUtil;
4   -import cn.hutool.json.JSONObject;
5   -import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceRespDTO;
6 3 import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
7 4 import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;
8 5 import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageSubscriber;
9 6 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
10 7 import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils;
11   -import cn.iocoder.yudao.module.iot.gateway.enums.IotDeviceTopicEnum;
12   -import cn.iocoder.yudao.module.iot.gateway.service.device.IotDeviceService;
  8 +import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router.IotMqttDownstreamHandler;
13 9 import jakarta.annotation.PostConstruct;
14   -import jakarta.annotation.Resource;
15   -import lombok.RequiredArgsConstructor;
16 10 import lombok.extern.slf4j.Slf4j;
17 11  
18 12 /**
... ... @@ -20,15 +14,18 @@ import lombok.extern.slf4j.Slf4j;
20 14 *
21 15 * @author 芋道源码
22 16 */
23   -@RequiredArgsConstructor
24 17 @Slf4j
25 18 public class IotMqttDownstreamSubscriber implements IotMessageSubscriber<IotDeviceMessage> {
26 19  
27   - private final IotMqttUpstreamProtocol protocol;
  20 + private final IotMqttDownstreamHandler downstreamHandler;
28 21 private final IotMessageBus messageBus;
  22 + private final IotMqttUpstreamProtocol protocol;
29 23  
30   - @Resource
31   - private IotDeviceService deviceService;
  24 + public IotMqttDownstreamSubscriber(IotMqttUpstreamProtocol protocol, IotMessageBus messageBus) {
  25 + this.protocol = protocol;
  26 + this.messageBus = messageBus;
  27 + this.downstreamHandler = new IotMqttDownstreamHandler(protocol);
  28 + }
32 29  
33 30 @PostConstruct
34 31 public void init() {
... ... @@ -51,7 +48,7 @@ public class IotMqttDownstreamSubscriber implements IotMessageSubscriber&lt;IotDevi
51 48 log.info("[onMessage][接收到下行消息][messageId: {}][method: {}][deviceId: {}]",
52 49 message.getId(), message.getMethod(), message.getDeviceId());
53 50 try {
54   - // 根据消息方法处理不同的下行消息
  51 + // 1. 校验
55 52 String method = message.getMethod();
56 53 if (method == null) {
57 54 log.warn("[onMessage][消息方法为空][messageId: {}][deviceId: {}]",
... ... @@ -59,15 +56,8 @@ public class IotMqttDownstreamSubscriber implements IotMessageSubscriber&lt;IotDevi
59 56 return;
60 57 }
61 58  
62   - // 过滤上行消息：下行订阅者只处理下行消息
63   - if (isUpstreamMessage(method)) {
64   - // TODO @haohao：打个 erroor log，按道理不会发生；
65   - log.debug("[onMessage][忽略上行消息][method: {}][messageId: {}]", method, message.getId());
66   - return;
67   - }
68   -
69   - // 处理下行消息
70   - handleDownstreamMessage(message);
  59 + // 2. 处理下行消息
  60 + downstreamHandler.handle(message);
71 61 } catch (Exception e) {
72 62 log.error("[onMessage][处理下行消息失败][messageId: {}][method: {}][deviceId: {}]",
73 63 message.getId(), message.getMethod(), message.getDeviceId(), e);
... ... @@ -75,99 +65,6 @@ public class IotMqttDownstreamSubscriber implements IotMessageSubscriber&lt;IotDevi
75 65 }
76 66  
77 67 /**
78   - * 处理下行消息
79   - *
80   - * @param message 设备消息
81   - */
82   - private void handleDownstreamMessage(IotDeviceMessage message) {
83   - // 1. 获取设备信息（使用缓存）
84   - IotDeviceRespDTO deviceInfo = deviceService.getDeviceFromCache(message.getDeviceId());
85   - if (deviceInfo == null) {
86   - log.warn("[handleDownstreamMessage][设备信息不存在][deviceId: {}]", message.getDeviceId());
87   - return;
88   - }
89   -
90   - // 2. 根据方法构建主题
91   - String topic = buildTopicByMethod(message.getMethod(), deviceInfo.getProductKey(), deviceInfo.getDeviceName());
92   - if (StrUtil.isBlank(topic)) {
93   - log.warn("[handleDownstreamMessage][未知的消息方法：{}]", message.getMethod());
94   - return;
95   - }
96   -
97   - // 3. 构建载荷
98   - JSONObject payload = buildDownstreamPayload(message, message.getMethod());
99   -
100   - // 4. 发送消息
101   - protocol.publishMessage(topic, payload.toString());
102   - log.info("[handleDownstreamMessage][发送下行消息成功][method: {}][topic: {}]", message.getMethod(), topic);
103   - }
104   -
105   - /**
106   - * 根据方法构建主题
107   - *
108   - * @param method 消息方法
109   - * @param productKey 产品标识
110   - * @param deviceName 设备名称
111   - * @return 构建的主题，如果方法不支持返回 null
112   - */
113   - private String buildTopicByMethod(String method, String productKey, String deviceName) {
114   - if (StrUtil.isBlank(method)) {
115   - return null;
116   - }
117   -
118   - // 属性相关操作
119   - if (method.startsWith(IotDeviceTopicEnum.PROPERTY_SERVICE_METHOD_PREFIX)) {
120   - if (IotDeviceTopicEnum.PROPERTY_SET_METHOD.equals(method)) {
121   - return IotDeviceTopicEnum.buildPropertySetTopic(productKey, deviceName);
122   - } else if (IotDeviceTopicEnum.PROPERTY_GET_METHOD.equals(method)) {
123   - return IotDeviceTopicEnum.buildPropertyGetTopic(productKey, deviceName);
124   - }
125   - return null;
126   - }
127   -
128   - // 配置设置操作
129   - if (method.startsWith(IotDeviceTopicEnum.CONFIG_SERVICE_METHOD_PREFIX)) {
130   - return IotDeviceTopicEnum.buildConfigSetTopic(productKey, deviceName);
131   - }
132   -
133   - // OTA 升级操作
134   - if (method.startsWith(IotDeviceTopicEnum.OTA_SERVICE_METHOD_PREFIX)) {
135   - return IotDeviceTopicEnum.buildOtaUpgradeTopic(productKey, deviceName);
136   - }
137   -
138   - // 一般服务调用操作
139   - if (method.startsWith(IotDeviceTopicEnum.SERVICE_METHOD_PREFIX)) {
140   - // 排除属性、配置、OTA 相关的服务调用
141   - if (method.contains("property") || method.contains("config") || method.contains("ota")) {
142   - return null; // 已在上面处理
143   - }
144   - // 从方法中提取服务标识符
145   - String serviceIdentifier = method.substring(IotDeviceTopicEnum.SERVICE_METHOD_PREFIX.length());
146   - return IotDeviceTopicEnum.buildServiceTopic(productKey, deviceName, serviceIdentifier);
147   - }
148   -
149   - // 不支持的方法
150   - return null;
151   - }
152   -
153   - // TODO @haohao：按道理说，这里的应该是通过 encodeMessage；
154   - /**
155   - * 构建下行消息载荷
156   - *
157   - * @param message 设备消息
158   - * @param method 方法名
159   - * @return JSON 载荷
160   - */
161   - private JSONObject buildDownstreamPayload(IotDeviceMessage message, String method) {
162   - JSONObject payload = new JSONObject();
163   - payload.set("id", message.getId()); // 使用正确的消息ID字段
164   - payload.set("version", "1.0");
165   - payload.set("method", method);
166   - payload.set("params", message.getData());
167   - return payload;
168   - }
169   -
170   - /**
171 68 * 判断是否为上行消息
172 69 *
173 70 * @param method 消息方法
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/IotMqttUpstreamProtocol.java
  View file @569eef4
... ... @@ -4,9 +4,9 @@ import cn.hutool.core.collection.CollUtil;
4 4 import cn.hutool.core.util.StrUtil;
5 5 import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils;
6 6 import cn.iocoder.yudao.module.iot.gateway.config.IotGatewayProperties;
7   -import cn.iocoder.yudao.module.iot.gateway.enums.IotDeviceTopicEnum;
8 7 import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router.IotMqttHttpAuthHandler;
9 8 import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router.IotMqttUpstreamHandler;
  9 +import cn.iocoder.yudao.module.iot.gateway.util.IotMqttTopicUtils;
10 10 import io.netty.handler.codec.mqtt.MqttQoS;
11 11 import io.vertx.core.Vertx;
12 12 import io.vertx.core.buffer.Buffer;
... ... @@ -21,13 +21,10 @@ import lombok.Getter;
21 21 import lombok.extern.slf4j.Slf4j;
22 22  
23 23 import java.util.List;
24   -import java.util.concurrent.CompletableFuture;
25 24 import java.util.concurrent.TimeUnit;
26 25  
27   -// TODO @haohao：看看有没多余的 log，可以不打噢。
28   -// TODO @haohao：有没多余的注释可以去掉，减少 ai 味，保持简洁；
29 26 /**
30   - * IoT 网关 MQTT 统一协议
  27 + * IoT 网关 MQTT 协议：接收设备上行消息
31 28 * <p>
32 29 * 1. MQTT 客户端：连接 EMQX，消费处理设备上行和下行消息
33 30 * 2. HTTP 认证服务：为 EMQX 提供设备认证、连接、断开接口
... ... @@ -37,24 +34,6 @@ import java.util.concurrent.TimeUnit;
37 34 @Slf4j
38 35 public class IotMqttUpstreamProtocol {
39 36  
40   - // TODO @haohao：是不是也丢到配置里？
41   - /**
42   - * 默认 QoS 级别 - 至少一次
43   - */
44   - private static final MqttQoS DEFAULT_QOS = MqttQoS.AT_LEAST_ONCE;
45   -
46   - // TODO @haohao：这个也是；
47   - /**
48   - * 连接超时时间（秒）
49   - */
50   - private static final int CONNECT_TIMEOUT_SECONDS = 10;
51   -
52   - // TODO @haohao：重连也是；
53   - /**
54   - * 重连延迟时间（毫秒）
55   - */
56   - private static final long RECONNECT_DELAY_MS = 5000;
57   -
58 37 private final IotGatewayProperties.EmqxProperties emqxProperties;
59 38  
60 39 private Vertx vertx;
... ... @@ -68,8 +47,6 @@ public class IotMqttUpstreamProtocol {
68 47  
69 48 // HTTP 认证服务相关
70 49 private HttpServer httpAuthServer;
71   - // TODO @haohao：authHandler 可以 local 哈；
72   - private IotMqttHttpAuthHandler authHandler;
73 50  
74 51 /**
75 52 * 服务运行状态标志
... ... @@ -92,7 +69,6 @@ public class IotMqttUpstreamProtocol {
92 69 try {
93 70 // 1. 创建共享的 Vertx 实例
94 71 this.vertx = Vertx.vertx();
95   - log.info("[start][共享 Vertx 实例创建成功]");
96 72  
97 73 // 2. 启动 HTTP 认证服务
98 74 startHttpAuthServer();
... ... @@ -103,7 +79,6 @@ public class IotMqttUpstreamProtocol {
103 79 isRunning = true;
104 80 log.info("[start][MQTT 统一协议服务启动完成]");
105 81 } catch (Exception e) {
106   - // TODO @haohao：失败，是不是直接 System.exit 哈！
107 82 log.error("[start][MQTT 统一协议服务启动失败]", e);
108 83 // 启动失败时清理资源
109 84 stop();
... ... @@ -150,12 +125,10 @@ public class IotMqttUpstreamProtocol {
150 125 router.route().handler(BodyHandler.create());
151 126  
152 127 // 创建认证处理器
153   - this.authHandler = new IotMqttHttpAuthHandler();
154   -
155   - // 添加认证路由
156   - router.post(IotDeviceTopicEnum.MQTT_AUTH_AUTHENTICATE_PATH).handler(authHandler::authenticate);
157   - router.post(IotDeviceTopicEnum.MQTT_AUTH_CONNECTED_PATH).handler(authHandler::connected);
158   - router.post(IotDeviceTopicEnum.MQTT_AUTH_DISCONNECTED_PATH).handler(authHandler::disconnected);
  128 + IotMqttHttpAuthHandler authHandler = new IotMqttHttpAuthHandler();
  129 + router.post(IotMqttTopicUtils.MQTT_AUTH_AUTHENTICATE_PATH).handler(authHandler::authenticate);
  130 + router.post(IotMqttTopicUtils.MQTT_AUTH_CONNECTED_PATH).handler(authHandler::connected);
  131 + router.post(IotMqttTopicUtils.MQTT_AUTH_DISCONNECTED_PATH).handler(authHandler::disconnected);
159 132  
160 133 // 启动 HTTP 服务器
161 134 int authPort = emqxProperties.getHttpAuthPort();
... ... @@ -175,14 +148,14 @@ public class IotMqttUpstreamProtocol {
175 148 * 停止 HTTP 认证服务
176 149 */
177 150 private void stopHttpAuthServer() {
178   - // TODO @haohao：一些 if return 最好搞下；
179   - if (httpAuthServer != null) {
180   - try {
181   - httpAuthServer.close().result();
182   - log.info("[stopHttpAuthServer][HTTP 认证服务已停止]");
183   - } catch (Exception e) {
184   - log.error("[stopHttpAuthServer][HTTP 认证服务停止失败]", e);
185   - }
  151 + if (httpAuthServer == null) {
  152 + return;
  153 + }
  154 + try {
  155 + httpAuthServer.close().result();
  156 + log.info("[stopHttpAuthServer][HTTP 认证服务已停止]");
  157 + } catch (Exception e) {
  158 + log.error("[stopHttpAuthServer][HTTP 认证服务停止失败]", e);
186 159 }
187 160 }
188 161  
... ... @@ -192,10 +165,10 @@ public class IotMqttUpstreamProtocol {
192 165 private void startMqttClient() {
193 166 log.info("[startMqttClient][开始启动 MQTT 客户端]");
194 167  
195   - // 初始化消息处理器
  168 + // 1. 初始化消息处理器
196 169 this.upstreamHandler = new IotMqttUpstreamHandler(this);
197 170  
198   - // 创建 MQTT 客户端
  171 + // 2. 创建 MQTT 客户端
199 172 log.info("[startMqttClient][使用 MQTT 客户端 ID: {}]", emqxProperties.getMqttClientId());
200 173  
201 174 MqttClientOptions options = new MqttClientOptions()
... ... @@ -205,7 +178,7 @@ public class IotMqttUpstreamProtocol {
205 178 .setSsl(emqxProperties.getMqttSsl());
206 179 this.mqttClient = MqttClient.create(vertx, options);
207 180  
208   - // 连接 MQTT Broker
  181 + // 3. 连接 MQTT Broker
209 182 connectMqtt();
210 183 }
211 184  
... ... @@ -229,13 +202,13 @@ public class IotMqttUpstreamProtocol {
229 202 }
230 203  
231 204 // 2. 关闭 MQTT 客户端
232   - try {
233   - if (mqttClient != null && mqttClient.isConnected()) {
  205 + if (mqttClient != null && mqttClient.isConnected()) {
  206 + try {
234 207 mqttClient.disconnect();
235 208 log.info("[stopMqttClient][MQTT 客户端已断开]");
  209 + } catch (Exception e) {
  210 + log.warn("[stopMqttClient][关闭 MQTT 客户端异常]", e);
236 211 }
237   - } catch (Exception e) {
238   - log.warn("[stopMqttClient][关闭 MQTT 客户端异常]", e);
239 212 }
240 213 }
241 214  
... ... @@ -243,10 +216,9 @@ public class IotMqttUpstreamProtocol {
243 216 * 连接 MQTT Broker 并订阅主题
244 217 */
245 218 private void connectMqtt() {
246   - // 参数校验
  219 + // 1. 参数校验
247 220 String host = emqxProperties.getMqttHost();
248 221 Integer port = emqxProperties.getMqttPort();
249   -
250 222 if (StrUtil.isBlank(host)) {
251 223 log.error("[connectMqtt][MQTT Host 为空，无法连接]");
252 224 throw new IllegalArgumentException("MQTT Host 不能为空");
... ... @@ -255,63 +227,47 @@ public class IotMqttUpstreamProtocol {
255 227 log.error("[connectMqtt][MQTT Port 无效：{}]", port);
256 228 throw new IllegalArgumentException("MQTT Port 必须为正整数");
257 229 }
258   -
259 230 log.info("[connectMqtt][开始连接 MQTT Broker][host: {}][port: {}]", host, port);
260 231  
261   - CompletableFuture<Void> connectFuture = mqttClient.connect(port, host)
262   - .toCompletionStage()
263   - .toCompletableFuture()
264   - .thenAccept(connAck -> {
265   - // TODO @haohao：是不是可以连接完，然后在执行里面；不用 通过 thenAccept 哈；
266   - log.info("[connectMqtt][MQTT 客户端连接成功][host: {}][port: {}]", host, port);
267   - // 设置断开重连监听器
268   - mqttClient.closeHandler(closeEvent -> {
269   - log.warn("[closeHandler][MQTT 连接已断开，准备重连]");
270   - reconnectWithDelay();
271   - });
272   - // 设置消息处理器
273   - setupMessageHandler();
274   - // 订阅主题
275   - subscribeToTopics();
276   - })
277   - .exceptionally(error -> {
278   - // TODO @haohao：这里的异常，是不是不用重连哈？因为直接就退出了。然后有 closeHandler 监听重连了；
279   - log.error("[connectMqtt][连接 MQTT Broker 失败][host: {}][port: {}]", host, port, error);
280   - // 连接失败时也要尝试重连
281   - reconnectWithDelay();
282   - return null;
283   - });
284   -
285   - // 等待连接完成
  232 + // 2. 连接
286 233 try {
287   - connectFuture.get(CONNECT_TIMEOUT_SECONDS, TimeUnit.SECONDS);
288   - log.info("[connectMqtt][MQTT 客户端启动完成]");
  234 + mqttClient.connect(port, host)
  235 + .toCompletionStage()
  236 + .toCompletableFuture()
  237 + .get(emqxProperties.getConnectTimeoutSeconds(), TimeUnit.SECONDS);
  238 + log.info("[connectMqtt][MQTT 客户端连接成功][host: {}][port: {}]", host, port);
  239 +
  240 + // 3. 设置处理器
  241 + // 3.1 设置断开重连监听器
  242 + mqttClient.closeHandler(closeEvent -> {
  243 + log.warn("[closeHandler][MQTT 连接已断开，准备重连]");
  244 + reconnectWithDelay();
  245 + });
  246 + // 3.2 设置消息处理器
  247 + mqttClient.publishHandler(upstreamHandler::handle);
  248 + log.debug("[setupMessageHandler][MQTT 消息处理器设置完成]");
  249 +
  250 + // 4. 订阅主题
  251 + subscribeToTopics();
289 252 } catch (Exception e) {
290   - log.error("[connectMqtt][MQTT 客户端启动失败]", e);
  253 + log.error("[connectMqtt][连接 MQTT Broker 失败][host: {}][port: {}]", host, port, e);
  254 + reconnectWithDelay(); // 连接失败时，也要尝试重连
291 255 throw new RuntimeException("MQTT 客户端启动失败", e);
292 256 }
293 257 }
294 258  
295 259 /**
296   - * 设置 MQTT 消息处理器
297   - */
298   - private void setupMessageHandler() {
299   - mqttClient.publishHandler(upstreamHandler::handle);
300   - log.debug("[setupMessageHandler][MQTT 消息处理器设置完成]");
301   - }
302   -
303   - /**
304 260 * 订阅设备上行消息主题
305 261 */
306 262 private void subscribeToTopics() {
307 263 List<String> topicList = emqxProperties.getMqttTopics();
  264 + int qos = emqxProperties.getMqttQos();
308 265  
309 266 log.info("[subscribeToTopics][开始订阅主题，共 {} 个]", topicList.size());
310   -
311 267 for (String topic : topicList) {
312   - mqttClient.subscribe(topic, DEFAULT_QOS.value(), subscribeResult -> {
  268 + mqttClient.subscribe(topic, qos, subscribeResult -> {
313 269 if (subscribeResult.succeeded()) {
314   - log.info("[subscribeToTopics][订阅主题成功: {}][QoS: {}]", topic, DEFAULT_QOS.value());
  270 + log.info("[subscribeToTopics][订阅主题成功: {}][QoS: {}]", topic, qos);
315 271 } else {
316 272 log.error("[subscribeToTopics][订阅主题失败: {}]", topic, subscribeResult.cause());
317 273 }
... ... @@ -323,17 +279,17 @@ public class IotMqttUpstreamProtocol {
323 279 * 延迟重连
324 280 */
325 281 private void reconnectWithDelay() {
326   - vertx.setTimer(RECONNECT_DELAY_MS, timerId -> {
327   - // TODO @haohao：if return，括号少一些；
328   - if (isRunning && (mqttClient == null || !mqttClient.isConnected())) {
329   - log.info("[reconnectWithDelay][开始重连 MQTT Broker，延迟 {} 毫秒]", RECONNECT_DELAY_MS);
330   - try {
331   - connectMqtt();
332   - } catch (Exception e) {
333   - log.error("[reconnectWithDelay][重连失败，将继续尝试重连]", e);
334   - // 重连失败时继续尝试重连
335   - reconnectWithDelay();
336   - }
  282 + long delay = emqxProperties.getReconnectDelayMs();
  283 + vertx.setTimer(delay, timerId -> {
  284 + if (!isRunning || (mqttClient != null && mqttClient.isConnected())) {
  285 + return;
  286 + }
  287 + log.info("[reconnectWithDelay][开始重连 MQTT Broker，延迟 {} 毫秒]", delay);
  288 + try {
  289 + connectMqtt();
  290 + } catch (Exception e) {
  291 + log.error("[reconnectWithDelay][重连失败，将继续尝试重连]", e);
  292 + reconnectWithDelay(); // 失败后，继续尝试
337 293 }
338 294 });
339 295 }
... ... @@ -345,12 +301,12 @@ public class IotMqttUpstreamProtocol {
345 301 * @param payload 消息内容
346 302 */
347 303 public void publishMessage(String topic, String payload) {
348   - if (mqttClient != null && mqttClient.isConnected()) {
349   - mqttClient.publish(topic, Buffer.buffer(payload), DEFAULT_QOS, false, false);
350   - log.debug("[publishMessage][发布消息成功][topic: {}]", topic);
351   - } else {
  304 + if (mqttClient == null || !mqttClient.isConnected()) {
352 305 log.warn("[publishMessage][MQTT 客户端未连接，无法发布消息][topic: {}]", topic);
  306 + return;
353 307 }
  308 + MqttQoS qos = MqttQoS.valueOf(emqxProperties.getMqttQos());
  309 + mqttClient.publish(topic, Buffer.buffer(payload), qos, false, false);
354 310 }
355 311  
356 312 }
357 313 \ No newline at end of file
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttAbstractHandler.java deleted
View file @b06556d
1   -package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router;
2   -
3   -import cn.hutool.core.util.StrUtil;
4   -import lombok.extern.slf4j.Slf4j;
5   -
6   -// TODO @haohao：是不是不用基类哈；
7   -/**
8   - * IoT 网关 MQTT 协议的处理器抽象基类
9   - * <p>
10   - * 提供通用的异常处理、参数校验等功能
11   - *
12   - * @author 芋道源码
13   - */
14   -@Slf4j
15   -public abstract class IotMqttAbstractHandler {
16   -
17   - /**
18   - * 处理 MQTT 消息的模板方法
19   - *
20   - * @param topic 主题
21   - * @param payload 消息内容
22   - */
23   - public final void handle(String topic, String payload) {
24   - try {
25   - // 1. 前置校验
26   - if (!validateInput(topic, payload)) {
27   - return;
28   - }
29   -
30   - // 2. 执行具体逻辑
31   - doHandle(topic, payload);
32   -
33   - } catch (Exception e) {
34   - log.error("[handle][处理 MQTT 消息失败][topic: {}][payload: {}]", topic, payload, e);
35   - handleException(topic, payload, e);
36   - }
37   - }
38   -
39   - /**
40   - * 具体的处理逻辑，由子类实现
41   - *
42   - * @param topic 主题
43   - * @param payload 消息内容
44   - */
45   - protected abstract void doHandle(String topic, String payload);
46   -
47   - /**
48   - * 输入参数校验
49   - *
50   - * @param topic 主题
51   - * @param payload 消息内容
52   - * @return 校验是否通过
53   - */
54   - protected boolean validateInput(String topic, String payload) {
55   - if (StrUtil.isBlank(topic)) {
56   - log.warn("[validateInput][主题为空，忽略消息]");
57   - return false;
58   - }
59   -
60   - if (StrUtil.isBlank(payload)) {
61   - log.warn("[validateInput][消息内容为空][topic: {}]", topic);
62   - return false;
63   - }
64   -
65   - return true;
66   - }
67   -
68   - /**
69   - * 异常处理
70   - *
71   - * @param topic 主题
72   - * @param payload 消息内容
73   - * @param e 异常
74   - */
75   - protected void handleException(String topic, String payload, Exception e) {
76   - // 默认实现：记录错误日志
77   - // 子类可以重写此方法，添加特定的异常处理逻辑
78   - log.error("[handleException][MQTT 消息处理异常][topic: {}]", topic, e);
79   - }
80   -
81   - /**
82   - * 解析主题，获取主题各部分
83   - *
84   - * @param topic 主题
85   - * @return 主题各部分数组，如果解析失败返回 null
86   - */
87   - protected String[] parseTopic(String topic) {
88   - String[] topicParts = topic.split("/");
89   - if (topicParts.length < 7) {
90   - log.warn("[parseTopic][主题格式不正确][topic: {}]", topic);
91   - return null;
92   - }
93   - return topicParts;
94   - }
95   -}
96 0 \ No newline at end of file
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttDownstreamHandler.java 0 → 100644
  View file @569eef4
  1 +package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router;
  2 +
  3 +import cn.hutool.core.util.StrUtil;
  4 +import cn.hutool.extra.spring.SpringUtil;
  5 +import cn.hutool.json.JSONObject;
  6 +import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceRespDTO;
  7 +import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
  8 +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
  9 +import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttUpstreamProtocol;
  10 +import cn.iocoder.yudao.module.iot.gateway.service.device.IotDeviceService;
  11 +import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
  12 +import cn.iocoder.yudao.module.iot.gateway.util.IotMqttTopicUtils;
  13 +import lombok.extern.slf4j.Slf4j;
  14 +
  15 +/**
  16 + * IoT 网关 MQTT 下行消息处理器
  17 + * <p>
  18 + * 从消息总线接收到下行消息，然后发布到 MQTT Broker
  19 + *
  20 + * @author 芋道源码
  21 + */
  22 +@Slf4j
  23 +public class IotMqttDownstreamHandler {
  24 +
  25 + private final IotMqttUpstreamProtocol protocol;
  26 + private final IotDeviceService deviceService;
  27 + private final IotDeviceMessageService deviceMessageService;
  28 +
  29 + public IotMqttDownstreamHandler(IotMqttUpstreamProtocol protocol) {
  30 + this.protocol = protocol;
  31 + this.deviceService = SpringUtil.getBean(IotDeviceService.class);
  32 + this.deviceMessageService = SpringUtil.getBean(IotDeviceMessageService.class);
  33 + }
  34 +
  35 + /**
  36 + * 处理下行消息
  37 + *
  38 + * @param message 设备消息
  39 + */
  40 + public void handle(IotDeviceMessage message) {
  41 + // 1. 获取设备信息（使用缓存）
  42 + IotDeviceRespDTO deviceInfo = deviceService.getDeviceFromCache(message.getDeviceId());
  43 + if (deviceInfo == null) {
  44 + log.warn("[handle][设备信息不存在][deviceId: {}]", message.getDeviceId());
  45 + return;
  46 + }
  47 +
  48 + // 2. 根据方法构建主题
  49 + String topic = buildTopicByMethod(message.getMethod(), deviceInfo.getProductKey(), deviceInfo.getDeviceName());
  50 + if (StrUtil.isBlank(topic)) {
  51 + log.warn("[handle][未知的消息方法：{}]", message.getMethod());
  52 + return;
  53 + }
  54 +
  55 + // 3. 构建载荷
  56 + JSONObject payload = buildDownstreamPayload(message);
  57 +
  58 + // 4. 发布消息
  59 + protocol.publishMessage(topic, payload.toString());
  60 + log.info("[handle][发布下行消息成功][method: {}][topic: {}]", message.getMethod(), topic);
  61 + }
  62 +
  63 + /**
  64 + * 根据方法构建主题
  65 + *
  66 + * @param method 消息方法
  67 + * @param productKey 产品标识
  68 + * @param deviceName 设备名称
  69 + * @return 构建的主题，如果方法不支持返回 null
  70 + */
  71 + private String buildTopicByMethod(String method, String productKey, String deviceName) {
  72 + IotDeviceMessageMethodEnum methodEnum = IotDeviceMessageMethodEnum.of(method);
  73 + if (methodEnum == null) {
  74 + return null;
  75 + }
  76 + return switch (methodEnum) {
  77 + case PROPERTY_POST -> IotMqttTopicUtils.buildPropertyPostReplyTopic(productKey, deviceName);
  78 + case PROPERTY_SET -> IotMqttTopicUtils.buildPropertySetTopic(productKey, deviceName);
  79 + default -> null;
  80 + };
  81 +
  82 + }
  83 +
  84 + /**
  85 + * 构建下行消息载荷
  86 + *
  87 + * @param message 设备消息
  88 + * @return JSON 载荷
  89 + */
  90 + private JSONObject buildDownstreamPayload(IotDeviceMessage message) {
  91 + // 使用 IotDeviceMessageService 进行消息编码
  92 + IotDeviceRespDTO device = deviceService.getDeviceFromCache(message.getDeviceId());
  93 + byte[] encodedBytes = deviceMessageService.encodeDeviceMessage(message, device.getProductKey(),
  94 + device.getDeviceName());
  95 + return new JSONObject(new String(encodedBytes));
  96 + }
  97 +
  98 +}
0 99 \ No newline at end of file
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttUpstreamHandler.java
  View file @569eef4
1 1 package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router;
2 2  
  3 +import cn.hutool.core.util.StrUtil;
3 4 import cn.hutool.extra.spring.SpringUtil;
4 5 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
5   -import cn.iocoder.yudao.module.iot.gateway.enums.IotDeviceTopicEnum;
6 6 import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttUpstreamProtocol;
7 7 import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
8 8 import io.vertx.mqtt.messages.MqttPublishMessage;
9 9 import lombok.extern.slf4j.Slf4j;
10   -
11   -import java.nio.charset.StandardCharsets;
  10 +import org.springframework.util.Assert;
12 11  
13 12 /**
14   - * IoT 网关 MQTT 协议的【上行】处理器
15   - * <p>
16   - * 处理设备上行消息，包括事件上报、属性上报、服务调用响应等
  13 + * IoT 网关 MQTT 上行消息处理器
17 14 *
18 15 * @author 芋道源码
19 16 */
20 17 @Slf4j
21   -public class IotMqttUpstreamHandler extends IotMqttAbstractHandler {
  18 +public class IotMqttUpstreamHandler {
22 19  
23 20 private final IotDeviceMessageService deviceMessageService;
24   -
25 21 private final String serverId;
26 22  
27 23 public IotMqttUpstreamHandler(IotMqttUpstreamProtocol protocol) {
... ... @@ -34,167 +30,66 @@ public class IotMqttUpstreamHandler extends IotMqttAbstractHandler {
34 30 */
35 31 public void handle(MqttPublishMessage message) {
36 32 String topic = message.topicName();
37   - // TODO @haohao： message.payload().getBytes();
38   - String payload = message.payload().toString(StandardCharsets.UTF_8);
  33 + byte[] payload = message.payload().getBytes();
39 34  
40 35 log.debug("[handle][收到 MQTT 消息][topic: {}]", topic);
41   - // 调用父类的 handle 方法，父类会进行参数校验
42   - handle(topic, payload);
43   - }
44 36  
45   - @Override
46   - protected void doHandle(String topic, String payload) {
47   - // 1. 识别并验证消息类型
48   - String messageType = getMessageType(topic);
49   - if (messageType == null) {
50   - // TODO @haohao：log 是不是把 payload 也打印下哈
51   - log.warn("[doHandle][未知的消息类型][topic: {}]", topic);
52   - return;
  37 + try {
  38 + // 1. 前置校验
  39 + if (StrUtil.isBlank(topic)) {
  40 + log.warn("[validateInput][主题为空，忽略消息]");
  41 + return;
  42 + }
  43 + // 注意：payload 可以为空
  44 +
  45 + // 2. 识别并验证消息类型
  46 + String messageType = getMessageType(topic);
  47 + Assert.notNull(messageType, String.format("未知的消息类型, topic(%s)", topic));
  48 + log.info("[handle][接收到{}][topic: {}]", messageType, topic);
  49 +
  50 + // 3. 解析主题，获取 productKey 和 deviceName
  51 + String[] topicParts = topic.split("/");
  52 + if (topicParts.length < 4) {
  53 + log.warn("[handle][主题格式不正确，无法解析 productKey 和 deviceName][topic: {}]", topic);
  54 + return;
  55 + }
  56 + String productKey = topicParts[2];
  57 + String deviceName = topicParts[3];
  58 + if (StrUtil.isAllBlank(productKey, deviceName)) {
  59 + log.warn("[handle][主题中 productKey 或 deviceName 为空][topic: {}]", topic);
  60 + return;
  61 + }
  62 +
  63 + // 4. 解码消息
  64 + IotDeviceMessage deviceMessage = deviceMessageService.decodeDeviceMessage(
  65 + payload, productKey, deviceName);
  66 + if (deviceMessage == null) {
  67 + log.warn("[handle][消息解码失败][topic: {}]", topic);
  68 + return;
  69 + }
  70 +
  71 + // 5. 发送消息到队列
  72 + deviceMessageService.sendDeviceMessage(deviceMessage, productKey, deviceName, serverId);
  73 +
  74 + // 6. 记录成功日志
  75 + log.info("[handle][处理{}成功，已转发到 MQ][topic: {}]", messageType, topic);
  76 + } catch (Exception e) {
  77 + log.error("[handle][处理 MQTT 消息失败][topic: {}][payload: {}]", topic, new String(payload), e);
53 78 }
54   -
55   - // 2. 处理消息
56   - processMessage(topic, payload, messageType);
57 79 }
58 80  
59 81 /**
60   - * 处理消息的统一逻辑
61   - */
62   - private void processMessage(String topic, String payload, String messageType) {
63   - // TODO @haohao：messageType 解析，是不是作用不大哈？
64   - log.info("[processMessage][接收到{}][topic: {}]", messageType, topic);
65   -
66   - // 解析主题获取设备信息
67   - // TODO @haohao：不一定是 7 个哈；阿里云 topic 有点差异的；可以考虑解析到 topicParts[2]、topicParts[3] 的 topic
68   - String[] topicParts = parseTopic(topic);
69   - if (topicParts == null) {
70   - return;
71   - }
72   -
73   - String productKey = topicParts[2];
74   - String deviceName = topicParts[3];
75   - // TODO @haohao：解析不到，可以打个 error log；
76   -
77   - // 解码消息
78   - byte[] messageBytes = payload.getBytes(StandardCharsets.UTF_8);
79   - IotDeviceMessage message = deviceMessageService.decodeDeviceMessage(
80   - messageBytes, productKey, deviceName);
81   -
82   - // 发送消息到队列
83   - deviceMessageService.sendDeviceMessage(message, productKey, deviceName, serverId);
84   -
85   - // 记录成功日志
86   - log.info("[processMessage][处理{}成功][topic: {}]", messageType, topic);
87   - }
88   -
89   - // TODO @haohao：合并下处理；不搞成每个 topic 一个处理；
90   - /**
91   - * 识别消息类型
  82 + * 从主题中，获得消息类型
92 83 *
93 84 * @param topic 主题
94   - * @return 消息类型描述，如果不支持返回 null
  85 + * @return 消息类型
95 86 */
96 87 private String getMessageType(String topic) {
97   - // 此方法由 doHandle 调用，topic 已经在父类中校验过，无需重复校验
98   -
99   - // 按优先级匹配主题类型，避免误匹配
100   -
101   - // 1. 设备属性上报: /sys/{productKey}/{deviceName}/thing/property/post
102   - if (isPropertyPostTopic(topic)) {
103   - return IotDeviceTopicEnum.PROPERTY_POST_TOPIC.getDescription();
104   - }
105   -
106   - // 2. 设备事件上报: /sys/{productKey}/{deviceName}/thing/{eventIdentifier}/post
107   - if (isEventPostTopic(topic)) {
108   - return "设备事件上报";
109   - }
110   -
111   - // 3. 设备属性设置响应: /sys/{productKey}/{deviceName}/thing/property/set_reply
112   - if (isPropertySetReplyTopic(topic)) {
113   - return "设备属性设置响应";
114   - }
115   -
116   - // 4. 设备属性获取响应: /sys/{productKey}/{deviceName}/thing/property/get_reply
117   - if (isPropertyGetReplyTopic(topic)) {
118   - return "设备属性获取响应";
  88 + String[] topicParts = topic.split("/");
  89 + if (topicParts.length < 7) {
  90 + return null;
119 91 }
120   -
121   - // 5. 设备配置设置响应: /sys/{productKey}/{deviceName}/thing/config/set_reply
122   - if (isConfigSetReplyTopic(topic)) {
123   - return IotDeviceTopicEnum.CONFIG_SET_TOPIC.getDescription() + "响应";
124   - }
125   -
126   - // 6. 设备 OTA 升级响应:
127   - // /sys/{productKey}/{deviceName}/thing/ota/upgrade_reply
128   - if (isOtaUpgradeReplyTopic(topic)) {
129   - return IotDeviceTopicEnum.OTA_UPGRADE_TOPIC.getDescription() + "响应";
130   - }
131   -
132   - // 7. 其他服务调用响应: 通用服务调用响应
133   - if (isServiceReplyTopic(topic)) {
134   - return "设备服务调用响应";
135   - }
136   -
137   - // 不支持的消息类型
138   - return null;
139   - }
140   -
141   - /**
142   - * 判断是否为属性上报主题
143   - */
144   - private boolean isPropertyPostTopic(String topic) {
145   - return topic.contains(IotDeviceTopicEnum.PROPERTY_POST_TOPIC.getTopic());
146   - }
147   -
148   - /**
149   - * 判断是否为事件上报主题
150   - */
151   - private boolean isEventPostTopic(String topic) {
152   - return topic.contains(IotDeviceTopicEnum.EVENT_POST_TOPIC_PREFIX.getTopic())
153   - && topic.endsWith(IotDeviceTopicEnum.EVENT_POST_TOPIC_SUFFIX.getTopic())
154   - && !topic.contains("property"); // 排除属性上报主题
155   - }
156   -
157   - /**
158   - * 判断是否为属性设置响应主题
159   - */
160   - private boolean isPropertySetReplyTopic(String topic) {
161   - return topic.contains(IotDeviceTopicEnum.PROPERTY_SET_TOPIC.getTopic())
162   - && topic.endsWith(IotDeviceTopicEnum.REPLY_SUFFIX);
163   - }
164   -
165   - /**
166   - * 判断是否为属性获取响应主题
167   - */
168   - private boolean isPropertyGetReplyTopic(String topic) {
169   - return topic.contains(IotDeviceTopicEnum.PROPERTY_GET_TOPIC.getTopic())
170   - && topic.endsWith(IotDeviceTopicEnum.REPLY_SUFFIX);
171   - }
172   -
173   - /**
174   - * 判断是否为配置设置响应主题
175   - */
176   - private boolean isConfigSetReplyTopic(String topic) {
177   - return topic.contains(IotDeviceTopicEnum.CONFIG_SET_TOPIC.getTopic())
178   - && topic.endsWith(IotDeviceTopicEnum.REPLY_SUFFIX);
179   - }
180   -
181   - /**
182   - * 判断是否为 OTA 升级响应主题
183   - */
184   - private boolean isOtaUpgradeReplyTopic(String topic) {
185   - return topic.contains(IotDeviceTopicEnum.OTA_UPGRADE_TOPIC.getTopic())
186   - && topic.endsWith(IotDeviceTopicEnum.REPLY_SUFFIX);
187   - }
188   -
189   - /**
190   - * 判断是否为服务调用响应主题（排除已处理的特殊服务）
191   - */
192   - private boolean isServiceReplyTopic(String topic) {
193   - return topic.contains(IotDeviceTopicEnum.SERVICE_TOPIC_PREFIX)
194   - && topic.endsWith(IotDeviceTopicEnum.REPLY_SUFFIX)
195   - && !topic.contains("property")
196   - && !topic.contains("config")
197   - && !topic.contains("ota");
  92 + return topicParts[3];
198 93 }
199 94  
200 95 }
201 96 \ No newline at end of file
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/util/IotMqttTopicUtils.java 0 → 100644
  View file @569eef4
  1 +package cn.iocoder.yudao.module.iot.gateway.util;
  2 +
  3 +/**
  4 + * IoT 网关 MQTT 主题工具类
  5 + * <p>
  6 + * 用于统一管理 MQTT 协议中的主题常量，基于 Alink 协议规范
  7 + *
  8 + * @author 芋道源码
  9 + */
  10 +public final class IotMqttTopicUtils {
  11 +
  12 + // ========== 静态常量 ==========
  13 +
  14 + /**
  15 + * 系统主题前缀
  16 + */
  17 + private static final String SYS_TOPIC_PREFIX = "/sys/";
  18 +
  19 + /**
  20 + * 服务调用主题前缀
  21 + */
  22 + private static final String SERVICE_TOPIC_PREFIX = "/thing/";
  23 +
  24 + // ========== MQTT 认证路径常量 ==========
  25 +
  26 + /**
  27 + * MQTT 认证路径
  28 + */
  29 + public static final String MQTT_AUTH_AUTHENTICATE_PATH = "/mqtt/auth/authenticate";
  30 +
  31 + /**
  32 + * MQTT 连接事件路径
  33 + */
  34 + public static final String MQTT_AUTH_CONNECTED_PATH = "/mqtt/auth/connected";
  35 +
  36 + /**
  37 + * MQTT 断开事件路径
  38 + */
  39 + public static final String MQTT_AUTH_DISCONNECTED_PATH = "/mqtt/auth/disconnected";
  40 +
  41 + // ========== 工具方法 ==========
  42 +
  43 + /**
  44 + * 构建设备主题前缀
  45 + *
  46 + * @param productKey 产品 Key
  47 + * @param deviceName 设备名称
  48 + * @return 设备主题前缀：/sys/{productKey}/{deviceName}
  49 + */
  50 + private static String buildDeviceTopicPrefix(String productKey, String deviceName) {
  51 + return SYS_TOPIC_PREFIX + productKey + "/" + deviceName;
  52 + }
  53 +
  54 + /**
  55 + * 构建设备属性设置主题
  56 + *
  57 + * @param productKey 产品 Key
  58 + * @param deviceName 设备名称
  59 + * @return 完整的主题路径
  60 + */
  61 + public static String buildPropertySetTopic(String productKey, String deviceName) {
  62 + return buildDeviceTopicPrefix(productKey, deviceName) + "/thing/property/set";
  63 + }
  64 +
  65 + /**
  66 + * 构建设备属性上报回复主题
  67 + * <p>
  68 + * 当设备上报属性时，会收到该主题的回复
  69 + *
  70 + * @param productKey 产品 Key
  71 + * @param deviceName 设备名称
  72 + * @return 完整的主题路径
  73 + */
  74 + public static String buildPropertyPostReplyTopic(String productKey, String deviceName) {
  75 + return buildDeviceTopicPrefix(productKey, deviceName) + "/thing/property/post_reply";
  76 + }
  77 +
  78 + /**
  79 + * 构建设备服务调用主题
  80 + *
  81 + * @param productKey 产品 Key
  82 + * @param deviceName 设备名称
  83 + * @param serviceIdentifier 服务标识符
  84 + * @return 完整的主题路径
  85 + */
  86 + public static String buildServiceTopic(String productKey, String deviceName, String serviceIdentifier) {
  87 + return buildDeviceTopicPrefix(productKey, deviceName) + SERVICE_TOPIC_PREFIX + serviceIdentifier;
  88 + }
  89 +
  90 +}
0 91 \ No newline at end of file
... ...