测试进行数据推送

mhfire-controller/src/main/resources/application-dev.properties

@@ -72,7 +72,7 @@ mqtt.completionTimeout=5000
 mqtt.keep-alive-interval=60
 mqtt.username=wjzn2021
 mqtt.password=wjzn2021
-mqtt.sub-topics=usky
+mqtt.sub-topics=mh/water/info,mh/water/alert,mh/water/statistics
 mqtt.url=tcp://124.71.175.91:1883
 
 

mhfire-service/src/main/java/com/bizmatics/mhfire/service/api/OneCardApi.java

@@ -16,5 +16,4 @@ public class OneCardApi {
      */
     private static final String LOGIN_URL = String.format("%s%s",URL,"/mhxfzd/mhapi/bsUser/user/login");
 
-
 }

mhfire-service/src/main/java/com/bizmatics/mhfire/service/config/mqtt/MqttInConfig.java

@@ -36,7 +36,7 @@ public class MqttInConfig {
         //创建消息适配器 TODO 这里一定要注意，多端部署时id不能重复
         String clientId = "h-backend-mqtt-in-" + System.currentTimeMillis();
         MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(clientId,
-                mqttBaseConfig.mqttClientFactory(), mqttBaseConfig.getMsgTopic());
+                mqttBaseConfig.mqttClientFactory(), mqttBaseConfig.getMsgTopic().split(","));
         adapter.setCompletionTimeout(mqttBaseConfig.getCompletionTimeout());
         adapter.setConverter(new DefaultPahoMessageConverter());
         adapter.setQos(1);

mhfire-service/src/main/java/com/bizmatics/mhfire/service/listener/MqttListener.java

@@ -0,0 +1,34 @@
+package com.bizmatics.mhfire.service.listener;
+
+import com.bizmatics.mhfire.service.config.mqtt.MqttInConfig;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.annotation.Bean;
+import org.springframework.integration.annotation.ServiceActivator;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.stereotype.Component;
+
+/**
+ * @author yq
+ * @date 2021/11/3 8:13
+ */
+@Slf4j
+@Component
+public class MqttListener {
+
+    public static final String MESSAGE_NAME = "messageInput";
+
+    /**
+     * 处理消息-消费者
+     * @return
+     */
+    @Bean(MESSAGE_NAME)
+    @ServiceActivator(inputChannel = MqttInConfig.CHANNEL_NAME_INPUT)
+    public MessageHandler handler() {
+        return message -> {
+            log.info("mqtt-----接受到的消息"+message);
+            String payload = message.getPayload().toString();
+            //进行接口推送
+
+        };
+    }
+}