BJ_ZTEITS-Developers / parking-cloud-yudao

Commit 2737ffa1162dbb92041119ebe0db4de8756f6780

Authored by haohao
1 parent 569eef4a

reactor：【IoT 物联网】重构 MQTT 客户端连接和订阅主题流程，更新配置属性

Inline Side-by-side
Showing 7 changed files with 308 additions and 127 deletions
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/config/IotGatewayProperties.java
  View file @2737ffa
1 1 package cn.iocoder.yudao.module.iot.gateway.config;
2 2  
  3 +import cn.hutool.core.util.StrUtil;
3 4 import jakarta.validation.constraints.NotEmpty;
4 5 import jakarta.validation.constraints.NotNull;
5 6 import lombok.Data;
... ... @@ -175,7 +176,7 @@ public class IotGatewayProperties {
175 176 * @return MQTT 客户端 ID
176 177 */
177 178 public String getMqttClientId() {
178   - if (cn.hutool.core.util.StrUtil.isBlank(mqttClientId)) {
  179 + if (StrUtil.isBlank(mqttClientId)) {
179 180 mqttClientId = "iot-gateway-mqtt-" + System.currentTimeMillis();
180 181 }
181 182 return mqttClientId;
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/IotMqttDownstreamSubscriber.java
  View file @2737ffa
1 1 package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt;
2 2  
3   -import cn.iocoder.yudao.module.iot.core.enums.IotDeviceMessageMethodEnum;
4 3 import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;
5 4 import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageSubscriber;
6 5 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
... ... @@ -45,13 +44,13 @@ public class IotMqttDownstreamSubscriber implements IotMessageSubscriber<IotDevi
45 44  
46 45 @Override
47 46 public void onMessage(IotDeviceMessage message) {
48   - log.info("[onMessage][接收到下行消息][messageId: {}][method: {}][deviceId: {}]",
  47 + log.debug("[onMessage][接收到下行消息, messageId: {}, method: {}, deviceId: {}]",
49 48 message.getId(), message.getMethod(), message.getDeviceId());
50 49 try {
51 50 // 1. 校验
52 51 String method = message.getMethod();
53 52 if (method == null) {
54   - log.warn("[onMessage][消息方法为空][messageId: {}][deviceId: {}]",
  53 + log.warn("[onMessage][消息方法为空, messageId: {}, deviceId: {}]",
55 54 message.getId(), message.getDeviceId());
56 55 return;
57 56 }
... ... @@ -59,20 +58,9 @@ public class IotMqttDownstreamSubscriber implements IotMessageSubscriber<IotDevi
59 58 // 2. 处理下行消息
60 59 downstreamHandler.handle(message);
61 60 } catch (Exception e) {
62   - log.error("[onMessage][处理下行消息失败][messageId: {}][method: {}][deviceId: {}]",
  61 + log.error("[onMessage][处理下行消息失败, messageId: {}, method: {}, deviceId: {}]",
63 62 message.getId(), message.getMethod(), message.getDeviceId(), e);
64 63 }
65 64 }
66 65  
67   - /**
68   - * 判断是否为上行消息
69   - *
70   - * @param method 消息方法
71   - * @return 是否为上行消息
72   - */
73   - private boolean isUpstreamMessage(String method) {
74   - IotDeviceMessageMethodEnum methodEnum = IotDeviceMessageMethodEnum.of(method);
75   - return methodEnum != null && methodEnum.getUpstream();
76   - }
77   -
78 66 }
79 67 \ No newline at end of file
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/IotMqttUpstreamProtocol.java
  View file @2737ffa
... ... @@ -21,7 +21,6 @@ import lombok.Getter;
21 21 import lombok.extern.slf4j.Slf4j;
22 22  
23 23 import java.util.List;
24   -import java.util.concurrent.TimeUnit;
25 24  
26 25 /**
27 26 * IoT 网关 MQTT 协议：接收设备上行消息
... ... @@ -61,10 +60,10 @@ public class IotMqttUpstreamProtocol {
61 60 @PostConstruct
62 61 public void start() {
63 62 if (isRunning) {
64   - log.warn("[start][MQTT 统一协议服务已经在运行中，请勿重复启动]");
  63 + log.warn("[start][MQTT 协议服务已经在运行中，请勿重复启动]");
65 64 return;
66 65 }
67   - log.info("[start][开始启动 MQTT 统一协议服务]");
  66 + log.info("[start][启动 MQTT 协议服务]");
68 67  
69 68 try {
70 69 // 1. 创建共享的 Vertx 实例
... ... @@ -77,9 +76,9 @@ public class IotMqttUpstreamProtocol {
77 76 startMqttClient();
78 77  
79 78 isRunning = true;
80   - log.info("[start][MQTT 统一协议服务启动完成]");
  79 + log.info("[start][MQTT 协议服务启动完成]");
81 80 } catch (Exception e) {
82   - log.error("[start][MQTT 统一协议服务启动失败]", e);
  81 + log.error("[start][MQTT 协议服务启动失败]", e);
83 82 // 启动失败时清理资源
84 83 stop();
85 84 throw e;
... ... @@ -89,10 +88,10 @@ public class IotMqttUpstreamProtocol {
89 88 @PreDestroy
90 89 public void stop() {
91 90 if (!isRunning) {
92   - log.warn("[stop][MQTT 统一协议服务已经停止，无需再次停止]");
  91 + log.warn("[stop][MQTT 协议服务已经停止，无需再次停止]");
93 92 return;
94 93 }
95   - log.info("[stop][开始停止 MQTT 统一协议服务]");
  94 + log.info("[stop][停止 MQTT 协议服务]");
96 95  
97 96 // 1. 停止 MQTT 客户端
98 97 stopMqttClient();
... ... @@ -104,28 +103,28 @@ public class IotMqttUpstreamProtocol {
104 103 if (vertx != null) {
105 104 try {
106 105 vertx.close();
107   - log.info("[stop][Vertx 实例已关闭]");
  106 + log.debug("[stop][Vertx 实例已关闭]");
108 107 } catch (Exception e) {
109 108 log.warn("[stop][关闭 Vertx 实例失败]", e);
110 109 }
111 110 }
112 111  
113 112 isRunning = false;
114   - log.info("[stop][MQTT 统一协议服务已停止]");
  113 + log.info("[stop][MQTT 协议服务已停止]");
115 114 }
116 115  
117 116 /**
118 117 * 启动 HTTP 认证服务
119 118 */
120 119 private void startHttpAuthServer() {
121   - log.info("[startHttpAuthServer][开始启动 HTTP 认证服务]");
  120 + log.info("[startHttpAuthServer][启动 HTTP 认证服务]");
122 121  
123 122 // 创建路由
124 123 Router router = Router.router(vertx);
125 124 router.route().handler(BodyHandler.create());
126 125  
127 126 // 创建认证处理器
128   - IotMqttHttpAuthHandler authHandler = new IotMqttHttpAuthHandler();
  127 + IotMqttHttpAuthHandler authHandler = new IotMqttHttpAuthHandler(this);
129 128 router.post(IotMqttTopicUtils.MQTT_AUTH_AUTHENTICATE_PATH).handler(authHandler::authenticate);
130 129 router.post(IotMqttTopicUtils.MQTT_AUTH_CONNECTED_PATH).handler(authHandler::connected);
131 130 router.post(IotMqttTopicUtils.MQTT_AUTH_DISCONNECTED_PATH).handler(authHandler::disconnected);
... ... @@ -137,7 +136,7 @@ public class IotMqttUpstreamProtocol {
137 136 .requestHandler(router)
138 137 .listen(authPort)
139 138 .result();
140   - log.info("[startHttpAuthServer][HTTP 认证服务启动成功，端口：{}]", authPort);
  139 + log.info("[startHttpAuthServer][HTTP 认证服务启动成功, 端口: {}]", authPort);
141 140 } catch (Exception e) {
142 141 log.error("[startHttpAuthServer][HTTP 认证服务启动失败]", e);
143 142 throw e;
... ... @@ -163,23 +162,24 @@ public class IotMqttUpstreamProtocol {
163 162 * 启动 MQTT 客户端
164 163 */
165 164 private void startMqttClient() {
166   - log.info("[startMqttClient][开始启动 MQTT 客户端]");
  165 + log.info("[startMqttClient][启动 MQTT 客户端]");
167 166  
168   - // 1. 初始化消息处理器
169   - this.upstreamHandler = new IotMqttUpstreamHandler(this);
  167 + try {
  168 + // 1. 初始化消息处理器
  169 + this.upstreamHandler = new IotMqttUpstreamHandler(this);
170 170  
171   - // 2. 创建 MQTT 客户端
172   - log.info("[startMqttClient][使用 MQTT 客户端 ID: {}]", emqxProperties.getMqttClientId());
  171 + // 2. 创建 MQTT 客户端
  172 + log.info("[startMqttClient][使用 MQTT 客户端 ID: {}]", emqxProperties.getMqttClientId());
  173 + createMqttClient();
173 174  
174   - MqttClientOptions options = new MqttClientOptions()
175   - .setClientId(emqxProperties.getMqttClientId())
176   - .setUsername(emqxProperties.getMqttUsername())
177   - .setPassword(emqxProperties.getMqttPassword())
178   - .setSsl(emqxProperties.getMqttSsl());
179   - this.mqttClient = MqttClient.create(vertx, options);
  175 + // 3. 连接 MQTT Broker（异步连接，不会抛出异常）
  176 + connectMqtt();
180 177  
181   - // 3. 连接 MQTT Broker
182   - connectMqtt();
  178 + log.info("[startMqttClient][MQTT 客户端启动完成，正在异步连接中...]");
  179 + } catch (Exception e) {
  180 + log.error("[startMqttClient][MQTT 客户端启动失败]", e);
  181 + throw new RuntimeException("MQTT 客户端启动失败", e);
  182 + }
183 183 }
184 184  
185 185 /**
... ... @@ -195,7 +195,7 @@ public class IotMqttUpstreamProtocol {
195 195 mqttClient.unsubscribe(topic);
196 196 log.debug("[stopMqttClient][取消订阅主题: {}]", topic);
197 197 } catch (Exception e) {
198   - log.warn("[stopMqttClient][取消订阅主题异常: {}]", topic, e);
  198 + log.warn("[stopMqttClient][取消订阅主题({})异常]", topic, e);
199 199 }
200 200 }
201 201 }
... ... @@ -216,43 +216,107 @@ public class IotMqttUpstreamProtocol {
216 216 * 连接 MQTT Broker 并订阅主题
217 217 */
218 218 private void connectMqtt() {
  219 + connectMqtt(false);
  220 + }
  221 +
  222 + /**
  223 + * 连接 MQTT Broker 并订阅主题
  224 + *
  225 + * @param isReconnect 是否为重连
  226 + */
  227 + private void connectMqtt(boolean isReconnect) {
219 228 // 1. 参数校验
220 229 String host = emqxProperties.getMqttHost();
221 230 Integer port = emqxProperties.getMqttPort();
222 231 if (StrUtil.isBlank(host)) {
223   - log.error("[connectMqtt][MQTT Host 为空，无法连接]");
  232 + log.error("[connectMqtt][MQTT Host 为空, 无法连接]");
224 233 throw new IllegalArgumentException("MQTT Host 不能为空");
225 234 }
226 235 if (port == null || port <= 0) {
227   - log.error("[connectMqtt][MQTT Port 无效：{}]", port);
  236 + log.error("[connectMqtt][MQTT Port({}) 无效]", port);
228 237 throw new IllegalArgumentException("MQTT Port 必须为正整数");
229 238 }
230   - log.info("[connectMqtt][开始连接 MQTT Broker][host: {}][port: {}]", host, port);
231 239  
232   - // 2. 连接
233   - try {
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 消息处理器设置完成]");
  240 + if (isReconnect) {
  241 + log.info("[connectMqtt][开始重连 MQTT Broker, host: {}, port: {}]", host, port);
  242 + // 重连时重新创建客户端实例
  243 + createMqttClient();
  244 + } else {
  245 + log.info("[connectMqtt][开始连接 MQTT Broker, host: {}, port: {}]", host, port);
  246 + }
249 247  
250   - // 4. 订阅主题
251   - subscribeToTopics();
252   - } catch (Exception e) {
253   - log.error("[connectMqtt][连接 MQTT Broker 失败][host: {}][port: {}]", host, port, e);
254   - reconnectWithDelay(); // 连接失败时，也要尝试重连
255   - throw new RuntimeException("MQTT 客户端启动失败", e);
  248 + // 2. 异步连接
  249 + mqttClient.connect(port, host, connectResult -> {
  250 + if (connectResult.succeeded()) {
  251 + if (isReconnect) {
  252 + log.info("[connectMqtt][MQTT 客户端重连成功, host: {}, port: {}]", host, port);
  253 + } else {
  254 + log.info("[connectMqtt][MQTT 客户端连接成功, host: {}, port: {}]", host, port);
  255 + }
  256 +
  257 + // 3. 设置处理器
  258 + setupMqttHandlers();
  259 +
  260 + // 4. 订阅主题
  261 + subscribeToTopics();
  262 +
  263 + } else {
  264 + log.error("[connectMqtt][连接 MQTT Broker 失败, host: {}, port: {}, isReconnect: {}]",
  265 + host, port, isReconnect, connectResult.cause());
  266 +
  267 + if (!isReconnect) {
  268 + // 首次连接失败时，也要尝试重连
  269 + log.warn("[connectMqtt][首次连接失败，将开始重连机制]");
  270 + reconnectWithDelay();
  271 + } else {
  272 + // 重连失败时，继续尝试重连
  273 + reconnectWithDelay();
  274 + }
  275 + }
  276 + });
  277 + }
  278 +
  279 + /**
  280 + * 创建 MQTT 客户端
  281 + */
  282 + private void createMqttClient() {
  283 + log.debug("[createMqttClient][创建 MQTT 客户端, clientId: {}]", emqxProperties.getMqttClientId());
  284 +
  285 + MqttClientOptions options = new MqttClientOptions()
  286 + .setClientId(emqxProperties.getMqttClientId())
  287 + .setUsername(emqxProperties.getMqttUsername())
  288 + .setPassword(emqxProperties.getMqttPassword())
  289 + .setSsl(emqxProperties.getMqttSsl());
  290 +
  291 + this.mqttClient = MqttClient.create(vertx, options);
  292 + }
  293 +
  294 + /**
  295 + * 设置 MQTT 处理器
  296 + */
  297 + private void setupMqttHandlers() {
  298 + if (mqttClient == null) {
  299 + log.warn("[setupMqttHandlers][MQTT 客户端为空，跳过处理器设置]");
  300 + return;
  301 + }
  302 +
  303 + // 设置断开重连监听器
  304 + mqttClient.closeHandler(closeEvent -> {
  305 + log.warn("[closeHandler][MQTT 连接已断开, 准备重连]");
  306 + reconnectWithDelay();
  307 + });
  308 +
  309 + // 设置异常处理器
  310 + mqttClient.exceptionHandler(exception -> {
  311 + log.error("[exceptionHandler][MQTT 客户端异常]", exception);
  312 + });
  313 +
  314 + // 设置消息处理器
  315 + if (upstreamHandler != null) {
  316 + mqttClient.publishHandler(upstreamHandler::handle);
  317 + log.debug("[setupMqttHandlers][MQTT 消息处理器设置完成]");
  318 + } else {
  319 + log.warn("[setupMqttHandlers][上行消息处理器为空，跳过设置]");
256 320 }
257 321 }
258 322  
... ... @@ -261,15 +325,43 @@ public class IotMqttUpstreamProtocol {
261 325 */
262 326 private void subscribeToTopics() {
263 327 List<String> topicList = emqxProperties.getMqttTopics();
  328 + if (CollUtil.isEmpty(topicList)) {
  329 + log.warn("[subscribeToTopics][订阅主题列表为空, 跳过订阅]");
  330 + return;
  331 + }
  332 +
  333 + if (mqttClient == null || !mqttClient.isConnected()) {
  334 + log.warn("[subscribeToTopics][MQTT 客户端未连接, 跳过订阅]");
  335 + return;
  336 + }
  337 +
264 338 int qos = emqxProperties.getMqttQos();
  339 + log.info("[subscribeToTopics][开始订阅主题, 共 {} 个, QoS: {}]", topicList.size(), qos);
  340 +
  341 + int[] successCount = {0}; // 使用数组以便在 lambda 中修改
  342 + int[] failCount = {0};
265 343  
266   - log.info("[subscribeToTopics][开始订阅主题，共 {} 个]", topicList.size());
267 344 for (String topic : topicList) {
268 345 mqttClient.subscribe(topic, qos, subscribeResult -> {
269 346 if (subscribeResult.succeeded()) {
270   - log.info("[subscribeToTopics][订阅主题成功: {}][QoS: {}]", topic, qos);
  347 + successCount[0]++;
  348 + log.debug("[subscribeToTopics][订阅主题成功, topic: {}, qos: {}]", topic, qos);
  349 +
  350 + // 当所有主题都处理完成时，记录汇总日志
  351 + if (successCount[0] + failCount[0] == topicList.size()) {
  352 + log.info("[subscribeToTopics][主题订阅完成, 成功: {}, 失败: {}, 总计: {}]",
  353 + successCount[0], failCount[0], topicList.size());
  354 + }
271 355 } else {
272   - log.error("[subscribeToTopics][订阅主题失败: {}]", topic, subscribeResult.cause());
  356 + failCount[0]++;
  357 + log.error("[subscribeToTopics][订阅主题失败, topic: {}, qos: {}, 原因: {}]",
  358 + topic, qos, subscribeResult.cause().getMessage(), subscribeResult.cause());
  359 +
  360 + // 当所有主题都处理完成时，记录汇总日志
  361 + if (successCount[0] + failCount[0] == topicList.size()) {
  362 + log.info("[subscribeToTopics][主题订阅完成, 成功: {}, 失败: {}, 总计: {}]",
  363 + successCount[0], failCount[0], topicList.size());
  364 + }
273 365 }
274 366 });
275 367 }
... ... @@ -281,15 +373,21 @@ public class IotMqttUpstreamProtocol {
281 373 private void reconnectWithDelay() {
282 374 long delay = emqxProperties.getReconnectDelayMs();
283 375 vertx.setTimer(delay, timerId -> {
284   - if (!isRunning || (mqttClient != null && mqttClient.isConnected())) {
  376 + if (!isRunning) {
  377 + log.debug("[reconnectWithDelay][服务已停止, 取消重连]");
  378 + return;
  379 + }
  380 + // 检查连接状态，如果已连接则无需重连
  381 + if (mqttClient != null && mqttClient.isConnected()) {
  382 + log.debug("[reconnectWithDelay][MQTT 客户端已连接, 无需重连]");
285 383 return;
286 384 }
287   - log.info("[reconnectWithDelay][开始重连 MQTT Broker，延迟 {} 毫秒]", delay);
  385 + log.info("[reconnectWithDelay][开始重连 MQTT Broker, 延迟: {} ms]", delay);
288 386 try {
289   - connectMqtt();
  387 + connectMqtt(true); // 标记为重连
290 388 } catch (Exception e) {
291   - log.error("[reconnectWithDelay][重连失败，将继续尝试重连]", e);
292   - reconnectWithDelay(); // 失败后，继续尝试
  389 + log.error("[reconnectWithDelay][重连失败, 将继续尝试]", e);
  390 + // 重连失败时，不需要重复调用，因为 connectMqtt(true) 内部已经处理了重连逻辑
293 391 }
294 392 });
295 393 }
... ... @@ -302,7 +400,7 @@ public class IotMqttUpstreamProtocol {
302 400 */
303 401 public void publishMessage(String topic, String payload) {
304 402 if (mqttClient == null || !mqttClient.isConnected()) {
305   - log.warn("[publishMessage][MQTT 客户端未连接，无法发布消息][topic: {}]", topic);
  403 + log.warn("[publishMessage][MQTT 客户端未连接, 无法发布消息到 topic({})]", topic);
306 404 return;
307 405 }
308 406 MqttQoS qos = MqttQoS.valueOf(emqxProperties.getMqttQos());
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttDownstreamHandler.java
  View file @2737ffa
... ... @@ -41,14 +41,14 @@ public class IotMqttDownstreamHandler {
41 41 // 1. 获取设备信息（使用缓存）
42 42 IotDeviceRespDTO deviceInfo = deviceService.getDeviceFromCache(message.getDeviceId());
43 43 if (deviceInfo == null) {
44   - log.warn("[handle][设备信息不存在][deviceId: {}]", message.getDeviceId());
  44 + log.warn("[handle][设备信息不存在, deviceId: {}]", message.getDeviceId());
45 45 return;
46 46 }
47 47  
48 48 // 2. 根据方法构建主题
49 49 String topic = buildTopicByMethod(message.getMethod(), deviceInfo.getProductKey(), deviceInfo.getDeviceName());
50 50 if (StrUtil.isBlank(topic)) {
51   - log.warn("[handle][未知的消息方法：{}]", message.getMethod());
  51 + log.warn("[handle][未知的消息方法: {}]", message.getMethod());
52 52 return;
53 53 }
54 54  
... ... @@ -57,7 +57,7 @@ public class IotMqttDownstreamHandler {
57 57  
58 58 // 4. 发布消息
59 59 protocol.publishMessage(topic, payload.toString());
60   - log.info("[handle][发布下行消息成功][method: {}][topic: {}]", message.getMethod(), topic);
  60 + log.debug("[handle][发布下行消息成功, method: {}, topic: {}]", message.getMethod(), topic);
61 61 }
62 62  
63 63 /**
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttHttpAuthHandler.java
  View file @2737ffa
... ... @@ -8,6 +8,7 @@ import cn.iocoder.yudao.module.iot.core.biz.IotDeviceCommonApi;
8 8 import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceAuthReqDTO;
9 9 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
10 10 import cn.iocoder.yudao.module.iot.core.util.IotDeviceAuthUtils;
  11 +import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttUpstreamProtocol;
11 12 import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
12 13 import io.vertx.core.json.JsonObject;
13 14 import io.vertx.ext.web.RoutingContext;
... ... @@ -46,14 +47,27 @@ public class IotMqttHttpAuthHandler {
46 47 private static final int INTERNAL_ERROR_STATUS_CODE = 500;
47 48  
48 49 /**
  50 + * MQTT 协议实例，用于获取服务器ID
  51 + */
  52 + private final IotMqttUpstreamProtocol protocol;
  53 +
  54 + /**
  55 + * 构造器
  56 + *
  57 + * @param protocol MQTT 协议实例
  58 + */
  59 + public IotMqttHttpAuthHandler(IotMqttUpstreamProtocol protocol) {
  60 + this.protocol = protocol;
  61 + }
  62 +
  63 + /**
49 64 * EMQX 认证接口
50 65 */
51 66 public void authenticate(RoutingContext context) {
52 67 try {
53 68 // 解析请求体
54   - JsonObject body = context.body().asJsonObject();
  69 + JsonObject body = parseRequestBody(context);
55 70 if (body == null) {
56   - sendErrorResponse(context, 400, "请求体不能为空");
57 71 return;
58 72 }
59 73  
... ... @@ -61,34 +75,23 @@ public class IotMqttHttpAuthHandler {
61 75 String username = body.getString("username");
62 76 String password = body.getString("password");
63 77  
64   - log.info("[authenticate][EMQX 设备认证请求][clientId: {}][username: {}]", clientid, username);
  78 + log.debug("[authenticate][EMQX 设备认证, clientId: {}, username: {}]", clientid, username);
65 79  
66 80 // 参数校验
67   - if (StrUtil.isEmpty(clientid) || StrUtil.isEmpty(username) || StrUtil.isEmpty(password)) {
68   - log.warn("[authenticate][认证参数不完整][clientId: {}][username: {}]", clientid, username);
69   - sendErrorResponse(context, BAD_REQUEST_STATUS_CODE, "认证参数不完整");
  81 + if (!validateAuthParams(context, clientid, username, password)) {
70 82 return;
71 83 }
72 84  
73 85 // 执行设备认证
74   - IotDeviceCommonApi deviceApi = SpringUtil.getBean(IotDeviceCommonApi.class);
75   - CommonResult<Boolean> result = deviceApi.authDevice(new IotDeviceAuthReqDTO()
76   - .setClientId(clientid)
77   - .setUsername(username)
78   - .setPassword(password));
79   -
80   - result.checkError();
81   - if (!BooleanUtil.isTrue(result.getData())) {
82   - log.warn("[authenticate][设备认证失败][clientId: {}][username: {}]", clientid, username);
83   - sendErrorResponse(context, UNAUTHORIZED_STATUS_CODE, DEVICE_AUTH_FAIL.getMsg());
  86 + if (!performDeviceAuth(context, clientid, username, password)) {
84 87 return;
85 88 }
86 89  
87   - log.info("[authenticate][设备认证成功][clientId: {}][username: {}]", clientid, username);
  90 + log.debug("[authenticate][设备认证成功, clientId: {}, username: {}]", clientid, username);
88 91 sendSuccessResponse(context, "认证成功");
89 92  
90 93 } catch (Exception e) {
91   - log.error("[authenticate][设备认证异常]", e);
  94 + log.error("[authenticate][设备认证异常, 详细信息: {}]", e.getMessage(), e);
92 95 sendErrorResponse(context, INTERNAL_ERROR_STATUS_CODE, "认证服务异常");
93 96 }
94 97 }
... ... @@ -99,9 +102,8 @@ public class IotMqttHttpAuthHandler {
99 102 public void connected(RoutingContext context) {
100 103 try {
101 104 // 解析请求体
102   - JsonObject body = context.body().asJsonObject();
  105 + JsonObject body = parseRequestBody(context);
103 106 if (body == null) {
104   - sendErrorResponse(context, BAD_REQUEST_STATUS_CODE, "请求体不能为空");
105 107 return;
106 108 }
107 109  
... ... @@ -109,13 +111,14 @@ public class IotMqttHttpAuthHandler {
109 111 String username = body.getString("username");
110 112 Long timestamp = body.getLong("timestamp");
111 113  
112   - log.info("[connected][设备连接事件][clientId: {}][username: {}]", clientid, username);
  114 + log.debug("[connected][设备连接, clientId: {}, username: {}, timestamp: {}]",
  115 + clientid, username, timestamp);
113 116  
114   - handleDeviceStateChange(username, true);
  117 + handleDeviceStateChange(username, true, "设备连接");
115 118 sendSuccessResponse(context, "处理成功");
116 119  
117 120 } catch (Exception e) {
118   - log.error("[connected][处理设备连接事件失败]", e);
  121 + log.error("[connected][处理设备连接事件失败, 详细信息: {}]", e.getMessage(), e);
119 122 sendErrorResponse(context, INTERNAL_ERROR_STATUS_CODE, "处理失败");
120 123 }
121 124 }
... ... @@ -126,9 +129,8 @@ public class IotMqttHttpAuthHandler {
126 129 public void disconnected(RoutingContext context) {
127 130 try {
128 131 // 解析请求体
129   - JsonObject body = context.body().asJsonObject();
  132 + JsonObject body = parseRequestBody(context);
130 133 if (body == null) {
131   - sendErrorResponse(context, BAD_REQUEST_STATUS_CODE, "请求体不能为空");
132 134 return;
133 135 }
134 136  
... ... @@ -137,58 +139,144 @@ public class IotMqttHttpAuthHandler {
137 139 String reason = body.getString("reason");
138 140 Long timestamp = body.getLong("timestamp");
139 141  
140   - log.info("[disconnected][设备断开连接事件][clientId: {}][username: {}][reason: {}]",
141   - clientid, username, reason);
  142 + log.debug("[disconnected][设备断开连接, clientId: {}, username: {}, reason: {}, timestamp: {}]",
  143 + clientid, username, reason, timestamp);
142 144  
143   - handleDeviceStateChange(username, false);
  145 + handleDeviceStateChange(username, false, "设备断开连接，原因：" + reason);
144 146 sendSuccessResponse(context, "处理成功");
145 147  
146 148 } catch (Exception e) {
147   - log.error("[disconnected][处理设备断开连接事件失败]", e);
  149 + log.error("[disconnected][处理设备断开连接事件失败, 详细信息: {}]", e.getMessage(), e);
148 150 sendErrorResponse(context, INTERNAL_ERROR_STATUS_CODE, "处理失败");
149 151 }
150 152 }
151 153  
152 154 /**
153   - * 处理设备状态变化
  155 + * 解析请求体
  156 + *
  157 + * @param context 路由上下文
  158 + * @return 请求体JSON对象，解析失败时返回null
  159 + */
  160 + private JsonObject parseRequestBody(RoutingContext context) {
  161 + try {
  162 + JsonObject body = context.body().asJsonObject();
  163 + if (body == null) {
  164 + log.warn("[parseRequestBody][请求体为空]");
  165 + sendErrorResponse(context, BAD_REQUEST_STATUS_CODE, "请求体不能为空");
  166 + return null;
  167 + }
  168 + return body;
  169 + } catch (Exception e) {
  170 + log.error("[parseRequestBody][解析请求体失败]", e);
  171 + sendErrorResponse(context, BAD_REQUEST_STATUS_CODE, "请求体格式错误");
  172 + return null;
  173 + }
  174 + }
  175 +
  176 + /**
  177 + * 验证认证参数
  178 + *
  179 + * @param context 路由上下文
  180 + * @param clientid 客户端ID
  181 + * @param username 用户名
  182 + * @param password 密码
  183 + * @return 验证是否通过
  184 + */
  185 + private boolean validateAuthParams(RoutingContext context, String clientid, String username, String password) {
  186 + if (StrUtil.hasEmpty(clientid, username, password)) {
  187 + log.warn("[validateAuthParams][认证参数不完整, clientId: {}, username: {}, password: {}]",
  188 + clientid, username, StrUtil.isNotEmpty(password) ? "***" : "空");
  189 + sendErrorResponse(context, BAD_REQUEST_STATUS_CODE, "认证参数不完整");
  190 + return false;
  191 + }
  192 + return true;
  193 + }
  194 +
  195 + /**
  196 + * 执行设备认证
154 197 *
  198 + * @param context 路由上下文
  199 + * @param clientid 客户端ID
155 200 * @param username 用户名
156   - * @param online 是否在线
  201 + * @param password 密码
  202 + * @return 认证是否成功
157 203 */
158   - private void handleDeviceStateChange(String username, boolean online) {
  204 + private boolean performDeviceAuth(RoutingContext context, String clientid, String username, String password) {
  205 + try {
  206 + IotDeviceCommonApi deviceApi = SpringUtil.getBean(IotDeviceCommonApi.class);
  207 + CommonResult<Boolean> result = deviceApi.authDevice(new IotDeviceAuthReqDTO()
  208 + .setClientId(clientid)
  209 + .setUsername(username)
  210 + .setPassword(password));
  211 +
  212 + result.checkError();
  213 + if (!BooleanUtil.isTrue(result.getData())) {
  214 + log.warn("[performDeviceAuth][设备认证失败, clientId: {}, username: {}]", clientid, username);
  215 + sendErrorResponse(context, UNAUTHORIZED_STATUS_CODE, DEVICE_AUTH_FAIL.getMsg());
  216 + return false;
  217 + }
  218 + return true;
  219 + } catch (Exception e) {
  220 + log.error("[performDeviceAuth][设备认证异常, clientId: {}, username: {}]", clientid, username, e);
  221 + sendErrorResponse(context, INTERNAL_ERROR_STATUS_CODE, "认证服务异常");
  222 + return false;
  223 + }
  224 + }
  225 +
  226 + /**
  227 + * 处理设备状态变化
  228 + *
  229 + * @param username 用户名
  230 + * @param online 是否在线
  231 + * @param actionDesc 操作描述
  232 + */
  233 + private void handleDeviceStateChange(String username, boolean online, String actionDesc) {
159 234 if (StrUtil.isEmpty(username) || "undefined".equals(username)) {
160   - log.warn("[handleDeviceStateChange][用户名为空或未定义][username: {}]", username);
  235 + log.warn("[handleDeviceStateChange][用户名为空或'undefined', username: {}, action: {}]",
  236 + username, actionDesc);
161 237 return;
162 238 }
163 239  
164 240 // 解析设备信息
165 241 IotDeviceAuthUtils.DeviceInfo deviceInfo = IotDeviceAuthUtils.parseUsername(username);
166 242 if (deviceInfo == null) {
167   - log.warn("[handleDeviceStateChange][无法解析设备信息][username: {}]", username);
  243 + log.warn("[handleDeviceStateChange][无法从 username({}) 解析设备信息, action: {}]",
  244 + username, actionDesc);
168 245 return;
169 246 }
170 247  
171 248 try {
172   - // 获取服务器 ID
173   - String serverId = "mqtt_auth_gateway";
  249 + // 从协议实例获取服务器 ID
  250 + String serverId = protocol.getServerId();
  251 + if (StrUtil.isEmpty(serverId)) {
  252 + log.error("[handleDeviceStateChange][获取服务器ID失败, username: {}, action: {}]",
  253 + username, actionDesc);
  254 + return;
  255 + }
174 256  
175 257 // 构建设备状态消息
176 258 IotDeviceMessageService deviceMessageService = SpringUtil.getBean(IotDeviceMessageService.class);
177 259 IotDeviceMessage message;
178 260 if (online) {
179 261 message = IotDeviceMessage.buildStateOnline();
180   - log.info("[handleDeviceStateChange][发送设备上线消息成功][username: {}]", username);
  262 + log.debug("[handleDeviceStateChange][发送设备上线消息, username: {}, serverId: {}]",
  263 + username, serverId);
181 264 } else {
182 265 message = IotDeviceMessage.buildStateOffline();
183   - log.info("[handleDeviceStateChange][发送设备下线消息成功][username: {}]", username);
  266 + log.debug("[handleDeviceStateChange][发送设备下线消息, username: {}, serverId: {}]",
  267 + username, serverId);
184 268 }
185 269  
186 270 // 发送消息到消息总线
187 271 deviceMessageService.sendDeviceMessage(message,
188 272 deviceInfo.getProductKey(), deviceInfo.getDeviceName(), serverId);
  273 +
  274 + log.info("[handleDeviceStateChange][{}处理成功, productKey: {}, deviceName: {}, serverId: {}]",
  275 + actionDesc, deviceInfo.getProductKey(), deviceInfo.getDeviceName(), serverId);
  276 +
189 277 } catch (Exception e) {
190   - log.error("[handleDeviceStateChange][发送设备状态消息失败][username: {}][online: {}]",
191   - username, online, e);
  278 + log.error("[handleDeviceStateChange][发送设备状态消息失败, username: {}, online: {}, action: {}]",
  279 + username, online, actionDesc, e);
192 280 }
193 281 }
194 282  
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttUpstreamHandler.java
  View file @2737ffa
... ... @@ -9,6 +9,8 @@ import io.vertx.mqtt.messages.MqttPublishMessage;
9 9 import lombok.extern.slf4j.Slf4j;
10 10 import org.springframework.util.Assert;
11 11  
  12 +import java.util.Arrays;
  13 +
12 14 /**
13 15 * IoT 网关 MQTT 上行消息处理器
14 16 *
... ... @@ -32,12 +34,12 @@ public class IotMqttUpstreamHandler {
32 34 String topic = message.topicName();
33 35 byte[] payload = message.payload().getBytes();
34 36  
35   - log.debug("[handle][收到 MQTT 消息][topic: {}]", topic);
  37 + log.debug("[handle][收到 MQTT 消息, topic: {}]", topic);
36 38  
37 39 try {
38 40 // 1. 前置校验
39 41 if (StrUtil.isBlank(topic)) {
40   - log.warn("[validateInput][主题为空，忽略消息]");
  42 + log.warn("[handle][主题为空, 忽略消息]");
41 43 return;
42 44 }
43 45 // 注意：payload 可以为空
... ... @@ -45,7 +47,7 @@ public class IotMqttUpstreamHandler {
45 47 // 2. 识别并验证消息类型
46 48 String messageType = getMessageType(topic);
47 49 Assert.notNull(messageType, String.format("未知的消息类型, topic(%s)", topic));
48   - log.info("[handle][接收到{}][topic: {}]", messageType, topic);
  50 + log.debug("[handle][接收到上行消息({}), topic: {}]", messageType, topic);
49 51  
50 52 // 3. 解析主题，获取 productKey 和 deviceName
51 53 String[] topicParts = topic.split("/");
... ... @@ -72,7 +74,7 @@ public class IotMqttUpstreamHandler {
72 74 deviceMessageService.sendDeviceMessage(deviceMessage, productKey, deviceName, serverId);
73 75  
74 76 // 6. 记录成功日志
75   - log.info("[handle][处理{}成功，已转发到 MQ][topic: {}]", messageType, topic);
  77 + log.debug("[handle][处理上行消息({})成功, topic: {}]", messageType, topic);
76 78 } catch (Exception e) {
77 79 log.error("[handle][处理 MQTT 消息失败][topic: {}][payload: {}]", topic, new String(payload), e);
78 80 }
... ... @@ -86,10 +88,13 @@ public class IotMqttUpstreamHandler {
86 88 */
87 89 private String getMessageType(String topic) {
88 90 String[] topicParts = topic.split("/");
89   - if (topicParts.length < 7) {
90   - return null;
  91 + // 约定：topic 第 4 个部分开始为消息类型
  92 + // 例如：/sys/{productKey}/{deviceName}/thing/property/post ->
  93 + // thing/property/post
  94 + if (topicParts.length > 4) {
  95 + return String.join("/", Arrays.copyOfRange(topicParts, 4, topicParts.length));
91 96 }
92   - return topicParts[3];
  97 + return topicParts[topicParts.length - 1];
93 98 }
94 99  
95 100 }
96 101 \ No newline at end of file
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/resources/application-local.yaml
  View file @2737ffa
... ... @@ -37,6 +37,7 @@ yudao:
37 37 mqtt-port: 1883 # MQTT Broker 端口
38 38 mqtt-username: admin # MQTT 用户名
39 39 mqtt-password: public # MQTT 密码
  40 + mqtt-client-id: iot-gateway-mqtt # MQTT 客户端 ID
40 41 mqtt-ssl: false # 是否开启 SSL
41 42 mqtt-topics:
42 43 - "/sys/#" # 系统主题
... ...