changg

.gitignore

@@ -1,3 +1,4 @@
 /.idea/
 /JX_Cover.iml
 /target
+/log/

README.md

@@ -0,0 +1,51 @@
+### MQTT配置说明
+
+```yml
+mqtt:
+  config: 
+    channel1:                                          #通道名称，可自定义，订阅消息时需要该名称
+      consumer-enable: true                            #是否开启consumer，默认开启
+      producer-enable: true                            #是否开启producer，默认开启    
+      url: [tcp://host1:1883, tcp://host1:1883]        #mqtt的url
+      topics: [topic1, topic2]                         #监听的主题，和qos一一对应
+      qos: [1, 0]                                      #监听主题的qos，和主题一一对应
+      username: admin                                  #用户名
+      password: public                                 #密码
+      timeout: 60                                      #连接超时时间，单位：秒
+      kep-alive-interval: 60                           #心跳时间，单位：秒
+      async: true                                      #发送消息时是否异步发送
+      client-id-append-ip: true                        #是否在clientId后面追加本机ip，因为clientid是唯一值，集群环境下不能使用相同的clientid，追加ip可解决该问题
+      consumer-client-id: consumer_client_test1        #consumer client id配置
+      producer-client-id: producer_client_test1        #producer client id配置
+      consumer-will:                                   #consumer遗嘱消息配置
+        qos: 1                                         #遗嘱qos
+        topic: will_topic                              #遗嘱主题
+        payload: '{"id": "consumer_client_test1"}'     #遗嘱内容
+        retained: false                                #是否发送保留消息
+      producer-will:                                   #producer遗嘱消息配置
+        qos: 1                                         #遗嘱qos
+        topic: will_topic                              #遗嘱主题
+        payload: '{"id": "producer_client_test1"}'     #遗嘱内容
+        retained: false                                #是否发送保留消息
+    channel2:                                          #通道名称，第二个配置
+      url: [tcp://host1:1883, tcp://host1:1883]
+      topics: [topic1, topic2]
+      qos: [1, 0]
+      username: admin
+      password: public
+      timeout: 60
+      kep-alive-interval: 60
+      async: true
+      consumer-client-id: consumer_client_test2
+      producer-client-id: producer_client_test2
+      consumer-will: 
+        qos: 1
+        topic: will_topic
+        payload: '{"id": "consumer_client_test2"}'
+        retained: false
+      producer-will: 
+        qos: 1
+        topic: will_topic
+        payload: '{"id": "producer_client_test2"}'
+        retained: false
+```

pom.xml

@@ -69,6 +69,12 @@
             <artifactId>shiro-spring</artifactId>
             <version>1.5.2</version>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-configuration-processor</artifactId>
+            <optional>true</optional>
+        </dependency>
+
         <dependency>
             <groupId>org.apache.commons</groupId>
             <artifactId>commons-lang3</artifactId>

src/main/java/com/usky/aspect/AutoLogAspect.java

@@ -35,15 +35,11 @@ import java.sql.Timestamp;
 @Aspect
 @Component
 public class AutoLogAspect {
-
     @Resource
     private LogService logService;
-
     @Pointcut("@annotation(com.usky.annotion.AutoLog)")
     public void logPointCut() {
-
     }
-
     @Around("logPointCut()")
     public Object around(ProceedingJoinPoint point) throws Throwable {
         long beginTime = System.currentTimeMillis();
@@ -61,7 +57,6 @@ public class AutoLogAspect {
     private void saveSysLog(ProceedingJoinPoint joinPoint, long time, Object obj) {
         MethodSignature signature = (MethodSignature) joinPoint.getSignature();
         Method method = signature.getMethod();
-
         SysLogDTO dto = new SysLogDTO();
         AutoLog syslog = method.getAnnotation(AutoLog.class);
         if(syslog != null){

src/main/java/com/usky/config/Swagger2Configuration.java

@@ -1,6 +1,7 @@
 package com.usky.config;
 
 import com.github.xiaoymin.knife4j.spring.annotations.EnableKnife4j;
+import lombok.Data;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -33,7 +34,7 @@ public class Swagger2Configuration {
 
     private ApiInfo buildApiInf() {
         return new ApiInfoBuilder()
-                .title("山东党校 v1.0")
+                .title("江西井盖 v1.0")
                 .version("1.0.0")
                 .description("springboot swagger2")
                 //   .termsOfServiceUrl("网址链接")

src/main/java/com/usky/config/mqtt/MqttCofigBean.java

@@ -1,94 +0,0 @@
-package com.usky.config.mqtt;
-
-import org.springframework.beans.factory.annotation.Value;
-import org.springframework.context.annotation.PropertySource;
-import org.springframework.stereotype.Component;
-
-@Component
-@PropertySource(value = {"classpath:mqtt-spring.properties"})
-public class MqttCofigBean {
-
-	@Value("${mqtt.username}")
-	private String username;
-
-	@Value("${mqtt.password}")
-	private String password;
-
-	@Value("${mqtt.url}")
-	private String hostUrl;
-
-	@Value("${mqtt.client-id-prefix}")
-	private String clientId;
-
-	@Value("${mqtt.sub-topics}")
-	private String msgTopic;
-
-	@Value("${mqtt.keep-alive-interval}")
-	//心跳间隔
-	private int keepAliveInterval;
-	@Value("${mqtt.completionTimeout}")
-	//心跳间隔
-	private int completionTimeout;
-
-	/**
-	 * 获取用户名
-	 *
-	 * @return
-	 */
-	public String getUsername() {
-		return this.username;
-	}
-
-	/**
-	 * 连接超时
-	 * @return
-	 */
-	public int getCompletionTimeout() {
-		return completionTimeout;
-	}
-
-	/**
-	 * 获取密码
-	 *
-	 * @return
-	 */
-	public String getPassword() {
-		return this.password;
-	}
-
-	/**
-	 * 获取服务器连接地址
-	 *
-	 * @return
-	 */
-	public String getHostUrl() {
-		return this.hostUrl;
-	}
-
-	/**
-	 * 获取客户端ID
-	 *
-	 * @return
-	 */
-	public String getClientId() {
-		return this.clientId;
-	}
-
-	/**
-	 * 获取默认主题
-	 *
-	 * @return
-	 */
-	public String[] getMsgTopic() {
-		String[] topic = msgTopic.split(",");
-		return topic;
-	}
-
-	/***
-	 * 获取连接超时时间
-	 * @return
-	 */
-	public int getKeepAliveInterval() {
-		return keepAliveInterval;
-	}
-}

src/main/java/com/usky/config/mqtt/SpringMqttConfig.java

@@ -1,111 +0,0 @@
-package com.usky.config.mqtt;
-
-import com.usky.utils.UUIDUtils;
-import org.aspectj.weaver.ast.Var;
-import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.integration.annotation.MessagingGateway;
-import org.springframework.integration.annotation.ServiceActivator;
-import org.springframework.integration.channel.DirectChannel;
-import org.springframework.integration.core.MessageProducer;
-import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
-import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
-import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
-import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
-import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
-import org.springframework.integration.mqtt.support.MqttHeaders;
-import org.springframework.messaging.MessageChannel;
-import org.springframework.messaging.MessageHandler;
-import org.springframework.messaging.handler.annotation.Header;
-
-import java.lang.reflect.Array;
-import java.util.Arrays;
-import java.util.List;
-
-@Configuration
-public class SpringMqttConfig {
-
-    @Autowired
-    public MqttCofigBean mqttCofigBean;
-
-    /*****
-     * 创建MqttPahoClientFactory，设置MQTT Broker连接属性，如果使用SSL验证，也在这里设置。
-     * @return
-     */
-    //  private List topicList = Arrays.asList(mqttCofigBean.getMsgTopic());
-    @Bean
-    public MqttPahoClientFactory mqttClientFactory() {
-        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
-        MqttConnectOptions options = new MqttConnectOptions();
-        options.setServerURIs(new String[]{mqttCofigBean.getHostUrl()});
-        options.setUserName(mqttCofigBean.getUsername());
-        options.setPassword(mqttCofigBean.getPassword().toCharArray());
-        factory.setConnectionOptions(options);
-        return factory;
-    }
-
-    @Bean
-    public MessageChannel mqttInputChannel() {
-        return new DirectChannel();
-    }
-
-    @Bean
-    public MessageProducer inbound() {
-        //创建消息适配器 TODO 这里一定要注意，多端部署时id不能重复
-        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(UUIDUtils.getUUID(),
-                mqttClientFactory(), mqttCofigBean.getMsgTopic());
-        adapter.setCompletionTimeout(mqttCofigBean.getCompletionTimeout());
-        adapter.setConverter(new DefaultPahoMessageConverter());
-        adapter.setQos(1);
-        adapter.setOutputChannel(mqttInputChannel());
-        return adapter;
-    }
-
-//    @Bean
-//    //ServiceActivator注解表明当前方法用于处理MQTT消息，inputChannel参数指定了用于消费消息的channel。
-//    @ServiceActivator(inputChannel = "mqttInputChannel")
-//    public MessageHandler handler() {
-//        return message -> {
-//            String payload = message.getPayload().toString();
-//            String topic = message.getHeaders().get("mqtt_receivedTopic").toString();
-//            // 根据topic分别进行消息处理。
-//            System.out.println(topic + ": 处理消息 " + payload);
-//
-////            if (Arrays.asList(mqttCofigBean.getMsgTopic()).contains(topic)) {
-////                System.out.println(topic + ": 处理消息 " + payload);
-////            } else {
-////                System.out.println(topic + ": 丢弃消息 " + payload);
-////            }
-//        };
-//    }
-    // 发送消息
-    @Bean
-    public MessageChannel mqttOutboundChannel() {
-        return new DirectChannel();
-    }
-    /*****
-     * 发送消息和消费消息Channel可以使用相同MqttPahoClientFactory
-     * @return
-     */
-    @Bean
-    @ServiceActivator(inputChannel = "mqttOutboundChannel")
-    public MessageHandler outbound() {
-        // 在这里进行mqttOutboundChannel的相关设置
-        MqttPahoMessageHandler messageHandler =
-                new MqttPahoMessageHandler("publishClient", mqttClientFactory());
-        //如果设置成true，发送消息时将不会阻塞。
-        messageHandler.setAsync(true);
-        messageHandler.setDefaultTopic("testTopic");
-        return messageHandler;
-    }
-    @MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
-    public interface MqttGateway {
-        // 定义重载方法，用于消息发送
-        void sendToMqtt(String payload);
-        // 指定topic进行消息发送
-        void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, String payload);
-        void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos, String payload);
-    }
-}

src/main/java/com/usky/config/mqtt/config/MqttAutoConfiguration.java

@@ -0,0 +1,160 @@
+package com.usky.config.mqtt.config;
+import com.usky.config.mqtt.config.MqttProperties.Config;
+import com.usky.config.mqtt.config.MqttProperties.Will;
+import com.usky.utils.mqtt.MqttUtils;
+import lombok.extern.log4j.Log4j2;
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.config.BeanDefinition;
+import org.springframework.beans.factory.config.BeanPostProcessor;
+import org.springframework.beans.factory.support.AbstractBeanDefinition;
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.support.DefaultListableBeanFactory;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
+import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
+import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
+import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
+import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
+import org.springframework.messaging.MessageChannel;
+
+import java.io.UnsupportedEncodingException;
+import java.util.Map;
+
+/**
+ * mqtt配置
+ * @author laowo
+ */
+@Log4j2
+@Configuration
+@EnableConfigurationProperties(MqttProperties.class)
+public class MqttAutoConfiguration implements ApplicationContextAware, BeanPostProcessor {
+
+    private ConfigurableApplicationContext applicationContext;
+    @Autowired
+    private MqttProperties mqttProperties;
+
+    @Override
+    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
+        this.applicationContext = (ConfigurableApplicationContext) applicationContext;
+        mqttProperties.getConfig().forEach((chnnelName, config) -> init(chnnelName, config));
+
+    }
+
+    /**
+     * 初始化
+     *
+     * @param channelName
+     * @param config
+     */
+    private void init(String channelName, Config config) {
+        DefaultListableBeanFactory beanFactory = (DefaultListableBeanFactory) applicationContext.getBeanFactory();
+        // 默认开启consumer
+        if (!Boolean.FALSE.equals(config.getConsumerEnable())) {
+            // 通道信息
+            beanFactory.registerBeanDefinition(channelName, mqttChannel());
+            log.info("初始化mqtt, channel {}, 配置 {} ", channelName, config);
+            MessageChannel mqttChannel = beanFactory.getBean(channelName, MessageChannel.class);
+            beanFactory.registerBeanDefinition(channelName + "MqttChannelAdapter", channelAdapter(config, mqttChannel));
+            log.info("初始化mqtt Channel Adapter");
+        }
+
+        // 默认开启consumer
+        if (!Boolean.FALSE.equals(config.getProducerEnable())) {
+            String handlerBeanName = channelName + MqttUtils.CHANNEL_NAME_SUFFIX;
+            beanFactory.registerBeanDefinition(handlerBeanName, mqttOutbound(config));
+            log.info("初始化mqtt MqttPahoMessageHandler");
+
+            MqttUtils.put(channelName, beanFactory.getBean(handlerBeanName, MqttPahoMessageHandler.class));
+        }
+    }
+
+    /**
+     * mqtt工厂
+     *
+     * @param config
+     * @param isConsumer
+     * @return
+     */
+    private MqttPahoClientFactory mqttClientFactory(Config config, boolean isConsumer) {
+        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
+        MqttConnectOptions options = new MqttConnectOptions();
+        options.setServerURIs(config.getUrl());
+        options.setCleanSession(true);
+        options.setKeepAliveInterval(config.getKepAliveInterval());
+        options.setPassword(config.getPassword().toCharArray());
+        options.setUserName(config.getUsername());
+        options.setConnectionTimeout(config.getTimeout());
+
+        Will will = null;
+        if (isConsumer && config.getConsumerWill() != null) {
+            will = config.getConsumerWill();
+        } else if (!isConsumer && config.getProducerWill() != null) {
+            will = config.getProducerWill();
+        }
+        if (will != null) {
+            try {
+                options.setWill(will.getTopic(), will.getPayload().getBytes("utf-8"), will.getQos(),
+                        will.getRetained());
+            } catch (UnsupportedEncodingException e) {
+                log.error(e.getMessage(), e);
+            }
+        }
+
+        factory.setConnectionOptions(options);
+        return factory;
+    }
+
+    /**
+     * 初始化通道
+     *
+     * @return
+     */
+    private AbstractBeanDefinition mqttChannel() {
+        BeanDefinitionBuilder messageChannelBuilder = BeanDefinitionBuilder.genericBeanDefinition(DirectChannel.class);
+        messageChannelBuilder.setScope(BeanDefinition.SCOPE_SINGLETON);
+        return messageChannelBuilder.getBeanDefinition();
+    }
+
+    /**
+     * mqtt消息驱动转换器
+     *
+     * @param config
+     * @param mqttChannel
+     * @return
+     */
+    private AbstractBeanDefinition channelAdapter(Config config, MessageChannel mqttChannel) {
+        BeanDefinitionBuilder messageProducerBuilder = BeanDefinitionBuilder
+                .genericBeanDefinition(MqttPahoMessageDrivenChannelAdapter.class);
+        messageProducerBuilder.setScope(BeanDefinition.SCOPE_SINGLETON);
+        messageProducerBuilder.addConstructorArgValue(config.getConsumerClientId());
+        messageProducerBuilder.addConstructorArgValue(mqttClientFactory(config, true));
+        messageProducerBuilder.addConstructorArgValue(config.getTopics());
+        messageProducerBuilder.addPropertyValue("converter", new DefaultPahoMessageConverter());
+        messageProducerBuilder.addPropertyValue("qos", config.getQos());
+        messageProducerBuilder.addPropertyValue("outputChannel", mqttChannel);
+
+        return messageProducerBuilder.getBeanDefinition();
+    }
+
+    /**
+     * 消息发送客户端
+     *
+     * @param config
+     * @return
+     */
+    private AbstractBeanDefinition mqttOutbound(Config config) {
+        BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(MqttPahoMessageHandler.class);
+        builder.addConstructorArgValue(config.getProducerClientId());
+        builder.addConstructorArgValue(mqttClientFactory(config, false));
+        builder.addPropertyValue("async", config.getAsync());
+
+        return builder.getBeanDefinition();
+    }
+}

src/main/java/com/usky/config/mqtt/config/MqttProperties.java

@@ -0,0 +1,140 @@
+package com.usky.config.mqtt.config;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.PropertySource;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.Map;
+
+/**
+ * mqtt配置
+ * @author laowo
+ */
+@Data
+@ConfigurationProperties(prefix = "mqtt")
+@Configuration
+public class MqttProperties {
+	/**
+	 * 本机ip作为clientid的后缀
+	 */
+	private static String hostAddress;
+
+	/**
+	 * 所有的配置
+	 */
+	private final Map<String, Config> config;
+
+	static {
+		try {
+			InetAddress address = InetAddress.getLocalHost();
+			hostAddress = "_ip_" + address.getHostAddress();
+		} catch (UnknownHostException e) {
+			e.printStackTrace();
+		}
+	}
+
+	@Data
+	public static class Config {
+		/**
+		 * 是否开启consumer,默认true开启
+		 */
+		private Boolean consumerEnable;
+		/**
+		 * 是否开启producer,默认true开启
+		 */
+		private Boolean producerEnable;
+		/**
+		 * 数组tcp://ip:port
+		 */
+		private String[] url;
+		/**
+		 * 超时时间，单位：秒
+		 */
+		private int timeout;
+		/**
+		 * 心跳时间，秒
+		 */
+		private int kepAliveInterval;
+		/**
+		 * qos设置，和topic一一对应
+		 */
+		private int[] qos;
+		/**
+		 * 主题，和qos一一对应
+		 */
+		private String[] topics;
+		/**
+		 * 账号
+		 */
+		private String username;
+		/**
+		 * 密码
+		 */
+		private String password;
+		/**
+		 * clientId后是否添加本机ip
+		 */
+		private Boolean clientIdAppendIp;
+		/**
+		 * consumer clientId
+		 */
+		private String consumerClientId;
+		/**
+		 * producer clientId
+		 */
+		private String producerClientId;
+		/**
+		 * 是否异步发送消息
+		 */
+		private Boolean async;
+		/**
+		 * consumer遗嘱配置
+		 */
+		private Will consumerWill;
+		/**
+		 * producer遗嘱配置
+		 */
+		private Will producerWill;
+
+		/**
+		 * 重写获取消费者客户端id
+		 * 
+		 * @return
+		 */
+		public String getConsumerClientId() {
+			return clientIdAppendIp == null || !clientIdAppendIp ? consumerClientId : consumerClientId + hostAddress;
+		}
+
+		/**
+		 * 重写获生产者客户端id
+		 * 
+		 * @return
+		 */
+		public String getProducerClientId() {
+			return clientIdAppendIp == null || !clientIdAppendIp ? producerClientId : producerClientId + hostAddress;
+		}
+	}
+
+	@Data
+	public static class Will {
+		/**
+		 * 遗嘱qos设置
+		 */
+		private int qos;
+		/**
+		 * 遗嘱主题
+		 */
+		private String topic;
+		/**
+		 * 遗嘱内容
+		 */
+		private String payload;
+		/**
+		 * 是否发送保留消息
+		 */
+		private Boolean retained;
+	}
+}

src/main/java/com/usky/webScoket/HttpAuthHandler.java → src/main/java/com/usky/config/webScoket/HttpAuthHandler.java

@@ -1,5 +1,6 @@
-package com.usky.webScoket;
+package com.usky.config.webScoket;
 
+import com.usky.utils.TimeUtil;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
 import org.springframework.web.socket.CloseStatus;
@@ -7,8 +8,6 @@ import org.springframework.web.socket.TextMessage;
 import org.springframework.web.socket.WebSocketSession;
 import org.springframework.web.socket.handler.TextWebSocketHandler;
 
-import java.time.LocalDateTime;
-
 @Component
 @Slf4j
 public class HttpAuthHandler extends TextWebSocketHandler {
@@ -22,14 +21,14 @@ public class HttpAuthHandler extends TextWebSocketHandler {
     public void afterConnectionEstablished(WebSocketSession session) throws Exception {
         String id = session.getId();
         System.out.println("id = " + id);
-      //  String query = session.getUri().getQuery();
-        if (null !=id) {
+        //  String query = session.getUri().getQuery();
+        if (null != id) {
             WsSessionManager.add(id, session);
         }
     }
-
     /**
      * 接收消息事件
+     *
      * @param session
      * @param message
      * @throws Exception
@@ -40,9 +39,8 @@ public class HttpAuthHandler extends TextWebSocketHandler {
         String payload = message.getPayload();
         Object token = session.getAttributes().get("token");
         System.out.println("server 接收到 " + token + " 发送的 " + payload);
-  //  WsSessionManager.sendMessageToAll(new TextMessage("server 发送给 " + token + " 消息 " + payload + " " + LocalDateTime.now().toString()));
+        //  WsSessionManager.sendMessageToAll(new TextMessage("server 发送给 " + token + " 消息 " + payload + " " + LocalDateTime.now().toString()));
     }
-
     /**
      * socket 断开连接时
      *
@@ -52,14 +50,13 @@ public class HttpAuthHandler extends TextWebSocketHandler {
      */
     @Override
     public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
-    //    String query = session.getUri().getQuery();
+        //    String query = session.getUri().getQuery();
         String id = session.getId();
-            // 用户退出，移除缓存
-            WsSessionManager.remove(id);
-            log.info("===================连接断开了===============");
+        // 用户退出，移除缓存
+        WsSessionManager.remove(id);
+        log.info("===================连接断开,当前时间{}===========", TimeUtil.getTime());
 
     }
 
 
-
 }

src/main/java/com/usky/webScoket/MyInterceptor.java → src/main/java/com/usky/config/webScoket/MyInterceptor.java

@@ -1,4 +1,4 @@
-package com.usky.webScoket;
+package com.usky.config.webScoket;
 
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.http.server.ServerHttpRequest;
@@ -9,9 +9,9 @@ import org.springframework.web.socket.server.HandshakeInterceptor;
 
 import java.util.Map;
 
+
 /**
- * @author buhao
- * @version MyInterceptor.java, v 0.1 2019-10-17 19:21 buhao
+ * @author laowo
  */
 @Component
 @Slf4j

src/main/java/com/usky/webScoket/WebSocketConfig.java → src/main/java/com/usky/config/webScoket/WebSocketConfig.java

@@ -1,4 +1,4 @@
-package com.usky.webScoket;
+package com.usky.config.webScoket;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Configuration;

src/main/java/com/usky/webScoket/WsSessionManager.java → src/main/java/com/usky/config/webScoket/WsSessionManager.java

@@ -1,4 +1,4 @@
-package com.usky.webScoket;
+package com.usky.config.webScoket;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.web.socket.TextMessage;
 import org.springframework.web.socket.WebSocketSession;

src/main/java/com/usky/constant/CommonConstant.java

@@ -5,80 +5,90 @@ import java.util.Map;
 
 public interface CommonConstant {
 
-	/**
-	 * 正常状态
-	 */
-	public static final Integer STATUS_NORMAL = 0;
-
-	/**
-	 * 禁用状态
-	 */
-	public static final Integer STATUS_DISABLE = -1;
-
-	/**
-	 * 删除标志
-	 */
-	public static final Integer DEL_FLAG_1 = 1;
-
-	/**
-	 * 未删除
-	 */
-	public static final Integer DEL_FLAG_0 = 0;
-
-	/**
-	 * 系统日志类型： 登录
-	 */
-	public static final int LOG_TYPE_1 = 1;
-	
-	/**
-	 * 系统日志类型： 操作
-	 */
-	public static final int LOG_TYPE_2 = 2;
-
-	/**
-	 * 操作日志类型： 查询
-	 */
-	public static final int OPERATE_TYPE_1 = 1;
-	
-	/**
-	 * 操作日志类型： 添加
-	 */
-	public static final int OPERATE_TYPE_2 = 2;
-	
-	/**
-	 * 操作日志类型： 更新
-	 */
-	public static final int OPERATE_TYPE_3 = 3;
-	
-	/**
-	 * 操作日志类型： 删除
-	 */
-	public static final int OPERATE_TYPE_4 = 4;
-	
-	/**
-	 * 操作日志类型： 导入
-	 */
-	public static final int OPERATE_TYPE_5 = 5;
-	
-	/**
-	 * 操作日志类型： 导出
-	 */
-	public static final int OPERATE_TYPE_6 = 6;
-	
-	
-	/** {@code 500 Server Error} (HTTP/1.0 - RFC 1945) */
+    /**
+     * 正常状态
+     */
+    public static final Integer STATUS_NORMAL = 0;
+
+    /**
+     * 禁用状态
+     */
+    public static final Integer STATUS_DISABLE = -1;
+
+    /**
+     * 删除标志
+     */
+    public static final Integer DEL_FLAG_1 = 1;
+
+    /**
+     * 未删除
+     */
+    public static final Integer DEL_FLAG_0 = 0;
+
+    /**
+     * 系统日志类型： 登录
+     */
+    public static final int LOG_TYPE_1 = 1;
+
+    /**
+     * 系统日志类型： 操作
+     */
+    public static final int LOG_TYPE_2 = 2;
+
+    /**
+     * 操作日志类型： 查询
+     */
+    public static final int OPERATE_TYPE_1 = 1;
+
+    /**
+     * 操作日志类型： 添加
+     */
+    public static final int OPERATE_TYPE_2 = 2;
+
+    /**
+     * 操作日志类型： 更新
+     */
+    public static final int OPERATE_TYPE_3 = 3;
+
+    /**
+     * 操作日志类型： 删除
+     */
+    public static final int OPERATE_TYPE_4 = 4;
+
+    /**
+     * 操作日志类型： 导入
+     */
+    public static final int OPERATE_TYPE_5 = 5;
+
+    /**
+     * 操作日志类型： 导出
+     */
+    public static final int OPERATE_TYPE_6 = 6;
+
+
+    /**
+     * {@code 500 Server Error} (HTTP/1.0 - RFC 1945)
+     */
     public static final Integer SC_INTERNAL_SERVER_ERROR_500 = 500;
-    /** {@code 200 OK} (HTTP/1.0 - RFC 1945) */
+    /**
+     * {@code 200 OK} (HTTP/1.0 - RFC 1945)
+     */
     public static final Integer SC_OK_200 = 200;
 
-	/**
-	 * 首页天气地址
-	 */
+    /**
+     * 首页天气地址
+     */
+
+    public static final String TOP_WEATHER_HOST = "https://weather01.market.alicloudapi.com";
+    public static final String TOP_WEATHER_PATH = "/area-to-weather";
+    public static final String TOP_WEATHER_METHOD = "GET";
+    public static final String TOP_WEATHER_APPCODE = "0f2b7fce6e104ba8835358b7b59b4fb6";
 
-	public static final String TOP_WEATHER_HOST  = "https://weather01.market.alicloudapi.com";
-	public static final String TOP_WEATHER_PATH  = "/area-to-weather";
-	public static final String TOP_WEATHER_METHOD  = "GET";
-	public static final String TOP_WEATHER_APPCODE  = "0f2b7fce6e104ba8835358b7b59b4fb6";
 
+    /**
+     * MQTT数据接收类型
+     */
+    public static final String MQTT_MESSAGE_TYPE_INFO = "INFO";
+    public static final String MQTT_MESSAGE_TYPE_ALARM = "ALARM";
 
 }

src/main/java/com/usky/controller/mqtt/MqttController.java

@@ -1,38 +0,0 @@
-package com.usky.controller.mqtt;
-
-import com.usky.config.mqtt.SpringMqttConfig;
-import com.usky.utils.RedisUtil;
-import io.swagger.annotations.Api;
-import io.swagger.models.auth.In;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.data.redis.core.RedisTemplate;
-import org.springframework.web.bind.annotation.GetMapping;
-import org.springframework.web.bind.annotation.PathVariable;
-import org.springframework.web.bind.annotation.PostMapping;
-import org.springframework.web.bind.annotation.RestController;
-
-/**
- * @author laowo
- * @version v1.0
- * @date 2021/6/21 11:40
- * @description TODO
- **/
-@RestController
-@Api(tags = "MQTT测试")
-@Slf4j
-public class MqttController {
-    @Autowired
-    private RedisUtil redisUtil;
-    @Autowired
-    private SpringMqttConfig.MqttGateway mqttGateway;
-    @Autowired
-    private RedisTemplate<String, String> redisTemplate;
-
-    @PostMapping("/send/{topic}/{message}")
-    public String send(@PathVariable String topic, @PathVariable String message) {
-        // 发送消息到指定topic
-        mqttGateway.sendToMqtt(topic, message);
-        return "send message : " + message;
-    }
-}

src/main/java/com/usky/controller/mqtt/MqttHandler.java

@@ -1,63 +0,0 @@
-package com.usky.controller.mqtt;
-
-import com.alibaba.fastjson.JSON;
-import com.alibaba.fastjson.JSONObject;
-import com.usky.utils.RedisUtil;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.context.annotation.Bean;
-import org.springframework.integration.annotation.ServiceActivator;
-import org.springframework.messaging.MessageHandler;
-import org.springframework.stereotype.Component;
-
-import java.io.UnsupportedEncodingException;
-import java.nio.charset.StandardCharsets;
-import java.util.List;
-import java.util.Map;
-
-/**
- * @author laowo
- * @version v1.0
- * @date 2021/6/24 11:01
- * @description TODO
- **/
-@Component
-@Slf4j
-public class MqttHandler {
-    @Autowired
-    private RedisUtil redisUtil;
-
-    @Bean
-    //ServiceActivator注解表明当前方法用于处理MQTT消息，inputChannel参数指定了用于消费消息的channel。
-    @ServiceActivator(inputChannel = "mqttInputChannel")
-    public MessageHandler handler() {
-        return message -> {
-            //获取实时数据
-            String payload = "";
-            String s = message.getPayload().toString();
-            byte[] bytes = s.getBytes();
-            try {
-                payload = new String(s.getBytes("gbk"),"utf-8");
-            } catch (UnsupportedEncodingException e) {
-                log.error("【MQTT】payload字符编码转换错误{}", s, e);
-                e.printStackTrace();
-            }
-
-            log.info("【MQTT】原始数据: {}", payload);
-            Map<String, Object> dataMap = JSON.parseObject(payload, Map.class);
-            if (dataMap.containsKey("sn")) {
-                String sn = (String) dataMap.get("sn");
-                List<JSONObject> devsList = (List) dataMap.get("devs");
-                String devName = null;
-                for (JSONObject jsonObject : devsList) {
-                    devName = (String) jsonObject.get("dev");
-                }
-                redisUtil.hmset(sn + "#" + devName, dataMap);
-            }
-            String topic = message.getHeaders().get("mqtt_receivedTopic").toString();
-            log.info(topic + ": 处理消息 " + payload);
-
-        };
-    }
-
-}

src/main/java/com/usky/controller/top/TopController.java

@@ -0,0 +1,116 @@
+package com.usky.controller.top;
+
+import com.alibaba.fastjson.JSONObject;
+import com.usky.annotion.AutoLog;
+import com.usky.constant.CommonConstant;
+import com.usky.entity.mqtt.TbDeviceInfoDTO;
+import com.usky.entity.mqtt.vo.TbDeviceAlarmsVO;
+import com.usky.entity.mqtt.vo.TbDeviceInfoVO;
+import com.usky.entity.mqtt.vo.TbDeviceVO;
+import com.usky.entity.mqtt.vo.TbDeviceVOTop;
+import com.usky.service.top.TopServcie;
+import com.usky.utils.HttpUtils;
+import com.usky.utils.RedisUtil;
+import com.usky.utils.Result;
+import io.swagger.annotations.Api;
+import io.swagger.annotations.ApiImplicitParam;
+import io.swagger.annotations.ApiImplicitParams;
+import io.swagger.annotations.ApiOperation;
+import org.apache.http.HttpEntity;
+import org.apache.http.HttpResponse;
+import org.apache.http.util.EntityUtils;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.*;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * @author laowo
+ * @version v1.0
+ * @date 2021/8/10 17:25
+ * @description TODO
+ **/
+@RestController
+@RequestMapping("jx/top")
+@Api(tags = "首页")
+public class TopController {
+    @Autowired
+    private TopServcie topServcie;
+    @Autowired
+    private RedisUtil redisUtil;
+
+    @GetMapping("listDeviceStatus")
+    @ApiOperation(value = "首页-设备状态统计")
+    @ApiImplicitParams({
+    })
+    public Result<?> listDeviceStatus() {
+        Map<String, Object> result = topServcie.listDeviceStatus();
+        return Result.OK(result);
+    }
+
+    @GetMapping("listAlarmsSe")
+    @ApiOperation(value = "首页-七日告警统计")
+    @ApiImplicitParams({
+    })
+    public Result<?> listAlarmsSe() {
+        List<Object> result = topServcie.listAlarmsSe();
+        return Result.OK(result);
+    }
+
+
+    @GetMapping("listDevcie")
+    @ApiOperation(value = "首页-首页设备查询")
+    @ApiImplicitParams({
+    })
+
+    public Result<?> listDevcie() {
+
+        List<TbDeviceVOTop> result = topServcie.listDevcie();
+        return Result.OK(result);
+    }
+
+
+    @ApiOperation(value = "首页天气查询")
+    @GetMapping("listWeather")
+    @AutoLog(value = "首页天气查询")
+    @ApiImplicitParams({
+            @ApiImplicitParam(name = "area", value = "地区", required = true, paramType = "query")
+    })
+    public Result<Object> listWeather(@RequestParam(name = "area", required = false) String area) {
+        JSONObject jsonObject = null;
+        if (redisUtil.hasKey(area)) {
+            return Result.OK(redisUtil.get(area));
+        }
+        Map<String, String> headers = new HashMap<String, String>();
+        headers.put("Authorization", "APPCODE " + CommonConstant.TOP_WEATHER_APPCODE);
+        Map<String, String> querys = new HashMap<String, String>();
+        querys.put("area", area);
+        try {
+            HttpResponse response = HttpUtils.doGet(CommonConstant.TOP_WEATHER_HOST, CommonConstant.TOP_WEATHER_PATH, CommonConstant.TOP_WEATHER_METHOD, headers, querys);
+            HttpEntity entity = response.getEntity();
+            if (null != entity) {
+                String s = EntityUtils.toString(response.getEntity());
+                jsonObject = JSONObject.parseObject(s);
+                redisUtil.set(area, jsonObject, 60 * 60 * 6);
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+        return Result.OK(jsonObject);
+    }
+
+
+    @ApiOperation(value = "首页-液位查询")
+    @GetMapping("listYw")
+    @ApiImplicitParams({
+            @ApiImplicitParam(name = "devId", value = "设备id", required = false, paramType = "query")
+    })
+    public Result<Object> listYw(@RequestParam(name = "devId",required = false, defaultValue = "10011") String devId) {
+        List<TbDeviceInfoDTO> tbDeviceInfoDTOS = topServcie.listYw(devId);
+        return Result.OK(tbDeviceInfoDTOS);
+    }
+
+
+}

src/main/java/com/usky/entity/build/TbBuildingEntity.java

@@ -11,7 +11,7 @@ import java.util.Objects;
  * @description TODO
  **/
 @Entity
-@Table(name = "tb_building", schema = "gx_Security", catalog = "")
+@Table(name = "tb_building", schema = "jx_cover", catalog = "")
 public class TbBuildingEntity {
     private int id;
     private String buildUuid;

src/main/java/com/usky/entity/build/TbFloorEntity.java

@@ -11,7 +11,7 @@ import java.util.Objects;
  * @description TODO
  **/
 @Entity
-@Table(name = "tb_floor", schema = "gx_Security", catalog = "")
+@Table(name = "tb_floor", schema = "jx_cover", catalog = "")
 public class TbFloorEntity {
     private int id;
     private String floorUuid;

src/main/java/com/usky/entity/build/TbRoomEntity.java

@@ -11,7 +11,7 @@ import java.util.Objects;
  * @description TODO
  **/
 @Entity
-@Table(name = "tb_room", schema = "gx_Security", catalog = "")
+@Table(name = "tb_room", schema = "jx_cover", catalog = "")
 public class TbRoomEntity {
     private int id;
     private String roomUuid;

src/main/java/com/usky/entity/mqtt/TbDeviceAlarmsDTO.java

@@ -0,0 +1,160 @@
+package com.usky.entity.mqtt;
+
+import javax.persistence.*;
+import java.util.Objects;
+
+/**
+ * @author laowo
+ * @version v1.0
+ * @date 2021/8/10 14:19
+ * @description TODO
+ **/
+@Entity
+@Table(name = "tb_device_alarms", schema = "jx_cover", catalog = "")
+public class TbDeviceAlarmsDTO {
+    private int id;
+    private String devId;
+    private String registerId;
+    private String deviceName;
+    private String deviceType;
+    private String devcieModel;
+    private String property;
+    private String serial;
+    private String value;
+    private Integer status;
+    private String createTime;
+    private String timeStamp;
+
+    @Id
+    @Column(name = "id", nullable = false)
+    public int getId() {
+        return id;
+    }
+
+    public void setId(int id) {
+        this.id = id;
+    }
+
+    @Basic
+    @Column(name = "dev_id", nullable = true, length = 50)
+    public String getDevId() {
+        return devId;
+    }
+
+    public void setDevId(String devId) {
+        this.devId = devId;
+    }
+
+    @Basic
+    @Column(name = "register_id", nullable = true, length = 50)
+    public String getRegisterId() {
+        return registerId;
+    }
+
+    public void setRegisterId(String registerId) {
+        this.registerId = registerId;
+    }
+
+    @Basic
+    @Column(name = "device_name", nullable = true, length = 50)
+    public String getDeviceName() {
+        return deviceName;
+    }
+
+    public void setDeviceName(String deviceName) {
+        this.deviceName = deviceName;
+    }
+
+    @Basic
+    @Column(name = "device_type", nullable = true, length = 50)
+    public String getDeviceType() {
+        return deviceType;
+    }
+
+    public void setDeviceType(String deviceType) {
+        this.deviceType = deviceType;
+    }
+
+    @Basic
+    @Column(name = "devcie_model", nullable = true, length = 50)
+    public String getDevcieModel() {
+        return devcieModel;
+    }
+
+    public void setDevcieModel(String devcieModel) {
+        this.devcieModel = devcieModel;
+    }
+
+    @Basic
+    @Column(name = "property", nullable = true, length = 50)
+    public String getProperty() {
+        return property;
+    }
+
+    public void setProperty(String property) {
+        this.property = property;
+    }
+
+    @Basic
+    @Column(name = "serial", nullable = true, length = 10)
+    public String getSerial() {
+        return serial;
+    }
+
+    public void setSerial(String serial) {
+        this.serial = serial;
+    }
+
+    @Basic
+    @Column(name = "value", nullable = true, length = 20)
+    public String getValue() {
+        return value;
+    }
+
+    public void setValue(String value) {
+        this.value = value;
+    }
+
+    @Basic
+    @Column(name = "status", nullable = true)
+    public Integer getStatus() {
+        return status;
+    }
+
+    public void setStatus(Integer status) {
+        this.status = status;
+    }
+
+    @Basic
+    @Column(name = "create_time", nullable = true, length = 50)
+    public String getCreateTime() {
+        return createTime;
+    }
+
+    public void setCreateTime(String createTime) {
+        this.createTime = createTime;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        TbDeviceAlarmsDTO that = (TbDeviceAlarmsDTO) o;
+        return id == that.id && Objects.equals(devId, that.devId) && Objects.equals(registerId, that.registerId) && Objects.equals(deviceName, that.deviceName) && Objects.equals(deviceType, that.deviceType) && Objects.equals(devcieModel, that.devcieModel) && Objects.equals(property, that.property) && Objects.equals(serial, that.serial) && Objects.equals(value, that.value) && Objects.equals(status, that.status) && Objects.equals(createTime, that.createTime);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(id, devId, registerId, deviceName, deviceType, devcieModel, property, serial, value, status, createTime);
+    }
+
+    @Basic
+    @Column(name = "time_stamp", nullable = true, length = 50)
+    public String getTimeStamp() {
+        return timeStamp;
+    }
+
+    public void setTimeStamp(String timeStamp) {
+        this.timeStamp = timeStamp;
+    }
+}

src/main/java/com/usky/entity/mqtt/TbDeviceDTO.java

@@ -0,0 +1,182 @@
+package com.usky.entity.mqtt;
+
+import javax.persistence.*;
+import java.util.Objects;
+
+/**
+ * @author laowo
+ * @version v1.0
+ * @date 2021/8/10 16:11
+ * @description TODO
+ **/
+@Entity
+@Table(name = "tb_device", schema = "jx_cover", catalog = "")
+public class TbDeviceDTO {
+    private int id;
+    private String devId;
+    private String registerId;
+    private String devceName;
+    private String devcieModel;
+    private String connType;
+    private String deviceType;
+    private String createTime;
+    private Integer del;
+    private String remark;
+    private String longitude;
+    private String dimension;
+    private String userId;
+    private String addr;
+
+    @Id
+    @Column(name = "id", nullable = false)
+    public int getId() {
+        return id;
+    }
+
+    public void setId(int id) {
+        this.id = id;
+    }
+
+    @Basic
+    @Column(name = "addr", nullable = false, length = 50)
+    public String getAddr() {
+        return addr;
+    }
+
+    public void setAddr(String addr) {
+        this.addr = addr;
+    }
+
+    @Basic
+    @Column(name = "dev_id", nullable = false, length = 10)
+    public String getDevId() {
+        return devId;
+    }
+
+    public void setDevId(String devId) {
+        this.devId = devId;
+    }
+
+    @Basic
+    @Column(name = "register_id", nullable = true, length = 50)
+    public String getRegisterId() {
+        return registerId;
+    }
+
+    public void setRegisterId(String registerId) {
+        this.registerId = registerId;
+    }
+
+    @Basic
+    @Column(name = "devce_name", nullable = true, length = 50)
+    public String getDevceName() {
+        return devceName;
+    }
+
+    public void setDevceName(String devceName) {
+        this.devceName = devceName;
+    }
+
+    @Basic
+    @Column(name = "devcie_model", nullable = true, length = 50)
+    public String getDevcieModel() {
+        return devcieModel;
+    }
+
+    public void setDevcieModel(String devcieModel) {
+        this.devcieModel = devcieModel;
+    }
+
+    @Basic
+    @Column(name = "conn_type", nullable = true, length = 10)
+    public String getConnType() {
+        return connType;
+    }
+
+    public void setConnType(String connType) {
+        this.connType = connType;
+    }
+
+    @Basic
+    @Column(name = "device_type", nullable = true, length = 50)
+    public String getDeviceType() {
+        return deviceType;
+    }
+
+    public void setDeviceType(String deviceType) {
+        this.deviceType = deviceType;
+    }
+
+    @Basic
+    @Column(name = "create_time", nullable = true, length = 50)
+    public String getCreateTime() {
+        return createTime;
+    }
+
+    public void setCreateTime(String createTime) {
+        this.createTime = createTime;
+    }
+
+    @Basic
+    @Column(name = "del", nullable = true)
+    public Integer getDel() {
+        return del;
+    }
+
+    public void setDel(Integer del) {
+        this.del = del;
+    }
+
+    @Basic
+    @Column(name = "remark", nullable = true, length = 100)
+    public String getRemark() {
+        return remark;
+    }
+
+    public void setRemark(String remark) {
+        this.remark = remark;
+    }
+
+    @Basic
+    @Column(name = "longitude", nullable = true, length = 50)
+    public String getLongitude() {
+        return longitude;
+    }
+
+    public void setLongitude(String longitude) {
+        this.longitude = longitude;
+    }
+
+    @Basic
+    @Column(name = "dimension", nullable = true, length = 50)
+    public String getDimension() {
+        return dimension;
+    }
+
+    public void setDimension(String dimension) {
+        this.dimension = dimension;
+    }
+
+    @Basic
+    @Column(name = "user_id", nullable = true, length = 20)
+    public String getUserId() {
+        return userId;
+    }
+
+    public void setUserId(String userId) {
+        this.userId = userId;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        TbDeviceDTO that = (TbDeviceDTO) o;
+        return id == that.id && Objects.equals(devId, that.devId) && Objects.equals(registerId, that.registerId) && Objects.equals(devceName, that.devceName) && Objects.equals(devcieModel, that.devcieModel) && Objects.equals(connType, that.connType) && Objects.equals(deviceType, that.deviceType) && Objects.equals(createTime, that.createTime) && Objects.equals(del, that.del) && Objects.equals(remark, that.remark) && Objects.equals(longitude, that.longitude) && Objects.equals(dimension, that.dimension) && Objects.equals(userId, that.userId) && Objects.equals(addr, that.addr);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(id, devId, registerId, devceName, devcieModel, connType, deviceType, createTime, del, remark, longitude, dimension, userId, addr);
+    }
+}

src/main/java/com/usky/entity/mqtt/TbDeviceInfoDTO.java

@@ -0,0 +1,149 @@
+package com.usky.entity.mqtt;
+
+import javax.persistence.*;
+import java.util.Objects;
+
+/**
+ * @author laowo
+ * @version v1.0
+ * @date 2021/8/10 16:11
+ * @description TODO
+ **/
+@Entity
+@Table(name = "tb_device_info", schema = "jx_cover", catalog = "")
+public class TbDeviceInfoDTO {
+    private int id;
+    private String devId;
+    private String registerId;
+    private String deviceName;
+    private String property;
+    private String serial;
+    private String value;
+    private String timeStamp;
+    private String createTime;
+    private String devceType;
+    private String devcieModel;
+
+    @Id
+    @Column(name = "id", nullable = false)
+    public int getId() {
+        return id;
+    }
+
+    public void setId(int id) {
+        this.id = id;
+    }
+
+    @Basic
+    @Column(name = "dev_id", nullable = false, length = 50)
+    public String getDevId() {
+        return devId;
+    }
+
+    public void setDevId(String devId) {
+        this.devId = devId;
+    }
+
+    @Basic
+    @Column(name = "register_id", nullable = true, length = 50)
+    public String getRegisterId() {
+        return registerId;
+    }
+
+    public void setRegisterId(String registerId) {
+        this.registerId = registerId;
+    }
+
+    @Basic
+    @Column(name = "device_name", nullable = true, length = 50)
+    public String getDeviceName() {
+        return deviceName;
+    }
+
+    public void setDeviceName(String deviceName) {
+        this.deviceName = deviceName;
+    }
+
+    @Basic
+    @Column(name = "property", nullable = true, length = 50)
+    public String getProperty() {
+        return property;
+    }
+
+    public void setProperty(String property) {
+        this.property = property;
+    }
+
+    @Basic
+    @Column(name = "serial", nullable = true, length = 50)
+    public String getSerial() {
+        return serial;
+    }
+
+    public void setSerial(String serial) {
+        this.serial = serial;
+    }
+
+    @Basic
+    @Column(name = "value", nullable = true, length = 50)
+    public String getValue() {
+        return value;
+    }
+
+    public void setValue(String value) {
+        this.value = value;
+    }
+
+    @Basic
+    @Column(name = "time_stamp", nullable = true, length = 50)
+    public String getTimeStamp() {
+        return timeStamp;
+    }
+
+    public void setTimeStamp(String timeStamp) {
+        this.timeStamp = timeStamp;
+    }
+
+    @Basic
+    @Column(name = "create_time", nullable = true, length = 50)
+    public String getCreateTime() {
+        return createTime;
+    }
+
+    public void setCreateTime(String createTime) {
+        this.createTime = createTime;
+    }
+
+    @Basic
+    @Column(name = "devce_type", nullable = true, length = 50)
+    public String getDevceType() {
+        return devceType;
+    }
+
+    public void setDevceType(String devceType) {
+        this.devceType = devceType;
+    }
+
+    @Basic
+    @Column(name = "devcie_model", nullable = true, length = 50)
+    public String getDevcieModel() {
+        return devcieModel;
+    }
+
+    public void setDevcieModel(String devcieModel) {
+        this.devcieModel = devcieModel;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        TbDeviceInfoDTO that = (TbDeviceInfoDTO) o;
+        return id == that.id && Objects.equals(devId, that.devId) && Objects.equals(registerId, that.registerId) && Objects.equals(deviceName, that.deviceName) && Objects.equals(property, that.property) && Objects.equals(serial, that.serial) && Objects.equals(value, that.value) && Objects.equals(timeStamp, that.timeStamp) && Objects.equals(createTime, that.createTime) && Objects.equals(devceType, that.devceType) && Objects.equals(devcieModel, that.devcieModel);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(id, devId, registerId, deviceName, property, serial, value, timeStamp, createTime, devceType, devcieModel);
+    }
+}

src/main/java/com/usky/entity/mqtt/vo/TbDeviceAlarmsVO.java

@@ -0,0 +1,45 @@
+package com.usky.entity.mqtt.vo;
+
+import com.usky.entity.mqtt.TbDeviceAlarmsDTO;
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.Data;
+
+import javax.persistence.*;
+import java.util.Objects;
+
+/**
+ * @author laowo
+ * @version v1.0
+ * @date 2021/8/10 14:19
+ * @description TODO
+ **/
+
+@Data
+@ApiModel(value = "设备告警VO")
+public class TbDeviceAlarmsVO {
+    private int id;
+    @ApiModelProperty("设备编码")
+    private String devId;
+    @ApiModelProperty("设备本地注册编码")
+    private String registerId;
+    @ApiModelProperty("设备名称")
+    private String deviceName;
+    @ApiModelProperty("设备类型")
+    private String deviceType;
+    @ApiModelProperty("设备型号")
+    private String devcieModel;
+    @ApiModelProperty("设备属性名")
+    private String property;
+    @ApiModelProperty("属性排序")
+    private String serial;
+    @ApiModelProperty("属性值")
+    private String value;
+    @ApiModelProperty("告警状态 0 未消除 1已消除")
+    private Integer status;
+    @ApiModelProperty("数据插入时间")
+    private String createTime;
+    @ApiModelProperty("告警上传时间")
+    private String timeStamp;
+
+}

src/main/java/com/usky/entity/mqtt/vo/TbDeviceInfoVO.java

@@ -0,0 +1,45 @@
+package com.usky.entity.mqtt.vo;
+
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.Data;
+
+import javax.persistence.*;
+import java.util.Objects;
+
+/**
+ * @author laowo
+ * @version v1.0
+ * @date 2021/8/10 14:19
+ * @description TODO
+ **/
+@Data
+@ApiModel(value = "设备心跳VO")
+public class TbDeviceInfoVO {
+    @ApiModelProperty("主键id")
+    private int id;
+    @ApiModelProperty("设备编号")
+    private String devId;
+    @ApiModelProperty("设备本地注册编码")
+    private String registerId;
+    @ApiModelProperty("属性名称")
+    private String property;
+    @ApiModelProperty("属性编号")
+    private String serial;
+    @ApiModelProperty("属性值")
+    private String value;
+    @ApiModelProperty("数据插入时间")
+    private String createTime;
+    @ApiModelProperty("设备名称")
+    private String deviceName;
+    @ApiModelProperty("数据上传时间")
+    private String timeStamp;
+    @ApiModelProperty("设备类型")
+    private String devceType;
+    @ApiModelProperty("设备型号")
+    private String devcieModel;
+    @ApiModelProperty("设备安装位置")
+    private String addr;
+
+
+}

src/main/java/com/usky/entity/mqtt/vo/TbDeviceVO.java

@@ -0,0 +1,47 @@
+package com.usky.entity.mqtt.vo;
+
+import com.usky.entity.mqtt.TbDeviceDTO;
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.Data;
+
+import javax.persistence.*;
+import java.util.Objects;
+
+/**
+ * @author laowo
+ * @version v1.0
+ * @date 2021/8/10 14:19
+ * @description TODO
+ **/
+@Data
+@ApiModel(value = "设备VO")
+public class TbDeviceVO {
+    @ApiModelProperty("主键ID")
+    private int id;
+    @ApiModelProperty("设备编号")
+    private String devId;
+    @ApiModelProperty("设备本地注册编码")
+    private String registerId;
+    @ApiModelProperty("设备名称")
+    private String devceName;
+    @ApiModelProperty("设备型号")
+    private String devcieModel;
+    @ApiModelProperty("通信方式 G:4G N:NB L:LoRa S:专线 B:总线")
+    private String connType;
+    @ApiModelProperty("设备类型")
+    private String deviceType;
+    @ApiModelProperty("添加时间")
+    private String createTime;
+    @ApiModelProperty("0未删除 1已删除")
+    private Integer del;
+    @ApiModelProperty("备注")
+    private String remark;
+    @ApiModelProperty("经度")
+    private String longitude;
+    @ApiModelProperty("维度")
+    private String dimension;
+    @ApiModelProperty("维保人编号")
+    private String userId;
+
+}

src/main/java/com/usky/entity/mqtt/vo/TbDeviceVOTop.java

@@ -0,0 +1,49 @@
+package com.usky.entity.mqtt.vo;
+
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.Data;
+
+/**
+ * @author laowo
+ * @version v1.0
+ * @date 2021/8/10 14:19
+ * @description TODO
+ **/
+@Data
+@ApiModel(value = "首页实时设备VO")
+public class TbDeviceVOTop<T> {
+    @ApiModelProperty("主键ID")
+    private int id;
+    @ApiModelProperty("设备编号")
+    private String devId;
+    @ApiModelProperty("设备本地注册编码")
+    private String registerId;
+    @ApiModelProperty("设备名称")
+    private String devceName;
+    @ApiModelProperty("设备型号")
+    private String devcieModel;
+    @ApiModelProperty("通信方式 G:4G N:NB L:LoRa S:专线 B:总线")
+    private String connType;
+    @ApiModelProperty("设备类型")
+    private String deviceType;
+    @ApiModelProperty("添加时间")
+    private String createTime;
+    @ApiModelProperty("0未删除 1已删除")
+    private Integer del;
+    @ApiModelProperty("备注")
+    private String remark;
+    @ApiModelProperty("经度")
+    private String longitude;
+    @ApiModelProperty("维度")
+    private String dimension;
+    @ApiModelProperty("姓名")
+    private String userName;
+    @ApiModelProperty("手机号")
+    private String phone;
+    @ApiModelProperty("设备安装位置")
+    private String addr;
+    @ApiModelProperty("实时数据")
+    private T data;
+
+}

src/main/java/com/usky/mqtthandler/Mqtt2MessageHandler.java

@@ -0,0 +1,66 @@
+package com.usky.mqtthandler;
+
+import com.usky.config.webScoket.WsSessionManager;
+import com.usky.constant.CommonConstant;
+import com.usky.service.mqtt.MqttService;
+import lombok.SneakyThrows;
+import lombok.extern.log4j.Log4j2;
+import net.sf.json.JSONObject;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.integration.annotation.ServiceActivator;
+import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.MessagingException;
+import org.springframework.stereotype.Component;
+import org.springframework.web.socket.TextMessage;
+
+import javax.annotation.PostConstruct;
+
+@Log4j2
+@Component
+public class Mqtt2MessageHandler implements MessageHandler {
+    @Autowired
+    private MqttPahoMessageDrivenChannelAdapter channel1MqttChannelAdapter;
+
+    @Autowired
+    private MqttService mqttService;
+
+    @PostConstruct
+    private void addFailEvent() {
+        channel1MqttChannelAdapter.setApplicationEventPublisher((event) -> {
+            log.error("【MQTT】失败事件 {}", event);
+        });
+    }
+
+    @SneakyThrows
+    @ServiceActivator(inputChannel = "channel1")
+    @Override
+    public void handleMessage(Message<?> message) throws MessagingException {
+        String payload = (String) message.getPayload();
+        JSONObject data = JSONObject.fromObject(payload);
+        if (data.has("type")) {
+            if (CommonConstant.MQTT_MESSAGE_TYPE_INFO.equals(data.get("type"))) {
+
+                log.info("【MQTT】心跳数据{}解析】", message);
+
+                mqttService.infoSava(data);
+
+            } else if (CommonConstant.MQTT_MESSAGE_TYPE_ALARM.equals(data.get("type"))) {
+
+                log.info("【MQTT】告警数据{}解析】", message);
+
+                mqttService.alarmSendAndSava(data);
+            } else {
+
+                log.error("【MQTT】数据类型异常{}", message);
+            }
+        } else {
+
+            log.error("【MQTT】数据格式异常{}", message);
+
+        }
+
+    }
+
+}

src/main/java/com/usky/service/mqtt/MqttService.java

@@ -0,0 +1,22 @@
+package com.usky.service.mqtt;
+
+import net.sf.json.JSONObject;
+
+/**
+ * @author laowo
+ */
+public interface MqttService {
+    /**
+     * 心跳数据保存
+     *
+     * @param data
+     */
+    void infoSava(JSONObject data);
+
+    /**
+     * 告警数据推送和保存
+     *
+     * @param data
+     */
+    void alarmSendAndSava(JSONObject data);
+}

src/main/java/com/usky/service/mqtt/MqttServiceImpl.java

@@ -0,0 +1,95 @@
+package com.usky.service.mqtt;
+
+import com.usky.config.webScoket.WsSessionManager;
+import com.usky.dao.impl.BaseDaoImpl;
+import com.usky.entity.mqtt.TbDeviceAlarmsDTO;
+import com.usky.entity.mqtt.TbDeviceInfoDTO;
+import com.usky.utils.RedisUtil;
+import com.usky.utils.TimeUtil;
+import net.sf.json.JSONArray;
+import net.sf.json.JSONObject;
+import org.hibernate.Session;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+import org.springframework.web.socket.TextMessage;
+
+/**
+ * @author laowo
+ * @version v1.0
+ * @date 2021/8/10 15:12
+ * @description TODO
+ **/
+@Service
+public class MqttServiceImpl extends BaseDaoImpl implements MqttService {
+    @Autowired
+    private RedisUtil redisUtil;
+
+    @Override
+    public void infoSava(JSONObject data) {
+        //获取设备列表
+        JSONArray devs = (JSONArray) data.get("devs");
+        for (int i = 0; i < devs.size(); i++) {
+            //获取属性列表
+            JSONObject device = (JSONObject) devs.get(i);
+            device.put("type", "INFO");
+            redisUtil.set((String) device.get("devId"), device.toString());
+            JSONArray dp = (JSONArray) device.get("dp");
+            for (int j = 0; j < dp.size(); j++) {
+                TbDeviceInfoDTO deviceInfoDTO = new TbDeviceInfoDTO();
+                //设备属性信息
+                JSONObject devDp = (JSONObject) dp.get(j);
+                deviceInfoDTO.setDevId((String) device.get("devId"));
+                deviceInfoDTO.setDevcieModel((String) device.get("deviceModel"));
+                deviceInfoDTO.setDeviceName((String) device.get("deviceName"));
+                deviceInfoDTO.setDevceType((String) device.get("deviceType"));
+                deviceInfoDTO.setTimeStamp((String) device.get("timeStamp"));
+                deviceInfoDTO.setProperty((String) devDp.get("property"));
+                deviceInfoDTO.setSerial((String) devDp.get("serial"));
+                deviceInfoDTO.setValue((String) devDp.get("value"));
+                deviceInfoDTO.setCreateTime(TimeUtil.getTime());
+                //设备实时状体存续
+                Session session = getSession();
+                session.save(deviceInfoDTO);
+                session.flush();
+                session.clear();
+            }
+        }
+    }
+
+    @Override
+    public void alarmSendAndSava(JSONObject data) {
+//获取设备列表
+        JSONArray devs = (JSONArray) data.get("alarams");
+        for (int i = 0; i < devs.size(); i++) {
+            //获取属性列表
+            JSONObject device = (JSONObject) devs.get(i);
+            device.put("type", "ALARM");
+            JSONArray dp = (JSONArray) device.get("dp");
+            redisUtil.set((String) device.get("devId"), device.toString());
+            WsSessionManager.sendMessageToAll(new TextMessage(device.toString()));
+            for (int j = 0; j < dp.size(); j++) {
+                TbDeviceAlarmsDTO deviceAlarmsDTO = new TbDeviceAlarmsDTO();
+                //设备属性信息
+                JSONObject devDp = (JSONObject) dp.get(j);
+                deviceAlarmsDTO.setDevId((String) device.get("devId"));
+                deviceAlarmsDTO.setDevcieModel((String) device.get("devcieModel"));
+                deviceAlarmsDTO.setDeviceName((String) device.get("deviceName"));
+                deviceAlarmsDTO.setDeviceType((String) device.get("deviceType"));
+                deviceAlarmsDTO.setTimeStamp((String) device.get("timeStamp"));
+                deviceAlarmsDTO.setProperty((String) devDp.get("property"));
+                deviceAlarmsDTO.setSerial((String) devDp.get("serial"));
+                deviceAlarmsDTO.setValue((String) devDp.get("value"));
+                deviceAlarmsDTO.setStatus((Integer) devDp.get("status"));
+                deviceAlarmsDTO.setCreateTime(TimeUtil.getTime());
+                //告警存储
+
+                Session session = getSession();
+                //webSocket告警推送
+
+                session.save(deviceAlarmsDTO);
+                session.flush();
+                session.clear();
+            }
+        }
+    }
+}

src/main/java/com/usky/service/top/TopServcie.java

@@ -0,0 +1,43 @@
+package com.usky.service.top;
+
+import com.usky.entity.mqtt.TbDeviceInfoDTO;
+import com.usky.entity.mqtt.vo.TbDeviceAlarmsVO;
+import com.usky.entity.mqtt.vo.TbDeviceInfoVO;
+import com.usky.entity.mqtt.vo.TbDeviceVO;
+import com.usky.entity.mqtt.vo.TbDeviceVOTop;
+import com.usky.utils.Result;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * @author laowo
+ */
+public interface TopServcie {
+    /**
+     * 首页设备状态统计
+     *
+     * @return
+     */
+    Map<String, Object> listDeviceStatus();
+
+    /**
+     * 首页七天告警统计
+     *
+     * @return
+     */
+    List<Object> listAlarmsSe();
+
+    /**
+     * 首页设备查询
+     * @return
+     */
+    List<TbDeviceVOTop> listDevcie();
+
+    /**
+     * 设备液位查询
+     * @return
+     * @param devId
+     */
+    List<TbDeviceInfoDTO> listYw(String devId);
+}

src/main/java/com/usky/service/top/TopServcieImpl.java

@@ -0,0 +1,112 @@
+package com.usky.service.top;
+
+import com.usky.dao.impl.BaseDaoImpl;
+import com.usky.entity.mqtt.TbDeviceAlarmsDTO;
+import com.usky.entity.mqtt.TbDeviceDTO;
+import com.usky.entity.mqtt.TbDeviceInfoDTO;
+import com.usky.entity.mqtt.vo.TbDeviceVO;
+import com.usky.entity.mqtt.vo.TbDeviceVOTop;
+import com.usky.utils.BeanHelp;
+import com.usky.utils.RedisUtil;
+import net.sf.json.JSONObject;
+import org.hibernate.transform.Transformers;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import java.util.*;
+import java.util.stream.Collectors;
+
+/**
+ * @author laowo
+ * @version v1.0
+ * @date 2021/8/10 17:34
+ * @description TODO
+ **/
+@Service
+public class TopServcieImpl extends BaseDaoImpl implements TopServcie {
+    @Autowired
+    private RedisUtil redisUtil;
+
+    @Override
+    public Map<String, Object> listDeviceStatus() {
+        List<TbDeviceDTO> list = getSession().createQuery("from TbDeviceDTO where del=0").list();
+        //所有告警
+        List<TbDeviceAlarmsDTO> alarmsDTOList = getSession().createQuery("from TbDeviceAlarmsDTO t where t.status=0").list();
+        //离线数量
+        long unOline = alarmsDTOList.stream().filter(s -> "1".equals(s.getSerial())).count();
+        //故障数量
+        long malfunction = alarmsDTOList.stream().filter(s -> "32".equals(s.getSerial())).count();
+        Map<String, Object> result = new HashMap<>();
+        //设备总数
+        result.put("deviceTotalNum", list.size());
+        result.put("unOline", unOline);
+        result.put("malfunction", malfunction);
+        return result;
+    }
+
+    @Override
+    public List<Object> listAlarmsSe() {
+        //查询七天告警数据
+        List<TbDeviceAlarmsDTO> list = getSession().createSQLQuery("SELECT id,dev_id as devId,register_id AS registerId,device_name AS deviceName,device_type AS deviceType,devcie_model AS devcieModel,property,serial,`value`,`status`,create_time AS createTime ,time_stamp AS `timeStamp` FROM tb_device_alarms WHERE date_sub(curdate(), interval 7 day) < date(create_time) order by id ASC")
+                .setResultTransformer(Transformers.aliasToBean(TbDeviceAlarmsDTO.class)).list();
+        //根据时间分组
+        Map<String, List<TbDeviceAlarmsDTO>> collect = list.stream().collect(Collectors.groupingBy(TbDeviceAlarmsDTO::getCreateTime));
+        List<Object> resultList = new ArrayList<>();
+        Set<String> strings = collect.keySet();
+        for (String string : strings) {
+            Map<String, Object> map = new HashMap<>();
+            List<TbDeviceAlarmsDTO> alarmsDTOList = collect.get(string);
+            //离线数量
+            long count = alarmsDTOList.stream().filter(s -> "1".equals(s.getSerial())).count();
+            map.put("time", string);
+            map.put("offline", count);
+            map.put("total", alarmsDTOList.size());
+            resultList.add(map);
+        }
+        //  List<TbDeviceAlarmsVO> tbDeviceAlarmsVOS = BeanHelp.copyWithCollection(list, TbDeviceAlarmsVO.class);
+        return resultList;
+    }
+
+    @Override
+    public List<TbDeviceVOTop> listDevcie() {
+
+
+        List<TbDeviceDTO> list = getSession().createQuery("from TbDeviceDTO t where t.del=0").list();
+        List<TbDeviceVOTop> tbDeviceVOS = BeanHelp.copyWithCollection(list, TbDeviceVOTop.class);
+        for (TbDeviceVOTop tbDeviceVOTop : tbDeviceVOS) {
+                  Object o = redisUtil.get(tbDeviceVOTop.getDevId());
+            JSONObject jsonObject = JSONObject.fromObject(o);
+            tbDeviceVOTop.setData(jsonObject);
+            tbDeviceVOTop.setUserName("吉超薄");
+            tbDeviceVOTop.setPhone("15365185591");
+        }
+        return tbDeviceVOS;
+    }
+
+    @Override
+    public List<TbDeviceInfoDTO> listYw(String devId) {
+        StringBuilder sb = new StringBuilder();
+        sb.append("SELECT\n" +
+                "\tid,\n" +
+                "\tdev_id AS devId,\n" +
+                "\tdevice_name AS deviceName,\n" +
+                "\tproperty,\n" +
+                "\tserial,\n" +
+                "\t`value`,\n" +
+                "\ttime_stamp AS `timeStamp`,\n" +
+                "\t create_time AS createTime,\n" +
+                "\t\tdevce_type AS devceType,\n" +
+                "\tdevcie_model AS devcieModel \n" +
+                "FROM\n" +
+                "\ttb_device_info t \n" +
+                "WHERE\n" +
+                "\tt.serial = '75' \n" +
+                "\tAND date_sub( curdate(), INTERVAL 7 DAY ) < date( create_time ) ");
+
+        sb.append(" AND  dev_id='" + devId + "' GROUP BY date( create_time )");
+//
+        List<TbDeviceInfoDTO> list = getSession().createSQLQuery(sb.toString()).setResultTransformer(Transformers.aliasToBean(TbDeviceInfoDTO.class)).list();
+        return list;
+    }
+}
+

src/main/java/com/usky/utils/mqtt/MqttUtils.java

@@ -0,0 +1,153 @@
+package com.usky.utils.mqtt;
+
+import lombok.extern.log4j.Log4j2;
+import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
+import org.springframework.integration.mqtt.support.MqttHeaders;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.messaging.Message;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+
+/**
+ * mqtt工具类,可以根据通道名称发送消息
+ */
+@Log4j2
+public class MqttUtils {
+
+    /**
+     * qos 0
+     */
+    public static final int QOS_0 = 0;
+    /**
+     * qos 1
+     */
+    public static final int QOS_1 = 1;
+    /**
+     * qos 2
+     */
+    public static final int QOS_2 = 2;
+
+    private final static Map<String, MqttPahoMessageHandler> HANDLER_MAP = new HashMap<>(16);
+    public final static String CHANNEL_NAME_SUFFIX = "MqttPahoMessageHandler";
+
+    /**
+     * 存放handler
+     *
+     * @param channelName
+     * @param handler
+     */
+    public static void put(String channelName, MqttPahoMessageHandler handler) {
+        HANDLER_MAP.put(channelName + CHANNEL_NAME_SUFFIX, handler);
+    }
+
+    /**
+     * 发送消息
+     *
+     * @param topic       要发送的主题
+     * @param message     消息内容
+     * @param qos         qos级别
+     * @param channelName 发送到指定的通道
+     */
+    public static void sendMessage(String topic, String message, int qos, String channelName) {
+        MqttPahoMessageHandler handler = getHandler(channelName);
+        Message<String> mqttMessage = MessageBuilder.withPayload(message).setHeader(MqttHeaders.TOPIC, topic)
+                .setHeader(MqttHeaders.QOS, qos).build();
+        handler.handleMessage(mqttMessage);
+    }
+
+    /**
+     * 发送消息,默认qos级别为1
+     *
+     * @param topic       要发送的主题
+     * @param message     消息内容
+     * @param channelName 发送到指定的通道
+     */
+    public static void sendMessage(String topic, String message, String channelName) {
+        MqttPahoMessageHandler handler = getHandler(channelName);
+        Message<String> mqttMessage = MessageBuilder.withPayload(message).setHeader(MqttHeaders.TOPIC, topic)
+                .setHeader(MqttHeaders.QOS, QOS_1).build();
+        handler.handleMessage(mqttMessage);
+    }
+
+    /**
+     * 发送消息
+     *
+     * @param mqttMessage 消息
+     * @param channelName 发送到指定的通道
+     */
+    public static void sendMessage(Message<String> mqttMessage, String channelName) {
+        MqttPahoMessageHandler handler = getHandler(channelName);
+        handler.handleMessage(mqttMessage);
+    }
+
+    /**
+     * 如果只有一个通道将使用该通道发送消息
+     *
+     * @param topic
+     * @param message
+     * @param qos
+     */
+    public static void sendMessage(String topic, String message, int qos) {
+        MqttPahoMessageHandler handler = getDefaultHeadler();
+        Message<String> mqttMessage = MessageBuilder.withPayload(message).setHeader(MqttHeaders.TOPIC, topic)
+                .setHeader(MqttHeaders.QOS, qos).build();
+        handler.handleMessage(mqttMessage);
+    }
+
+    /**
+     * 如果只有一个通道将使用该通道发送消息，默认qos级别为1
+     *
+     * @param topic
+     * @param message
+     */
+    public static void sendMessage(String topic, String message) {
+        MqttPahoMessageHandler handler = getDefaultHeadler();
+        Message<String> mqttMessage = MessageBuilder.withPayload(message).setHeader(MqttHeaders.TOPIC, topic)
+                .setHeader(MqttHeaders.QOS, QOS_1).build();
+        handler.handleMessage(mqttMessage);
+    }
+
+    /**
+     * 如果只有一个通道将使用该通道发送消息，默认qos级别为1
+     *
+     * @param mqttMessage 消息信息
+     */
+    public static void sendMessage(Message<String> mqttMessage) {
+        MqttPahoMessageHandler handler = getDefaultHeadler();
+        handler.handleMessage(mqttMessage);
+    }
+
+    /**
+     * 获取默认的handler
+     *
+     * @return
+     */
+    private static MqttPahoMessageHandler getDefaultHeadler() {
+        Collection<MqttPahoMessageHandler> values = HANDLER_MAP.values();
+        Iterator<MqttPahoMessageHandler> iterator = values.iterator();
+        MqttPahoMessageHandler handler = iterator.next();
+        if (handler == null) {
+            log.error("发送消息失败,无可用的headler");
+            throw new RuntimeException("发送消息失败,无可用的headler");
+        }
+        return handler;
+    }
+
+    /**
+     * 根据通道获取handler
+     *
+     * @param channelName
+     * @return
+     */
+    private static MqttPahoMessageHandler getHandler(String channelName) {
+        MqttPahoMessageHandler handler = HANDLER_MAP.get(channelName + CHANNEL_NAME_SUFFIX);
+        if (handler == null) {
+            log.error("未查询到相应通道{}的handler，存在的通道名称{}", channelName, HANDLER_MAP.keySet());
+            throw new IllegalArgumentException("未查询到相应通道" + channelName + "的handler");
+        }
+        return handler;
+    }
+}

src/main/main.iml

@@ -0,0 +1,11 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<module type="JAVA_MODULE" version="4">
+  <component name="NewModuleRootManager" inherit-compiler-output="true">
+    <exclude-output />
+    <content url="file://$MODULE_DIR$">
+      <sourceFolder url="file://$MODULE_DIR$/java" isTestSource="false" />
+    </content>
+    <orderEntry type="inheritedJdk" />
+    <orderEntry type="sourceFolder" forTests="false" />
+  </component>
+</module>

src/main/resources/application.yml

@@ -1,6 +1,6 @@
 #开发环境
 server:
-  port: 8082
+  port: 8085
 spring:
   redis:
     lettuce:
@@ -28,12 +28,12 @@ spring:
       testOnReturn: false
       poolPreparedStatements: true
       aopPatterns: com.usky.*
-      filters: stat,slf4j
+     # filters: stat,slf4j
       maxPoolPreparedStatementPerConnectionSize: 20
-      useGlobalDataSourceStat: true
+      useGlobalDataSourceStat: false
       connectionProperties: druid.stat.mergeSql=true;druid.stat.slowSqlMillis=500
       db-type: com.alibaba.druid.pool.DruidDataSource
-      url: jdbc:mysql://47.111.81.118:3306/sd_party_school?useunicode=true&characterEncoding=utf8&serverTimezone=Asia/Shanghai
+      url: jdbc:mysql://47.111.81.118:3306/jx_cover?useunicode=true&characterEncoding=utf8&serverTimezone=Asia/Shanghai
       username: root
       password: Yt2021
   jpa:
@@ -65,7 +65,35 @@ sms:
   version: 2017-05-25 # API版本，固定值
   regionID: cn-hangzhou # 区域id
 
-
+mqtt:
+  config:
+    channel1:
+      consumer-enable: true
+      producer-enable: true
+   #   url: [tcp://124.71.175.91:1883]
+      url: [tcp://47.98.201.73:1883]
+      topics: [usky,test]
+      qos: [1,0]
+  #   username: wjzn2021
+  #   password: wjzn2021
+      username: usky
+      password: usky
+      timeout: 60
+      kep-alive-interval: 60
+      async: true
+      client-id-append-ip: true
+      consumer-client-id: consumer_client_test1
+      producer-client-id: producer_client_test1
+      consumer-will:
+        qos: 1
+        topic: will_topic
+        payload: '{"id": "consumer_client_test1"}'
+        retained: false
+      producer-will:
+        qos: 1
+        topic: will_topic
+        payload: '{"id": "producer_client_test1"}'
+        retained: false
 
 
 

src/main/resources/mqtt-spring.properties

@@ -1,10 +0,0 @@
-mqtt.client-id-prefix=dev
-#mqtt.url=tcp://www.iiotsaas.com:1883
-#mqtt.url=tcp://47.98.201.73:1883
-mqtt.url=tcp://124.71.175.91:1883
-mqtt.username=wjzn2021
-mqtt.password=wjzn2021
-mqtt.keep-alive-interval=60
-mqtt.completionTimeout=5000
-#mqtt.sub-topics=/edge/usky/sn/#,usky
-mqtt.sub-topics=usky

src/main/test/java/com/usky/controller/Basetest.java

@@ -1,4 +1,4 @@
-package com.usky.controller;
+package java.com.usky.controller;
 
 import org.junit.Test;
 import org.junit.runner.RunWith;

src/main/test/java/com/usky/controller/TestTest.java

@@ -1,7 +1,8 @@
-package com.usky.controller;
+package java.com.usky.controller;
 
 import org.junit.Test;
 
+import java.com.usky.controller.Basetest;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.function.Consumer;
@@ -102,6 +103,10 @@ public class TestTest implements Basetest {
 
 
 
+
+
+
+
     
 
 

src/main/test/java/com/usky/controller/top/TopConfigControllerTest.java

@@ -1,24 +0,0 @@
-package com.usky.controller.top;
-
-import com.usky.controller.Basetest;
-import com.usky.entity.top.vo.TopConfigVO;
-import com.usky.service.top.TopService.TopService;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.test.context.SpringBootTest;
-import org.springframework.test.context.junit4.SpringRunner;
-
-import java.util.List;
-
-@RunWith(SpringRunner.class)
-@SpringBootTest
-public class TopConfigControllerTest implements Basetest {
-    @Autowired
-    private TopService topService;
-    @Test
-    public void listTopConfig() {
-        List<TopConfigVO> topConfigVOS = topService.queryTopConfig("SD");
-        System.out.println("topConfigVOS = " + topConfigVOS);
-    }
-}

src/main/test/test.iml

@@ -0,0 +1,12 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<module type="JAVA_MODULE" version="4">
+  <component name="NewModuleRootManager" inherit-compiler-output="true">
+    <exclude-output />
+    <content url="file://$MODULE_DIR$">
+      <sourceFolder url="file://$MODULE_DIR$/java" isTestSource="true" />
+    </content>
+    <orderEntry type="inheritedJdk" />
+    <orderEntry type="sourceFolder" forTests="false" />
+    <orderEntry type="module" module-name="main" />
+  </component>
+</module>