修改：优化 EmqxService 实现

安浩浩
1 parent b47176c9
Showing 3 changed files with 52 additions and 22 deletions
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/emq/client/EmqxClient.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/emq/service/EmqxService.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/emq/service/EmqxServiceImpl.java
@@ -43,7 +43,7 @@ public class EmqxClient {
     private void createMqttClient() {
         try {
-            mqttClient = new MqttClient(mqttConfig.getHostUrl(), "yudao-" + mqttConfig.getClientId(), new MemoryPersistence());
+            mqttClient = new MqttClient(mqttConfig.getHostUrl(), "yudao" + mqttConfig.getClientId(), new MemoryPersistence());
             mqttClient.setCallback(emqxCallback);
         } catch (MqttException e) {
             log.error("创建MQTT客户端失败", e);
 package cn.iocoder.yudao.module.iot.emq.service;
-import lombok.extern.slf4j.Slf4j;
 import org.eclipse.paho.client.mqttv3.MqttClient;
 import org.eclipse.paho.client.mqttv3.MqttMessage;
-import org.springframework.stereotype.Service;
 /**
  * 用于处理MQTT消息的具体业务逻辑，如订阅回调
  *
  * @author ahh
  */
-@Slf4j
-@Service
-public class EmqxService {
+public interface EmqxService {
-    public void subscribeCallback(String topic, MqttMessage mqttMessage) {
-        log.info("收到消息，主题: {}, 内容: {}", topic, new String(mqttMessage.getPayload()));
-        // 根据不同的主题，处理不同的业务逻辑
-        if (topic.contains("/property/post")) {
-            // 设备上报数据
-        }
-    }
+    /**
+     * 订阅回调
+     *
+     * @param topic       主题
+     * @param mqttMessage 消息
+     */
+    void subscribeCallback(String topic, MqttMessage mqttMessage);
-    public void subscribe(MqttClient client) {
-        try {
-            // 订阅默认主题，可以根据需要修改
-            client.subscribe("$share/yudao/+/+/#", 1);
-            log.info("订阅默认主题成功");
-        } catch (Exception e) {
-            log.error("订阅默认主题失败", e);
-        }
-    }
+    /**
+     * 订阅主题
+     *
+     * @param client MQTT 客户端
+     */
+    void subscribe(MqttClient client);
 }
+package cn.iocoder.yudao.module.iot.emq.service;
+
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.springframework.stereotype.Service;
+
+/**
+ * 用于处理MQTT消息的具体业务逻辑，如订阅回调
+ *
+ * @author ahh
+ */
+@Slf4j
+@Service
+public class EmqxServiceImpl implements EmqxService {
+
+    // TODO 多线程处理消息
+    @Override
+    public void subscribeCallback(String topic, MqttMessage mqttMessage) {
+        log.info("收到消息，主题: {}, 内容: {}", topic, new String(mqttMessage.getPayload()));
+        // 根据不同的主题，处理不同的业务逻辑
+        if (topic.contains("/property/post")) {
+            // 设备上报数据
+        }
+    }
+
+    @Override
+    public void subscribe(MqttClient client) {
+        try {
+            // 订阅默认主题，可以根据需要修改
+            client.subscribe("$share/yudao/+/+/#", 1);
+            log.info("订阅默认主题成功");
+        } catch (Exception e) {
+            log.error("订阅默认主题失败", e);
+        }
+    }
+}