mqtt重构

sys-sdk/pom.xml

@@ -30,7 +30,7 @@
         </dependency>
         <dependency>
             <groupId>com.xy</groupId>
-            <artifactId>xy-mqtt</artifactId>
+            <artifactId>mqtt-api</artifactId>
             <version>1.0</version>
         </dependency>
         <dependency>

sys-sdk/src/main/java/com/xy/config/SpringConfigs.java

@@ -0,0 +1,27 @@
+package com.xy.config;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.stereotype.Component;
+
+@Data
+@Component
+@ConfigurationProperties(prefix = "spring")
+public class SpringConfigs {
+
+    private Profiles profiles;
+
+    private Application application;
+
+    @Data
+    public static class Profiles {
+
+        private String active;
+    }
+
+    @Data
+    public static class Application {
+
+        private String name;
+    }
+}

sys-sdk/src/main/java/com/xy/producer/LogsProducer.java

@@ -1,26 +0,0 @@
-package com.xy.producer;
-
-import org.springframework.integration.annotation.MessagingGateway;
-import org.springframework.integration.mqtt.support.MqttHeaders;
-import org.springframework.messaging.handler.annotation.Header;
-import org.springframework.stereotype.Component;
-
-/**
- * 生产者接口
- */
-@Component
-@MessagingGateway(defaultRequestChannel = LogsProducerConfiguration.CHANNEL_NAME_OUT)
-public interface LogsProducer {
-
-    /**
-     * ata是发送消息的内容
-     * topic是消息发送的主题,就是配置文件的主题
-     * qos是mqtt 对消息处理机制分为0,1,2 其中0表示的是订阅者没收到消息不会再次发送,消息会丢失,1表示的是会尝试重试,一直到接收到消息,但这种情况可能导致订阅者收到多次重复消息,2相比多了一次去重的动作,确保订阅者收到的消息有一次
-     */
-
-    void sendToMqtt(String data);
-
-    void sendToMqtt(String data, @Header(MqttHeaders.TOPIC) String topic);
-
-    void sendToMqtt(String data, @Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos);
-}

sys-sdk/src/main/java/com/xy/producer/LogsProducerConfiguration.java

@@ -1,46 +0,0 @@
-package com.xy.producer;
-
-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.messaging.MessageChannel;
-import org.springframework.messaging.MessageHandler;
-
-@Configuration
-@IntegrationComponentScan
-public class LogsProducerConfiguration {
-
-    /**
-     * topic
-     */
-    public final static String TOPIC = "log";
-
-    /**
-     * 出站通道名（生产者）发布的bean名称
-     */
-    public static final String CHANNEL_NAME_OUT = TOPIC + "MqttOutboundChannel";
-
-
-    /*******************************生产者*******************************************/
-
-    /**
-     * MQTT信息通道（生产者）
-     */
-    @Bean(name = CHANNEL_NAME_OUT)
-    public MessageChannel mqttOutboundChannel() {
-        return new DirectChannel();
-    }
-
-    /**
-     * MQTT消息处理器（生产者）
-     */
-    @Bean(name = TOPIC + "MqttOutbound")
-    @ServiceActivator(inputChannel = CHANNEL_NAME_OUT)
-    public MessageHandler mqttOutbound() {
-        return MqttConfigUtils.mqttOutbound(TOPIC);
-    }
-
-}

sys-sdk/src/main/java/com/xy/utils/LogUtils.java

@@ -2,10 +2,10 @@ package com.xy.utils;
 
 import cn.hutool.json.JSONObject;
 import com.xy.config.SpringConfigs;
+import com.xy.dto.MqttDto;
 import com.xy.entity.SysCodeConfigureRedis;
 import com.xy.enums.LogEnum;
-import com.xy.producer.LogsProducer;
-import com.xy.producer.LogsProducerConfiguration;
+import com.xy.service.SysMqttSendService;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 import org.springframework.web.context.request.RequestContextHolder;
@@ -19,7 +19,7 @@ public class LogUtils {
 
     private static SpringConfigs springConfigs;
 
-    private static LogsProducer logsProducer;
+    private static SysMqttSendService sysMqttSendService;
 
     @Autowired(required = false)
     public void setSpringConfigs(SpringConfigs springConfigs) {
@@ -27,8 +27,8 @@ public class LogUtils {
     }
 
     @Autowired(required = false)
-    public void setMqttProducer(LogsProducer logsProducer) {
-        LogUtils.logsProducer = logsProducer;
+    public void setSysMqttSendService(SysMqttSendService sysMqttSendService) {
+        LogUtils.sysMqttSendService = sysMqttSendService;
     }
 
     /**
@@ -64,7 +64,7 @@ public class LogUtils {
                 .set("clientIp", getIp())
                 .set("serviceName", springConfigs.getApplication().getName())
                 .set("type", "operate");
-        logsProducer.sendToMqtt(jsonObject.toString(), LogsProducerConfiguration.TOPIC, 0);
+        sysMqttSendService.log(new MqttDto.RequestParams().setQos(0).setData(jsonObject.toString()));
     }
 
     /**
@@ -86,7 +86,7 @@ public class LogUtils {
                 .set("serviceName", springConfigs.getApplication().getName())
                 .set("createTime", DataTime.getSring())
                 .set("type", "events");
-        logsProducer.sendToMqtt(jsonObject.toString(), LogsProducerConfiguration.TOPIC, 0);
+        sysMqttSendService.log(new MqttDto.RequestParams().setQos(0).setData(jsonObject.toString()));
     }
 
     /**
@@ -106,7 +106,7 @@ public class LogUtils {
                 .set("serviceName", springConfigs.getApplication().getName())
                 .set("createTime", DataTime.getSring())
                 .set("type", "tasks");
-        logsProducer.sendToMqtt(jsonObject.toString(), LogsProducerConfiguration.TOPIC, 0);
+        sysMqttSendService.log(new MqttDto.RequestParams().setQos(0).setData(jsonObject.toString()));
     }
 
     /**