```

feat(mqtt): 集成Redis和MongoDB客户端并优化MQTT连接逻辑 - 移除application.properties中硬编码的Redis配置 - 在mqtt-vertx模块中添加Redis和MongoDB依赖 - 在mqtt-web模块中添加Redis和MongoDB Spring Boot Starter - 重构MqttVerticle中的连接和发布逻辑，使用响应式API处理连接和消息发送 - 优化错误处理机制，添加连接失败和消息发送失败的回调 - 将VerticleLifeCycle中的Vertx注入方式改为@Resource注解 - 调整JsonUtils导入顺序 ```

```
feat(mqtt): 集成Redis和MongoDB客户端并优化MQTT连接逻辑 - 移除application.properties中硬编码的Redis配置 - 在mqtt-vertx模块中添加Redis和MongoDB依赖 - 在mqtt-web模块中添加Redis和MongoDB Spring Boot Starter - 重构MqttVerticle中的连接和发布逻辑，使用响应式API处理连接和消息发送 - 优化错误处理机制，添加连接失败和消息发送失败的回调 - 将VerticleLifeCycle中的Vertx注入方式改为@Resource注解 - 调整JsonUtils导入顺序 ```
zhangmeiyang
1 parent 152d3a4e
Showing 5 changed files with 40 additions and 31 deletions
mqtt-boot/src/main/resources/application.properties
mqtt-vertx/pom.xml
mqtt-vertx/src/main/java/com/diligrp/mqtt/vertx/config/VerticleLifeCycle.java
mqtt-vertx/src/main/java/com/diligrp/mqtt/vertx/verticle/MqttVerticle.java
mqtt-web/pom.xml
@@ -9,12 +9,3 @@ spring.cloud.nacos.config.server-addr=nacos.diligrp.com:8848
 spring.cloud.nacos.config.namespace=2267e673-b41f-458d-9643-2a03e4fd92fb
 spring.config.import[0]=nacos:${spring.application.name}.properties
 spring.config.import[1]=nacos:${spring.application.name}-${spring.profiles.active}.properties
-
-
-spring.data.redis.host=redis.diligrp.com
-spring.data.redis.port=6379
-spring.data.redis.database=12
-#spring.data.redis.username=
-#spring.data.redis.password=
-spring.data.redis.connect-timeout=15000
-spring.data.redis.timeout=30000
@@ -28,6 +28,14 @@
             <groupId>io.vertx</groupId>
             <artifactId>vertx-mqtt</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.vertx</groupId>
+            <artifactId>vertx-redis-client</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.vertx</groupId>
+            <artifactId>vertx-mongo-client</artifactId>
+        </dependency>
     </dependencies>
  
 </project>
@@ -2,7 +2,12 @@ package com.diligrp.mqtt.vertx.config;
  
 import com.diligrp.mqtt.vertx.verticle.MqttVerticle;
 import io.vertx.core.Vertx;
+import io.vertx.core.json.JsonObject;
+import io.vertx.ext.mongo.MongoClient;
+import io.vertx.redis.client.Redis;
+import jakarta.annotation.Resource;
 import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.CommandLineRunner;
 import org.springframework.stereotype.Component;
  
@@ -15,11 +20,8 @@ import org.springframework.stereotype.Component;
 @Component
 public class VerticleLifeCycle implements CommandLineRunner, DisposableBean {
  
-    private final Vertx vertx;
-
-    public VerticleLifeCycle(Vertx vertx) {
-        this.vertx = vertx;
-    }
+    @Resource
+    private Vertx vertx;
  
     @Override
     public void run(String... args) throws Exception {
 package com.diligrp.mqtt.vertx.verticle;
  
-import com.diligrp.mqtt.core.util.JsonUtils;
 import com.diligrp.mqtt.core.model.Printer;
+import com.diligrp.mqtt.core.util.JsonUtils;
 import io.vertx.core.AbstractVerticle;
 import io.vertx.core.Promise;
 import io.vertx.core.buffer.Buffer;
@@ -37,23 +37,23 @@ public class MqttVerticle extends AbstractVerticle {
         });
     }
  
-    private static void doSend(MqttClient client, Printer printer, EventBus eventBus) {
-        client.connect(printer.getPort(), printer.getHost(), connResult -> {
-            if (connResult.succeeded()) {
-                // 连接成功后发布消息
-                client.publish(printer.getTopic(), Buffer.buffer(printer.getData()), printer.getQos(), false, false, pubResult -> {
-                    if (pubResult.succeeded()) {
-                        LOGGER.info("Message published");
-                        eventBus.send(printer.getSuccessCallBackEventBus(), JsonUtils.toJsonString(printer));
-                    } else {
-                        LOGGER.error("Failed to publish message", pubResult.cause());
-                        eventBus.send(printer.getFailCallBackEventBus(), JsonUtils.toJsonString(printer));
-                    }
+    private void doSend(MqttClient client, Printer printer, EventBus eventBus) {
+        client.connect(printer.getPort(), printer.getHost())
+                .onFailure(throwable -> {
+                    LOGGER.error("连接到[MQTT-SERVER]失败", throwable);
+                    eventBus.send(PRINTER_TOPIC, JsonUtils.toJsonString(printer));
+                })
+                .onSuccess(connResult -> {
+                    client.publish(printer.getTopic(), Buffer.buffer(printer.getData()), printer.getQos(), false, false)
+                            .onFailure(throwable -> {
+                                LOGGER.error("MQTT消息发送失败", throwable);
+                                eventBus.send(printer.getFailCallBackEventBus(), JsonUtils.toJsonString(printer));
+                            })
+                            .onSuccess(pubResult -> {
+                                LOGGER.info("MQTT消息发送成功");
+                                eventBus.send(printer.getSuccessCallBackEventBus(), JsonUtils.toJsonString(printer));
+                            });
                     client.disconnect();
                 });
-            } else {
-                LOGGER.error("Failed to connect to MQTT server", connResult.cause());
-            }
-        });
     }
 }
@@ -28,6 +28,14 @@
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-validation</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-data-mongodb</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-data-redis</artifactId>
+        </dependency>
     </dependencies>
  
 </project>