BJ_ZTEITS-Developers / parking-cloud-yudao

Commit 85c1b05bcaa51cad318ee2ca35ddf063dfd6bea0

Authored by YunaiV
1 parent e2b995d4

feat（iot）：【协议改造】emqx 初步改造（50%）：修复 Review Agent 反馈的 bug（增强稳定性）

Inline Side-by-side
Showing 1 changed file with 44 additions and 27 deletions
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/IotEmqxProtocol.java
  View file @85c1b05
... ... @@ -176,7 +176,8 @@ public class IotEmqxProtocol implements IotProtocol {
176 176 // 2.4 关闭 Vertx
177 177 if (vertx != null) {
178 178 try {
179   - vertx.close().result();
  179 + vertx.close().toCompletionStage().toCompletableFuture()
  180 + .get(10, TimeUnit.SECONDS);
180 181 log.info("[stop][IoT EMQX 协议 {} Vertx 已关闭]", getId());
181 182 } catch (Exception e) {
182 183 log.error("[stop][IoT EMQX 协议 {} Vertx 关闭失败]", getId(), e);
... ... @@ -195,7 +196,7 @@ public class IotEmqxProtocol implements IotProtocol {
195 196 private void startHttpServer() {
196 197 // 1. 创建路由
197 198 Router router = Router.router(vertx);
198   - router.route().handler(BodyHandler.create());
  199 + router.route().handler(BodyHandler.create().setBodyLimit(1024 * 1024)); // 限制 body 大小为 1MB，防止大包攻击
199 200  
200 201 // 2. 创建处理器
201 202 IotEmqxAuthEventHandler handler = new IotEmqxAuthEventHandler(serverId);
... ... @@ -218,12 +219,13 @@ public class IotEmqxProtocol implements IotProtocol {
218 219 httpServer = vertx.createHttpServer(options)
219 220 .requestHandler(router)
220 221 .listen()
221   - .result();
  222 + .toCompletionStage().toCompletableFuture()
  223 + .get(10, TimeUnit.SECONDS);
222 224 log.info("[startHttpServer][IoT EMQX 协议 {} HTTP Hook 服务启动成功, port: {}, ssl: {}]",
223 225 getId(), properties.getPort(), httpConfig != null && Boolean.TRUE.equals(httpConfig.getSslEnabled()));
224 226 } catch (Exception e) {
225 227 log.error("[startHttpServer][IoT EMQX 协议 {} HTTP Hook 服务启动失败, port: {}]", getId(), properties.getPort(), e);
226   - throw e;
  228 + throw new RuntimeException("HTTP Hook 服务启动失败", e);
227 229 }
228 230 }
229 231  
... ... @@ -232,7 +234,8 @@ public class IotEmqxProtocol implements IotProtocol {
232 234 return;
233 235 }
234 236 try {
235   - httpServer.close().result();
  237 + httpServer.close().toCompletionStage().toCompletableFuture()
  238 + .get(5, TimeUnit.SECONDS);
236 239 log.info("[stopHttpServer][IoT EMQX 协议 {} HTTP Hook 服务已停止]", getId());
237 240 } catch (Exception e) {
238 241 log.error("[stopHttpServer][IoT EMQX 协议 {} HTTP Hook 服务停止失败]", getId(), e);
... ... @@ -258,19 +261,21 @@ public class IotEmqxProtocol implements IotProtocol {
258 261  
259 262 private void stopMqttClient() {
260 263 MqttClient client = this.mqttClient;
261   - if (client == null || !client.isConnected()) {
  264 + this.mqttClient = null; // 先清理引用
  265 + if (client == null) {
262 266 return;
263 267 }
264   - this.mqttClient = null;
265 268  
266   - // 1. 批量取消订阅
267   - List<String> topicList = emqxConfig.getMqttTopics();
268   - if (CollUtil.isNotEmpty(topicList)) {
269   - try {
270   - client.unsubscribe(topicList).toCompletionStage().toCompletableFuture()
271   - .get(5, TimeUnit.SECONDS);
272   - } catch (Exception e) {
273   - log.warn("[stopMqttClient][IoT EMQX 协议 {} 取消订阅异常]", getId(), e);
  269 + // 1. 批量取消订阅（仅在连接时）
  270 + if (client.isConnected()) {
  271 + List<String> topicList = emqxConfig.getMqttTopics();
  272 + if (CollUtil.isNotEmpty(topicList)) {
  273 + try {
  274 + client.unsubscribe(topicList).toCompletionStage().toCompletableFuture()
  275 + .get(5, TimeUnit.SECONDS);
  276 + } catch (Exception e) {
  277 + log.warn("[stopMqttClient][IoT EMQX 协议 {} 取消订阅异常]", getId(), e);
  278 + }
274 279 }
275 280 }
276 281  
... ... @@ -296,15 +301,15 @@ public class IotEmqxProtocol implements IotProtocol {
296 301 .setClientId(emqxConfig.getMqttClientId())
297 302 .setUsername(emqxConfig.getMqttUsername())
298 303 .setPassword(emqxConfig.getMqttPassword())
299   - .setSsl(emqxConfig.getMqttSsl())
300   - .setCleanSession(emqxConfig.getCleanSession())
  304 + .setSsl(Boolean.TRUE.equals(emqxConfig.getMqttSsl()))
  305 + .setCleanSession(Boolean.TRUE.equals(emqxConfig.getCleanSession()))
301 306 .setKeepAliveInterval(emqxConfig.getKeepAliveIntervalSeconds())
302 307 .setMaxInflightQueue(emqxConfig.getMaxInflightQueue());
303 308 options.setConnectTimeout(emqxConfig.getConnectTimeoutSeconds() * 1000); // Vert.x 需要毫秒
304   - options.setTrustAll(emqxConfig.getTrustAll());
  309 + options.setTrustAll(Boolean.TRUE.equals(emqxConfig.getTrustAll()));
305 310 // 1.2 配置遗嘱消息
306 311 IotEmqxConfig.Will will = emqxConfig.getWill();
307   - if (will.isEnabled()) {
  312 + if (will != null && will.isEnabled()) {
308 313 Assert.notBlank(will.getTopic(), "遗嘱消息主题(emqx.will.topic)不能为空");
309 314 Assert.notNull(will.getPayload(), "遗嘱消息内容(emqx.will.payload)不能为空");
310 315 options.setWillFlag(true)
... ... @@ -313,9 +318,11 @@ public class IotEmqxProtocol implements IotProtocol {
313 318 .setWillQoS(will.getQos())
314 319 .setWillRetain(will.isRetain());
315 320 }
316   - // 1.3 配置高级 SSL/TLS（仅在启用 SSL 且不信任所有证书时生效）
317   - if (Boolean.TRUE.equals(emqxConfig.getMqttSsl()) && !Boolean.TRUE.equals(emqxConfig.getTrustAll())) {
318   - IotEmqxConfig.Ssl sslOptions = emqxConfig.getSslOptions();
  321 + // 1.3 配置高级 SSL/TLS（仅在启用 SSL 且不信任所有证书时生效，且需要 sslOptions 非空）
  322 + IotEmqxConfig.Ssl sslOptions = emqxConfig.getSslOptions();
  323 + if (Boolean.TRUE.equals(emqxConfig.getMqttSsl())
  324 + && Boolean.FALSE.equals(emqxConfig.getTrustAll())
  325 + && sslOptions != null) {
319 326 if (StrUtil.isNotBlank(sslOptions.getTrustStorePath())) {
320 327 options.setTrustStoreOptions(new JksOptions()
321 328 .setPath(sslOptions.getTrustStorePath())
... ... @@ -365,10 +372,11 @@ public class IotEmqxProtocol implements IotProtocol {
365 372 */
366 373 private void closeMqttClient() {
367 374 MqttClient oldClient = this.mqttClient;
368   - if (oldClient == null || !oldClient.isConnected()) {
  375 + this.mqttClient = null; // 先清理引用
  376 + if (oldClient == null) {
369 377 return;
370 378 }
371   - this.mqttClient = null;
  379 + // 尽力释放（无论是否连接都尝试 disconnect）
372 380 try {
373 381 oldClient.disconnect().toCompletionStage().toCompletableFuture()
374 382 .get(5, TimeUnit.SECONDS);
... ... @@ -391,7 +399,11 @@ public class IotEmqxProtocol implements IotProtocol {
391 399 return;
392 400 }
393 401 log.info("[startMqttClientReconnectChecker][IoT EMQX 协议 {} 检测到断开，尝试重连]", getId());
394   - tryReconnectMqttClient();
  402 + // 用 executeBlocking 避免阻塞 event-loop（tryReconnectMqttClient 内部有同步等待）
  403 + vertx.executeBlocking(() -> {
  404 + tryReconnectMqttClient();
  405 + return null;
  406 + });
395 407 });
396 408 }
397 409  
... ... @@ -449,7 +461,11 @@ public class IotEmqxProtocol implements IotProtocol {
449 461 return;
450 462 }
451 463 log.warn("[setupMqttClientHandlers][IoT EMQX 协议 {} 连接断开，立即尝试重连]", getId());
452   - vertx.runOnContext(v -> tryReconnectMqttClient());
  464 + // 用 executeBlocking 避免阻塞 event-loop（tryReconnectMqttClient 内部有同步等待）
  465 + vertx.executeBlocking(() -> {
  466 + tryReconnectMqttClient();
  467 + return null;
  468 + });
453 469 });
454 470  
455 471 // 2. 异常处理
... ... @@ -497,7 +513,8 @@ public class IotEmqxProtocol implements IotProtocol {
497 513 return;
498 514 }
499 515 MqttQoS qos = MqttQoS.valueOf(emqxConfig.getMqttQos());
500   - mqttClient.publish(topic, Buffer.buffer(payload), qos, false, false);
  516 + mqttClient.publish(topic, Buffer.buffer(payload), qos, false, false)
  517 + .onFailure(e -> log.error("[publishMessage][IoT EMQX 协议 {} 发布失败, topic: {}]", getId(), topic, e));
501 518 }
502 519  
503 520 }
... ...