添加sn号更改接口

device-api-service/src/main/java/com/xy/consumer/TestUpConsumer.java

@@ -1,29 +0,0 @@
-package com.xy.consumer;
-
-import cn.hutool.json.JSONObject;
-import cn.hutool.json.JSONUtil;
-import com.xy.producer.MqttProducer;
-import lombok.AllArgsConstructor;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.stereotype.Service;
-
-/**
- * 测试消费者
- */
-@Slf4j
-@Service
-@AllArgsConstructor
-public class TestUpConsumer implements MqttConsumer {
-
-    private MqttProducer mqttProducer;
-
-    @Override
-    public boolean message(String topic, String payload) {
-        log.info("测试数据：{}", payload);
-        JSONObject jsonObject = JSONUtil.parseObj(payload);
-        String deviceId = jsonObject.getStr("deviceId");
-        jsonObject.set("time", System.currentTimeMillis());
-        mqttProducer.sendToMqtt(jsonObject.toString(), String.format("%s-test-down", deviceId));
-        return true;
-    }
-}

device-api-service/src/main/java/com/xy/consumer/ConnectedConsumer.java → device-api-service/src/main/java/com/xy/consumer/connected/ConnectedConsumer.java

@@ -1,7 +1,8 @@
-package com.xy.consumer;
+package com.xy.consumer.connected;
 
 import cn.hutool.json.JSONObject;
 import cn.hutool.json.JSONUtil;
+import com.xy.consumer.MqttConsumer;
 import com.xy.dto.DeviceStatusDto;
 import com.xy.service.DeviceStatusServiceImpl;
 import lombok.AllArgsConstructor;

device-api-service/src/main/java/com/xy/consumer/connected/ConnectedMqttConfiguration.java

@@ -0,0 +1,55 @@
+package com.xy.consumer.connected;
+
+import com.xy.configuration.MqttConfigUtils;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.annotation.IntegrationComponentScan;
+import org.springframework.integration.annotation.ServiceActivator;
+import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.core.MessageProducer;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHandler;
+
+@Configuration
+@IntegrationComponentScan
+public class ConnectedMqttConfiguration {
+
+    /**
+     * topic
+     */
+    public final static String TOPIC = "connectedNotify";
+
+    /**
+     * 入站通道名（消费者）订阅的bean名称
+     */
+    public static final String CHANNEL_NAME_IN = TOPIC + "MqttInboundChannel";
+
+
+    /*******************************消费者*******************************************/
+
+    /**
+     * MQTT信息通道（消费者）
+     */
+    @Bean(name = CHANNEL_NAME_IN)
+    public MessageChannel mqttInboundChannel() {
+        return new DirectChannel();
+    }
+
+    /**
+     * MQTT消息订阅绑定（消费者）
+     */
+    @Bean(name = TOPIC + "Inbound")
+    public MessageProducer inbound() {
+        return MqttConfigUtils.inbound(TOPIC, mqttInboundChannel());
+    }
+
+    /**
+     * MQTT消息处理器（消费者）
+     */
+    @Bean(name = TOPIC + "Handler")
+    @ServiceActivator(inputChannel = CHANNEL_NAME_IN)
+    public MessageHandler handler() {
+        return MqttConfigUtils.handler();
+    }
+
+}

device-api-service/src/main/java/com/xy/consumer/DisconnectedConsumer.java → device-api-service/src/main/java/com/xy/consumer/disconnect/DisconnectedConsumer.java

@@ -1,5 +1,7 @@
-package com.xy.consumer;
+package com.xy.consumer.disconnect;
 
+import com.xy.consumer.MqttConsumer;
+import com.xy.consumer.connected.ConnectedConsumer;
 import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;

device-api-service/src/main/java/com/xy/consumer/disconnect/DisconnectedMqttConfiguration.java

@@ -0,0 +1,55 @@
+package com.xy.consumer.disconnect;
+
+import com.xy.configuration.MqttConfigUtils;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.annotation.IntegrationComponentScan;
+import org.springframework.integration.annotation.ServiceActivator;
+import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.core.MessageProducer;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHandler;
+
+@Configuration
+@IntegrationComponentScan
+public class DisconnectedMqttConfiguration {
+
+    /**
+     * topic
+     */
+    public final static String TOPIC = "disConnectedNotify";
+
+    /**
+     * 入站通道名（消费者）订阅的bean名称
+     */
+    public static final String CHANNEL_NAME_IN = TOPIC + "MqttInboundChannel";
+
+
+    /*******************************消费者*******************************************/
+
+    /**
+     * MQTT信息通道（消费者）
+     */
+    @Bean(name = CHANNEL_NAME_IN)
+    public MessageChannel mqttInboundChannel() {
+        return new DirectChannel();
+    }
+
+    /**
+     * MQTT消息订阅绑定（消费者）
+     */
+    @Bean(name = TOPIC + "Inbound")
+    public MessageProducer inbound() {
+        return MqttConfigUtils.inbound(TOPIC, mqttInboundChannel());
+    }
+
+    /**
+     * MQTT消息处理器（消费者）
+     */
+    @Bean(name = TOPIC + "Handler")
+    @ServiceActivator(inputChannel = CHANNEL_NAME_IN)
+    public MessageHandler handler() {
+        return MqttConfigUtils.handler();
+    }
+
+}

device-api-service/src/main/java/com/xy/service/MqttServiceImpl.java

@@ -16,7 +16,7 @@ import com.xy.dto.MqttDto;
 import com.xy.dto.PaterDto;
 import com.xy.entity.MqttCommand;
 import com.xy.mapper.MqttCommandMapper;
-import com.xy.producer.MqttProducer;
+import com.xy.producer.CommProducer;
 import com.xy.utils.*;
 import com.xy.utils.consts.CommConsts;
 import com.xy.utils.enums.CmdTypeEnum;
@@ -44,7 +44,7 @@ import static com.xy.utils.PlusBeans.*;
 @RequiredArgsConstructor
 public class MqttServiceImpl extends ServiceImpl<MqttCommandMapper, MqttCommand> implements MqttService {
 
-    private final MqttProducer mqttProducer;
+    private final CommProducer mqttProducer;
 
     private final RedisService<String> redisService;
 

device-start/src/main/resources/bootstrap-uat.yml

@@ -1,10 +1,8 @@
 spring:
   mqtt:
     topics:
-      - {topic: "device-log", handler: com.xy.consumer.LogConsumer}
-      - {topic: "device-disconnected", handler: com.xy.consumer.DisconnectedConsumer}
-      - {topic: "device-connected", handler: com.xy.consumer.ConnectedConsumer}
-      - {topic: "test-up", handler: com.xy.consumer.TestUpConsumer}
+      - {topic: "disConnectedNotify", handler: com.xy.consumer.disconnect.DisconnectedConsumer}
+      - {topic: "connectedNotify", handler: com.xy.consumer.connected.ConnectedConsumer}
 
 #微服务相关配置
 cloud:

device-start/src/main/resources/bootstrap.yml

@@ -1,10 +1,6 @@
 spring:
   profiles:
     include: cloud
-  mqtt:
-    is-retry: true
-    topics:
-      - {topic: "dev-device-log", handler: com.xy.consumer.LogConsumer}
 
 server:
   port: 9051