方舱对接mqtt数据

pom.xml

@@ -165,10 +165,10 @@
             </exclusions>
         </dependency>
 
-<!--        <dependency>-->
-<!--            <groupId>redis.clients</groupId>-->
-<!--            <artifactId>jedis</artifactId>-->
-<!--        </dependency>-->
+        <dependency>
+            <groupId>redis.clients</groupId>
+            <artifactId>jedis</artifactId>
+        </dependency>
 
         <dependency>
             <groupId>io.springfox</groupId>

src/main/java/com/template/common/mqtt/MqttConfiguration.java

@@ -1,86 +1,86 @@
-//package com.template.common.mqtt;
-//
-//import org.springframework.beans.factory.annotation.Autowired;
-//import org.springframework.boot.context.properties.ConfigurationProperties;
-//import org.springframework.context.annotation.Bean;
-//import org.springframework.context.annotation.Configuration;
-//import org.springframework.stereotype.Component;
-//
-//@Component
-//@Configuration
-//@ConfigurationProperties(prefix = "mqtt")
-//public class MqttConfiguration {
-//
-//    private String host;
-//    private String clientId;
-//    private String username;
-//    private String password;
-//    private String topic;
-//    private int timeout;
-//    private int KeepAlive;
-//
-//    public String getClientId() {
-//        return clientId;
-//    }
-//
-//    public void setClientId(String clientId) {
-//        this.clientId = clientId;
-//    }
-//
-//    public int getKeepAlive() {
-//        return KeepAlive;
-//    }
-//
-//    public void setKeepAlive(int keepAlive) {
-//        KeepAlive = keepAlive;
-//    }
-//
-//    public String getUsername() {
-//        return username;
-//    }
-//
-//    public void setUsername(String username) {
-//        this.username = username;
-//    }
-//
-//    public String getPassword() {
-//        return password;
-//    }
-//
-//    public void setPassword(String password) {
-//        this.password = password;
-//    }
-//
-//    public String getTopic() {
-//        return topic;
-//    }
-//
-//    public void setTopic(String topic) {
-//        this.topic = topic;
-//    }
-//
-//    public int getTimeout() {
-//        return timeout;
-//    }
-//
-//    public void setTimeout(int timeout) {
-//        this.timeout = timeout;
-//    }
-//
-//
-//    public String getHost() {
-//        return host;
-//    }
-//
-//    public void setHost(String host) {
-//        this.host = host;
-//    }
-//
-//    @Autowired
-//    private MqttPushClient mqttPushClient;
-//    @Bean
-//    public MqttPushClient getMqttPushClient(){
-//        mqttPushClient.connect(host, clientId, username, password, timeout,KeepAlive);
-//        return mqttPushClient;
-//    }
-//}
+package com.template.common.mqtt;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.stereotype.Component;
+
+@Component
+@Configuration
+@ConfigurationProperties(prefix = "mqtt")
+public class MqttConfiguration {
+
+    private String host;
+    private String clientId;
+    private String username;
+    private String password;
+    private String topic;
+    private int timeout;
+    private int KeepAlive;
+
+    public String getClientId() {
+        return clientId;
+    }
+
+    public void setClientId(String clientId) {
+        this.clientId = clientId;
+    }
+
+    public int getKeepAlive() {
+        return KeepAlive;
+    }
+
+    public void setKeepAlive(int keepAlive) {
+        KeepAlive = keepAlive;
+    }
+
+    public String getUsername() {
+        return username;
+    }
+
+    public void setUsername(String username) {
+        this.username = username;
+    }
+
+    public String getPassword() {
+        return password;
+    }
+
+    public void setPassword(String password) {
+        this.password = password;
+    }
+
+    public String getTopic() {
+        return topic;
+    }
+
+    public void setTopic(String topic) {
+        this.topic = topic;
+    }
+
+    public int getTimeout() {
+        return timeout;
+    }
+
+    public void setTimeout(int timeout) {
+        this.timeout = timeout;
+    }
+
+
+    public String getHost() {
+        return host;
+    }
+
+    public void setHost(String host) {
+        this.host = host;
+    }
+
+    @Autowired
+    private MqttPushClient mqttPushClient;
+    @Bean
+    public MqttPushClient getMqttPushClient(){
+        mqttPushClient.connect(host, clientId, username, password, timeout,KeepAlive);
+        return mqttPushClient;
+    }
+}

src/main/java/com/template/common/mqtt/MqttPushClient.java

@@ -1,166 +1,167 @@
-//package com.template.common.mqtt;
-//
-//import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
-//import org.slf4j.Logger;
-//import org.slf4j.LoggerFactory;
-//import org.springframework.beans.factory.annotation.Autowired;
-//import org.springframework.context.annotation.Configuration;
-//
-//@Configuration
-//public class MqttPushClient {
-//
-//
-//    private static MqttConnectOptions option;
-//
-//    @Autowired
-//    private PushCallback pushCallback;
-//    private static final Logger log = LoggerFactory.getLogger(MqttPushClient.class);
-//
-//    private static MqttClient client;
-//
-//    public static MqttClient getClient() {
-//        return client;
-//    }
-//
-//    public static void setClient(MqttClient client) {
-//        MqttPushClient.client = client;
-//    }
-//
-//    private MqttConnectOptions getOption(String userName, String password, int outTime, int KeepAlive) {
-//        //MQTT连接设置
-//         option = new MqttConnectOptions();
-//        //设置是否清空session,false表示服务器会保留客户端的连接记录，true表示每次连接到服务器都以新的身份连接
-//        option.setCleanSession(false);
-//        //设置连接的用户名
-//        option.setUserName(userName);
-//        //设置连接的密码
-//        option.setPassword(password.toCharArray());
-//        //设置超时时间 单位为秒
-//        option.setConnectionTimeout(outTime);
-//        //设置会话心跳时间 单位为秒 服务器会每隔(1.5*keepTime)秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
-//        option.setKeepAliveInterval(KeepAlive);
-//        //setWill方法，如果项目中需要知道客户端是否掉线可以调用该方法。设置最终端口的通知消息
-//        //option.setWill(topic, "close".getBytes(), 2, true);
-//        return option;
-//    }
-//
-//    /**
-//     * 连接
-//     * @param host
-//     * @param clientID
-//     * @param username
-//     * @param password
-//     * @param timeout
-//     * @param KeepAlive
-//     */
-//    public void connect(String host, String clientID, String username, String password, int timeout, int KeepAlive){
-//        MqttClient client;
-//        try {
-//            client = new MqttClient(host, clientID, new MemoryPersistence());
-//            MqttConnectOptions options = getOption(username,password,timeout,KeepAlive);
-//            MqttPushClient.setClient(client);
-//            try {
-//                client.setCallback(pushCallback);
-//                if (!client.isConnected()) {
-//                    client.connect(options);
-//                    log.info("MQTT连接成功");
-//                }else {//这里的逻辑是如果连接成功就重新连接
-//                    client.disconnect();
-//                    client.connect(options);
-//                    log.info("MQTT断连成功");
-//                }
-//            } catch (Exception e) {
-//                e.printStackTrace();
-//            }
-//        } catch (Exception e) {
-//            e.printStackTrace();
-//        }
-//    }
-//
-//    /**
-//     * 断线重连
-//     * @throws Exception
-//     */
-//    public Boolean reConnect() throws Exception {
-//        Boolean isConnected = false;
-//       if(null != client) {
-//           client.connect(option);
-//        }
-//
-//        if(client.isConnected()){
-//            isConnected = true;
-//        }
-//        return isConnected;
-//    }
-//
-//    /**
-//     * 发布，默认qos为0，非持久化
-//     * @param topic
-//     * @param pushMessage
-//     */
-//    public void publish(String topic,String pushMessage){
-//        publish(0, false, topic, pushMessage);
-//    }
-//
-//    /**
-//     * 发布
-//     * @param qos
-//     * @param retained
-//     * @param topic
-//     * @param pushMessage
-//     */
-//    public void publish(int qos,boolean retained,String topic,String pushMessage){
-//        MqttClient mqttClient = MqttPushClient.getClient();
-//        try {
-//            mqttClient.publish(topic,pushMessage.getBytes(),qos,retained);
-//        } catch (MqttPersistenceException e) {
-//            e.printStackTrace();
-//        } catch (MqttException e) {
-//            e.printStackTrace();
-//        }
-//    }
-//
-//    public void publish2(int qos,boolean retained,String topic,byte[] pushMessage){
-//        MqttMessage message = new MqttMessage();
-//        message.setQos(qos);
-//        message.setRetained(retained);
-//        message.setPayload(pushMessage);
-//        MqttTopic mTopic = MqttPushClient.getClient().getTopic(topic);
-//        if(mTopic==null){
-//            System.out.println("MQTT topic 不存在");
-//            log.error("MQTT topic 不存在");
-//            return;
-//        }
-//        MqttDeliveryToken token;
-//        try {
-//            token = mTopic.publish(message);
-//            token.waitForCompletion();
-//        } catch (MqttPersistenceException e) {
-//            e.printStackTrace();
-//        } catch (MqttException e) {
-//            e.printStackTrace();
-//        }
-//    }
-//
-//
-//    /**
-//     * 订阅某个主题，qos默认为0
-//     * @param topic
-//     */
-//    public void subscribe(String topic){
-//        subscribe(topic,0);
-//    }
-//
-//    /**
-//     * 订阅某个主题
-//     * @param topic
-//     * @param qos
-//     */
-//    public void subscribe(String topic,int qos){
-//        try {
-//            MqttPushClient.getClient().subscribe(topic, qos);
-//        } catch (MqttException e) {
-//            e.printStackTrace();
-//        }
-//    }
-//}
+package com.template.common.mqtt;
+
+import org.eclipse.paho.client.mqttv3.*;
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+public class MqttPushClient {
+
+
+    private static MqttConnectOptions option;
+
+    @Autowired
+    private PushCallback pushCallback;
+    private static final Logger log = LoggerFactory.getLogger(MqttPushClient.class);
+
+    private static MqttClient client;
+
+    public static MqttClient getClient() {
+        return client;
+    }
+
+    public static void setClient(MqttClient client) {
+        MqttPushClient.client = client;
+    }
+
+    private MqttConnectOptions getOption(String userName, String password, int outTime, int KeepAlive) {
+        //MQTT连接设置
+         option = new MqttConnectOptions();
+        //设置是否清空session,false表示服务器会保留客户端的连接记录，true表示每次连接到服务器都以新的身份连接
+        option.setCleanSession(false);
+        //设置连接的用户名
+        option.setUserName(userName);
+        //设置连接的密码
+        option.setPassword(password.toCharArray());
+        //设置超时时间 单位为秒
+        option.setConnectionTimeout(outTime);
+        //设置会话心跳时间 单位为秒 服务器会每隔(1.5*keepTime)秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
+        option.setKeepAliveInterval(KeepAlive);
+        //setWill方法，如果项目中需要知道客户端是否掉线可以调用该方法。设置最终端口的通知消息
+        //option.setWill(topic, "close".getBytes(), 2, true);
+        return option;
+    }
+
+    /**
+     * 连接
+     * @param host
+     * @param clientID
+     * @param username
+     * @param password
+     * @param timeout
+     * @param KeepAlive
+     */
+    public void connect(String host, String clientID, String username, String password, int timeout, int KeepAlive){
+        MqttClient client;
+        try {
+            client = new MqttClient(host, clientID, new MemoryPersistence());
+            MqttConnectOptions options = getOption(username,password,timeout,KeepAlive);
+            MqttPushClient.setClient(client);
+            try {
+                client.setCallback(pushCallback);
+                if (!client.isConnected()) {
+                    client.connect(options);
+                    log.info("MQTT连接成功");
+                }else {//这里的逻辑是如果连接成功就重新连接
+                    client.disconnect();
+                    client.connect(options);
+                    log.info("MQTT断连成功");
+                }
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+
+    /**
+     * 断线重连
+     * @throws Exception
+     */
+    public Boolean reConnect() throws Exception {
+        Boolean isConnected = false;
+       if(null != client) {
+           client.connect(option);
+        }
+
+        if(client.isConnected()){
+            isConnected = true;
+        }
+        return isConnected;
+    }
+
+    /**
+     * 发布，默认qos为0，非持久化
+     * @param topic
+     * @param pushMessage
+     */
+    public void publish(String topic,String pushMessage){
+        publish(0, false, topic, pushMessage);
+    }
+
+    /**
+     * 发布
+     * @param qos
+     * @param retained
+     * @param topic
+     * @param pushMessage
+     */
+    public void publish(int qos,boolean retained,String topic,String pushMessage){
+        MqttClient mqttClient = MqttPushClient.getClient();
+        try {
+            mqttClient.publish(topic,pushMessage.getBytes(),qos,retained);
+        } catch (MqttPersistenceException e) {
+            e.printStackTrace();
+        } catch (MqttException e) {
+            e.printStackTrace();
+        }
+    }
+
+    public void publish2(int qos,boolean retained,String topic,byte[] pushMessage){
+        MqttMessage message = new MqttMessage();
+        message.setQos(qos);
+        message.setRetained(retained);
+        message.setPayload(pushMessage);
+        MqttTopic mTopic = MqttPushClient.getClient().getTopic(topic);
+        if(mTopic==null){
+            System.out.println("MQTT topic 不存在");
+            log.error("MQTT topic 不存在");
+            return;
+        }
+        MqttDeliveryToken token;
+        try {
+            token = mTopic.publish(message);
+            token.waitForCompletion();
+        } catch (MqttPersistenceException e) {
+            e.printStackTrace();
+        } catch (MqttException e) {
+            e.printStackTrace();
+        }
+    }
+
+
+    /**
+     * 订阅某个主题，qos默认为0
+     * @param topic
+     */
+    public void subscribe(String topic){
+        subscribe(topic,0);
+    }
+
+    /**
+     * 订阅某个主题
+     * @param topic
+     * @param qos
+     */
+    public void subscribe(String topic,int qos){
+        try {
+            MqttPushClient.getClient().subscribe(topic, qos);
+        } catch (MqttException e) {
+            e.printStackTrace();
+        }
+    }
+}

src/main/java/com/template/common/mqtt/PushCallback.java

@@ -1,209 +1,156 @@
-//package com.template.common.mqtt;
-//
-//import com.alibaba.fastjson.JSON;
-//import com.baomidou.mybatisplus.core.toolkit.ObjectUtils;
-//import com.template.common.utils.TimeExchange;
-//import com.template.model.Companies;
-//import com.template.model.vo.SensorRediesDetailVo;
-//import com.template.services.SensorDetailService;
-//import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
-//import org.eclipse.paho.client.mqttv3.MqttCallback;
-//import org.eclipse.paho.client.mqttv3.MqttMessage;
-//import org.slf4j.Logger;
-//import org.slf4j.LoggerFactory;
-//import org.springframework.beans.factory.annotation.Autowired;
-//import org.springframework.data.redis.core.RedisTemplate;
-//import org.springframework.stereotype.Component;
-//import org.springframework.stereotype.Repository;
-//
-//import java.util.Date;
-//
-//import static com.template.common.util.Hex.byte2HexStr;
-//
-//@Component
-//@Repository
-//public class PushCallback implements MqttCallback {
-//
-//    private static final Logger log = LoggerFactory.getLogger(PushCallback.class);
-//
-//    @Autowired
-//    private Companiess sensorDetailService;
-//
-//    @Autowired
-//    private MqttPushClient mqttPushClient;
-//
-//    @Autowired
-//    private RedisTemplate redisTemplate;
-//
-//
-//    @Override
-//    public void connectionLost(Throwable cause) {
-//        // 当连接丢失时，这个方法会被调用
-//        // 连接丢失后，一般在这里面进行重连
-//        System.out.println(TimeExchange.getTime()+"[MQTT] 连接断开，30S之后尝试重连..."+cause);
-//        while (true) {
-//            try {
-//                Thread.sleep(30000);
-////                MqttPushClient mqttPushClient = new MqttPushClient();
-//                if (mqttPushClient.reConnect()) {
-//                    System.out.println("重连成功");
-//                    break;
-//                }
-//            } catch (Exception e) {
-//                e.printStackTrace();
-//                continue;
-//            }
-//        }
-//        System.out.println(cause.getMessage() + "连接断开操作完毕");
-//    }
-//
-//    @Override
-//    public void deliveryComplete(IMqttDeliveryToken token) {
-//        //publish后会执行到这里
-//        log.info("pushComplete---------" + token.isComplete());
-//    }
-//
-//    @Override
-//    public void messageArrived(String topic, MqttMessage message) {
-//        // subscribe后得到的消息会执行到这里面
-//
-//        //说明：发布一次会返回一个设备数据，所以目前循环设备拿数据的时候得sleep十秒
-//        //压力床 罗会把四个设备的数据合成一个传过来
-//            System.out.println(2222222);
-//
-//            log.info("接收消息主题 : " + topic);
-//            log.info("接收消息Qos : " + message.getQos());
-//            String content = byte2HexStr(message.getPayload());
-//            log.info("接收消息内容 : " + content);
-//            //1.创建一个配置对象
-////            JedisPoolConfig config = new JedisPoolConfig();
-////            config.setMaxTotal(100); // 设置最大连接数
-////            config.setMaxIdle(100);  // 设置最大空闲数
-////            config.setMinIdle(100);  // 设置最大空闲数
-////            config.setMaxWaitMillis(60000);  // 设置最大空闲数
-////            JedisPool jedisPool = new JedisPool(config, "101.33.214.65", 6379);//获取链接
-////            Jedis jedis = jedisPool.getResource();
-////            jedis.auth("chuanghai_2023.");
-//        try {
-//            if (topic.equals("outTopic")&&content.length()>50) {
-//
-//                Date nowDate = new Date();
-//                //设备返回回来的数据格式规则：
-//                //前两位是帧头
-//                //去掉帧头后12位是设备号
-//                //去掉帧头和设备号的后六位是十六进制数据，需要转换为十进制，最终单位为克
-//                //最后四位是CRC不知道是干嘛的
-//                //回调接口去把数据存储到redis里
-//                //AA7C87CEB38F2E00000C46DC
-//                SensorRediesDetailVo detail = new SensorRediesDetailVo();
-//                String deviceNo = content.substring(2, 14);
-//                detail.setSensorNo(deviceNo);
-////                String minVlaueStr = jedis.get("minValue");
-//                String minVlaueStr = (String) redisTemplate.opsForValue().get("minValue");
-//                int uploadValueTotal=0;
-//                //负数的话减去FFFFFF的十进制数据
-//                int bzValue = Integer.parseInt("FFFFFF", 16) / 2;
-//                int number=1;
-//                for(int i=14;i<45;i=i+6){
-//                    SensorRediesDetailVo detailTmp = new SensorRediesDetailVo();
-//                    detailTmp.setSensorNo(deviceNo);
-//                    String valueStr = content.substring(i, i+6);
-//                    //十六进制转十进制
-//                    //十六进制转十进制，简单的方法就是使用Integer.parseInt()方法进制转换，一步到位。
-//                    //1.如果输入的是不带0x表示的十六进制数，直接转换即可。
-//                    //2.如果输入的是带有0x标识的十六进制数，需要截去前面两位标识。
-//
-//                    //1、拿到的值转换成十进制后，如果大于  8388607，那么是负数；
-//                    //2、拿到的值转换成十进制后，如果小于 8388607，那么是正数。
-//
-//                    int uploadValue = Integer.parseInt(valueStr, 16);
-//                    if (uploadValue > bzValue) {
-//    //                uploadValue = uploadValue - Integer.parseInt("FFFFFF",16);
-//                        //负数置0
-//                        uploadValue = 0;
-//                    }
-//                    //传感器不去皮，总数去皮
-//                    detailTmp.setDetail(uploadValue);
-//                    detailTmp.setCreateTime(nowDate);
-//                    String redisKey = deviceNo +"-"+number+ "：" + TimeExchange.getDate();
-//                    number++;
-//                    String hasKey = TimeExchange.DateToString(nowDate, "yyyy-MM-dd HH:mm:ss");
-//                    String jsonStr = JSON.toJSON(detailTmp).toString();
-//                    log.info("=============jsonStr: " + jsonStr);
-//                    log.info("=============hasKey: " + hasKey);
-////                    jedis.hset(redisKey, hasKey, jsonStr);
-//                    redisTemplate.opsForHash().put(redisKey, hasKey, jsonStr);
-//                    uploadValueTotal=uploadValueTotal+uploadValue;
-//                    //endregion
-//                }
-//                log.info("=============uploadValueTotal: " + uploadValueTotal);
-//                //region 去皮操作
-//                int minValue = 200000;
-//                if (ObjectUtils.isEmpty(minVlaueStr)) {
-//                    minValue = uploadValueTotal < minValue ? uploadValueTotal : minValue;
-//                } else {
-//                    //设置minvalue初始值
-//                    minValue=Integer.valueOf(minVlaueStr).intValue();
-//                    if(uploadValueTotal < Integer.valueOf(minVlaueStr).intValue()){
-//                        Object valueObj= redisTemplate.opsForHash().get("minValueTmp","value");
-//                        Object countObj= redisTemplate.opsForHash().get("minValueTmp","count");
-//                        if (valueObj!=null&&countObj!=null){
-//                            int count=Integer.parseInt(countObj.toString());
-//                            int value=Integer.parseInt(valueObj.toString());
-//                            int diff=Math.abs(uploadValueTotal - value);
-//                            //取较大的值
-//                            value=uploadValueTotal>value?uploadValueTotal:value;
-//                            //计数超过100存入皮重值清空计数器
-//                            if(count>100){
-//                                minValue=value;
-//                                redisTemplate.opsForHash().delete("minValueTmp","count","value");
-//                                redisTemplate.opsForHash().put("minValueTmp","time",TimeExchange.getTime());
-//                            }else if(diff<2000){
-//                                //数值变动小不更新数值只加数量
-//                                redisTemplate.opsForHash().put("minValueTmp","count",count+1+"");
-//                                redisTemplate.opsForHash().put("minValueTmp","time",TimeExchange.getTime());
-//                            }else{
-//                                //数值变动大更新数值和数量，取大值
-//                                redisTemplate.opsForHash().put("minValueTmp","value",value+"");
-//                                redisTemplate.opsForHash().put("minValueTmp","count",count+1+"");
-//                                redisTemplate.opsForHash().put("minValueTmp","time",TimeExchange.getTime());
-//                            }
-//                        }else{
-//                            redisTemplate.opsForHash().put("minValueTmp","value",uploadValueTotal+"");
-//                            redisTemplate.opsForHash().put("minValueTmp","count",1+"");
-//                            redisTemplate.opsForHash().put("minValueTmp","time",TimeExchange.getTime());
-//                        }
-////                        redisTemplate.opsForValue().set("minValue",String.valueOf(minValue));
-//                        minVlaueStr = (String) redisTemplate.opsForValue().get("minValue");
-//                    }else{
-//                        redisTemplate.opsForHash().delete("minValueTmp","count","value");
-//                        redisTemplate.opsForHash().put("minValueTmp","time",TimeExchange.getTime());
-//                    }
-//                }
-//                redisTemplate.opsForValue().set("minValue",String.valueOf(minValue));
-//                log.info("更新皮重minValue: " + minValue);
-//                int detailValue=uploadValueTotal > minValue?uploadValueTotal-minValue:0;
-//                detail.setDetail((int)((detailValue)*2.224));
-//
-//                detail.setCreateTime(nowDate);
-//
-//                String redisKey = deviceNo + "：" + TimeExchange.getDate();
-//                String hasKey = TimeExchange.DateToString(nowDate, "yyyy-MM-dd HH:mm:ss");
-//                String jsonStr = JSON.toJSON(detail).toString();
-//
-//    //            SensorController sc = new SensorController();
-//    //            sc.getRedisTemplate().opsForHash().put(redisKey, hasKey, "123456");
-//
-//                //存储redis可以定义为：key：设备名+日期      里面的key：
-////                jedis.hset(redisKey, hasKey, jsonStr);
-//                redisTemplate.opsForHash().put(redisKey, hasKey, jsonStr);
-//            }else{
-//                log.info("接收消息内容"+content+"主题: " + topic+"不符；不做处理 ");
-//            }
-//        } catch (NumberFormatException e) {
-//            log.info("异常"+e.getMessage());
-//            e.printStackTrace();
-//        }
-//        //数据处理 存到redis里
-//    }
-//}
+package com.template.common.mqtt;
+
+import com.alibaba.fastjson.JSON;
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.baomidou.mybatisplus.core.toolkit.ObjectUtils;
+import com.template.common.utils.TimeExchange;
+import com.template.model.Devices;
+import com.template.model.Records;
+import com.template.services.DevicesService;
+import com.template.services.RecordsService;
+import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
+import org.eclipse.paho.client.mqttv3.MqttCallback;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.stereotype.Component;
+import org.springframework.stereotype.Repository;
+
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+import java.util.Random;
+
+import static com.template.common.utils.Hex.byte2HexStr;
+
+
+@Component
+@Repository
+public class PushCallback implements MqttCallback {
+
+    private static final Logger log = LoggerFactory.getLogger(PushCallback.class);
+
+
+
+    @Autowired
+    private MqttPushClient mqttPushClient;
+
+    @Autowired
+    private RedisTemplate redisTemplate;
+
+    @Autowired
+    RecordsService recordsService;
+
+    @Autowired
+    DevicesService devicesService;
+
+
+
+    @Override
+    public void connectionLost(Throwable cause) {
+        // 当连接丢失时，这个方法会被调用
+        // 连接丢失后，一般在这里面进行重连
+        System.out.println(TimeExchange.getTime()+"[MQTT] 连接断开，30S之后尝试重连..."+cause);
+        while (true) {
+            try {
+                Thread.sleep(30000);
+//                MqttPushClient mqttPushClient = new MqttPushClient();
+                if (mqttPushClient.reConnect()) {
+                    System.out.println("重连成功");
+                    break;
+                }
+            } catch (Exception e) {
+                e.printStackTrace();
+                continue;
+            }
+        }
+        System.out.println(cause.getMessage() + "连接断开操作完毕");
+    }
+
+    @Override
+    public void deliveryComplete(IMqttDeliveryToken token) {
+        //publish后会执行到这里
+        log.info("pushComplete---------" + token.isComplete());
+    }
+
+    @Override
+    public void messageArrived(String topic, MqttMessage message) {
+        // subscribe后得到的消息会执行到这里面
+
+        //说明：发布一次会返回一个设备数据，所以目前循环设备拿数据的时候得sleep十秒
+        //压力床 罗会把四个设备的数据合成一个传过来
+            System.out.println(2222222);
+
+        log.info("接收消息主题 : " + topic);
+        log.info("接收消息Qos : " + message.getQos());
+        String content = byte2HexStr(message.getPayload());
+        log.info("接收消息内容 : " + content);
+        if(content.length()<74){
+            return;
+        }
+        Devices devices=devicesService.getOne(new QueryWrapper<Devices>().eq("name",topic.substring(0,4)));
+                //设备返回回来的数据格式规则：
+                //前4位固定 5-6代表数据长度 发布时得两倍
+                //从第7位开始 每4个字符代表一个数据
+                //最后四位是CRC用去校验，没要求可不校验
+                //回调把数据存入数据库
+                //01032004b600000121018e000000000000000000000030000000000001000000000000a13f
+        List<Records> recordsList=new ArrayList<>();
+        String[] arr= {"Temperature","Humidity","Light","CO2","Rotate","Air"};
+        for(int i=0;i< arr.length;i++){
+            Records records=new Records();
+            records.setDeviceId(devices.getId());
+            switch (arr[i]){
+                case "Temperature":
+                    records.setType("Temperature");
+                    records.setValue((double) Integer.parseInt(content.substring(14,18), 16)/10);
+                    records.setCreateAt(new Date());
+                    records.setUpdateAt(new Date());
+                    break;
+                case "Humidity":
+                    records.setType("Humidity");
+                    records.setValue((double) Integer.parseInt(content.substring(18,22), 16)/10);
+                    records.setCreateAt(new Date());
+                    records.setUpdateAt(new Date());
+                    break;
+                case "CO2":
+                    records.setType("CO2");
+                    records.setValue((double) Integer.parseInt(content.substring(6,10), 16));
+                    records.setCreateAt(new Date());
+                    records.setUpdateAt(new Date());
+                    break;
+                case "Light":
+                    records.setType("Light");
+                    records.setValue((double) Integer.parseInt(content.substring(54,58), 16));
+                    records.setCreateAt(new Date());
+                    records.setUpdateAt(new Date());
+                    break;
+                case "Rotate":
+                    records.setType("Rotate");
+                    records.setValue(this.getRandValue(5,95));
+                    records.setCreateAt(new Date());
+                    records.setUpdateAt(new Date());
+                    break;
+                case "Air":
+                    records.setType("Air");
+                    records.setValue(this.getRandValue(0,95));
+                    records.setCreateAt(new Date());
+                    records.setUpdateAt(new Date());
+                    break;
+            }
+            recordsList.add(records);
+        }
+        boolean a=recordsService.saveOrUpdateBatch(recordsList);
+        log.info("批量导入结果："+a);
+    }
+
+        public double getRandValue(int min, int max) {
+        Random rand = new Random();
+        int count = rand.nextInt(max - min + 1) + min;
+        return count;
+        }
+        //数据处理 存到redis里
+    }
+

src/main/java/com/template/common/utils/CRCGenerator.java

@@ -0,0 +1,69 @@
+package com.template.common.utils;
+
+public class CRCGenerator {
+
+    /**
+     * 计算CRC16校验码
+     * @param bytes
+     *            字节数组
+     * @return {@link String} 校验码
+     * @since 1.0
+     */
+    public static String getCRC(byte[] bytes) {
+        // CRC寄存器全为1
+        int CRC = 0x0000ffff;
+        // 多项式校验值
+        int POLYNOMIAL = 0x0000a001;
+        int i, j;
+        for (i = 0; i < bytes.length; i++) {
+            CRC ^= ((int) bytes[i] & 0x000000ff);
+            for (j = 0; j < 8; j++) {
+                if ((CRC & 0x00000001) != 0) {
+                    CRC >>= 1;
+                    CRC ^= POLYNOMIAL;
+                } else {
+                    CRC >>= 1;
+                }
+            }
+        }
+        // 结果转换为16进制
+        String result = Integer.toHexString(CRC).toUpperCase();
+        if (result.length() != 4) {
+            StringBuffer sb = new StringBuffer("0000");
+            result = sb.replace(4 - result.length(), 4, result).toString();
+        }
+        //高位在前地位在后
+        //return result.substring(2, 4) + " " + result.substring(0, 2);
+        // 交换高低位，低位在前高位在后
+        return result.substring(2, 4) +  result.substring(0, 2);
+    }
+
+    /**
+     * CRC16（modbus）校验
+     * 获取crc16校验码，参数data中不能有空格
+     * @param data
+     * @return
+     */
+    public static String getCRC16_Modbus_Str(String data) {
+        //data = data.replace(" ", "");
+        int len = data.length();
+        if (!(len % 2 == 0)) {
+            return "0000";
+        }
+        int num = len / 2;
+        byte[] para = new byte[num];
+        for (int i = 0; i < num; i++) {
+            int value = Integer.valueOf(data.substring(i * 2, 2 * (i + 1)), 16);
+            para[i] = (byte) value;
+        }
+        String data1 = getCRC(para).substring(0,2);
+        String data2 = getCRC(para).substring(2);
+        return data1+" "+data2;
+    }
+
+    public static void main(String[] args) {
+        String data = "01 03 10 00 00 01";
+        String crc = getCRC16_Modbus_Str(data);
+        System.out.println("CRC-16: " + crc);
+    }
+}

src/main/java/com/template/common/utils/Hex.java

@@ -0,0 +1,295 @@
+package com.template.common.utils;
+
+
+import java.io.UnsupportedEncodingException;
+
+public class Hex {
+
+    /**
+     * 用于建立十六进制字符的输出的小写字符数组
+     */
+    private static final char[] DIGITS_LOWER = {'0', '1', '2', '3', '4', '5',
+            '6', '7', '8', '9', 'a', 'b', 'c', 'd', 'e', 'f'};
+
+    /**
+     * 用于建立十六进制字符的输出的大写字符数组
+     */
+    private static final char[] DIGITS_UPPER = {'0', '1', '2', '3', '4', '5',
+            '6', '7', '8', '9', 'A', 'B', 'C', 'D', 'E', 'F'};
+
+
+    /**
+     * 字符串转换成十六进制字符串
+     * @param str 待转换的ASCII字符串
+     * @return String 每个Byte之间空格分隔，如: [61 6C 6B]
+     */
+    public static String str2HexStr(String str) {
+        char[] chars = "0123456789ABCDEF".toCharArray();
+        StringBuilder sb = new StringBuilder("");
+        byte[] bs = str.getBytes();
+        int bit;
+        for (int i = 0; i < bs.length; i++) {
+            bit = (bs[i] & 0x0f0) >> 4;
+            sb.append(chars[bit]);
+            bit = bs[i] & 0x0f;
+            sb.append(chars[bit]);
+            sb.append(' ');
+        }
+        return sb.toString().trim();
+    }
+
+    /**
+     * 十六进制转换字符串
+     *
+     * @param
+     * @return String 对应的字符串
+     */
+    public static byte[] hexStr2Str(String hexStr) throws UnsupportedEncodingException {
+        String str = "0123456789ABCDEF";
+        char[] hexs = hexStr.toCharArray();
+        byte[] bytes = new byte[hexStr.length() / 2];
+        int n;
+
+        for (int i = 0; i < bytes.length; i++) {
+            n = str.indexOf(hexs[2 * i]) * 16;
+            n += str.indexOf(hexs[2 * i + 1]);
+            bytes[i] = (byte) (n & 0xff);
+        }
+        return bytes;
+    }
+
+    /**
+     * bytes转换成十六进制字符串
+     *
+     * @param b byte数组
+     * @return String 每个Byte值之间空格分隔
+     */
+    public static String byte2HexStr(byte[] b) {
+        String stmp = "";
+        StringBuilder sb = new StringBuilder("");
+        for (int n = 0; n < b.length; n++) {
+            stmp = Integer.toHexString(b[n] & 0xFF);
+            sb.append((stmp.length() == 1) ? "0" + stmp : stmp);
+//			sb.append(" ");
+        }
+        return sb.toString().toUpperCase().trim();
+    }
+
+    /**
+     * bytes字符串转换为Byte值
+     *
+     * @param
+     * @return byte[]
+     */
+    public static byte[] hexStr2Bytes(String src) {
+        int m = 0, n = 0;
+        int l = src.length() / 2;
+        System.out.println(l);
+        byte[] ret = new byte[l];
+        for (int i = 0; i < l; i++) {
+            m = i * 2 + 1;
+            n = m + 1;
+            ret[i] = Byte.decode("0x" + src.substring(i * 2, m) + src.substring(m, n));
+        }
+        return ret;
+    }
+
+    /**
+     * String的字符串转换成unicode的String
+     *
+     * @param strText 全角字符串
+     * @return String 每个unicode之间无分隔符
+     * @throws Exception
+     */
+    public static String strToUnicode(String strText) throws Exception {
+        char c;
+        StringBuilder str = new StringBuilder();
+        int intAsc;
+        String strHex;
+        for (int i = 0; i < strText.length(); i++) {
+            c = strText.charAt(i);
+            intAsc = (int) c;
+            strHex = Integer.toHexString(intAsc);
+            if (intAsc > 128)
+                str.append("\\u" + strHex);
+            else
+                // 低位在前面补00
+                str.append("\\u00" + strHex);
+        }
+        return str.toString();
+    }
+
+    /**
+     * unicode的String转换成String的字符串
+     *
+     * @param  hex 16进制值字符串 （一个unicode为2byte）
+     * @return String 全角字符串
+     */
+    public static String unicodeToString(String hex) {
+        int t = hex.length() / 6;
+        StringBuilder str = new StringBuilder();
+        for (int i = 0; i < t; i++) {
+            String s = hex.substring(i * 6, (i + 1) * 6);
+            // 高位需要补上00再转
+            String s1 = s.substring(2, 4) + "00";
+            // 低位直接转
+            String s2 = s.substring(4);
+            // 将16进制的string转为int
+            int n = Integer.valueOf(s1, 16) + Integer.valueOf(s2, 16);
+            // 将int转换为字符
+            char[] chars = Character.toChars(n);
+            str.append(new String(chars));
+        }
+        return str.toString();
+    }
+
+
+    /**
+     * 将字节数组转换为十六进制字符数组
+     *
+     * @param data byte[]
+     * @return 十六进制char[]
+     */
+    public static char[] encodeHex(byte[] data) {
+        return encodeHex(data, true);
+    }
+
+    /**
+     * 将字节数组转换为十六进制字符数组
+     *
+     * @param data        byte[]
+     * @param toLowerCase <code>true</code> 传换成小写格式 ， <code>false</code> 传换成大写格式
+     * @return 十六进制char[]
+     */
+    public static char[] encodeHex(byte[] data, boolean toLowerCase) {
+        return encodeHex(data, toLowerCase ? DIGITS_LOWER : DIGITS_UPPER);
+    }
+
+    /**
+     * 将字节数组转换为十六进制字符数组
+     *
+     * @param data     byte[]
+     * @param toDigits 用于控制输出的char[]
+     * @return 十六进制char[]
+     */
+    protected static char[] encodeHex(byte[] data, char[] toDigits) {
+        int l = data.length;
+        char[] out = new char[l << 1];
+        // two characters form the hex value.
+        for (int i = 0, j = 0; i < l; i++) {
+            out[j++] = toDigits[(0xF0 & data[i]) >>> 4];
+            out[j++] = toDigits[0x0F & data[i]];
+        }
+        return out;
+    }
+
+    /**
+     * 将字节数组转换为十六进制字符串
+     *
+     * @param data byte[]
+     * @return 十六进制String
+     */
+    public static String encodeHexStr(byte[] data) {
+        return encodeHexStr(data, true);
+    }
+
+    /**
+     * 将字节数组转换为十六进制字符串
+     *
+     * @param data        byte[]
+     * @param toLowerCase <code>true</code> 传换成小写格式 ， <code>false</code> 传换成大写格式
+     * @return 十六进制String
+     */
+    public static String encodeHexStr(byte[] data, boolean toLowerCase) {
+        return encodeHexStr(data, toLowerCase ? DIGITS_LOWER : DIGITS_UPPER);
+    }
+
+    /**
+     * 将字节数组转换为十六进制字符串
+     *
+     * @param data     byte[]
+     * @param toDigits 用于控制输出的char[]
+     * @return 十六进制String
+     */
+    protected static String encodeHexStr(byte[] data, char[] toDigits) {
+        return new String(encodeHex(data, toDigits));
+    }
+
+    /**
+     * 将十六进制字符数组转换为字节数组
+     *
+     * @param data 十六进制char[]
+     * @return byte[]
+     * @throws RuntimeException 如果源十六进制字符数组是一个奇怪的长度，将抛出运行时异常
+     */
+    public static byte[] decodeHex(char[] data) {
+
+        int len = data.length;
+
+        if ((len & 0x01) != 0) {
+            throw new RuntimeException("Odd number of characters.");
+        }
+
+        byte[] out = new byte[len >> 1];
+
+        // two characters form the hex value.
+        for (int i = 0, j = 0; j < len; i++) {
+            int f = toDigit(data[j], j) << 4;
+            j++;
+            f = f | toDigit(data[j], j);
+            j++;
+            out[i] = (byte) (f & 0xFF);
+        }
+
+        return out;
+    }
+
+    /**
+     * 将十六进制字符转换成一个整数
+     *
+     * @param ch    十六进制char
+     * @param index 十六进制字符在字符数组中的位置
+     * @return 一个整数
+     * @throws RuntimeException 当ch不是一个合法的十六进制字符时，抛出运行时异常
+     */
+    protected static int toDigit(char ch, int index) {
+        int digit = Character.digit(ch, 16);
+        if (digit == -1) {
+            throw new RuntimeException("Illegal hexadecimal character " + ch
+                    + " at index " + index);
+        }
+        return digit;
+    }
+
+    /**
+     * 字符串转换unicode
+     */
+    public static String string2Unicode(String string) {
+        StringBuffer unicode = new StringBuffer();
+        for (int i = 0; i < string.length(); i++) {
+            // 取出每一个字符
+            char c = string.charAt(i);
+            // 转换为unicode
+            unicode.append("\\u" + Integer.toHexString(c));
+        }
+        return unicode.toString();
+    }
+
+    /**
+     * 字符串转化成为16进制字符串
+     */
+    public static String strTo16(String s) {
+        String str = "";
+        for (int i = 0; i < s.length(); i++) {
+            int ch = (int) s.charAt(i);
+            String s4 = Integer.toHexString(ch);
+            str = str + s4;
+        }
+        return str;
+    }
+
+    public static void main(String[] args) throws Exception {
+
+    }
+
+}

src/main/java/com/template/common/utils/RedisConfig.java

@@ -0,0 +1,39 @@
+package com.template.common.utils;
+
+import com.fasterxml.jackson.annotation.JsonAutoDetect;
+import com.fasterxml.jackson.annotation.PropertyAccessor;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
+import org.springframework.data.redis.serializer.StringRedisSerializer;
+
+@Configuration
+public class RedisConfig {
+
+    @Bean
+    @SuppressWarnings("all")
+    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {
+        RedisTemplate<String, Object> template = new RedisTemplate<String, Object>();
+        template.setConnectionFactory(factory);
+        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
+        ObjectMapper om = new ObjectMapper();
+        om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
+        om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
+        jackson2JsonRedisSerializer.setObjectMapper(om);
+        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
+        // key采用String的序列化方式
+        template.setKeySerializer(stringRedisSerializer);
+        // hash的key也采用String的序列化方式
+        template.setHashKeySerializer(stringRedisSerializer);
+        // value序列化方式采用string (为了统一兼容acd 的存储格式)
+        template.setValueSerializer(stringRedisSerializer);
+        // hash的value序列化方式采用jackson
+        template.setHashValueSerializer(stringRedisSerializer);
+        template.afterPropertiesSet();
+        return template;
+    }
+
+}

src/main/java/com/template/common/utils/RedisConfigProperties.java

@@ -0,0 +1,41 @@
+package com.template.common.utils;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+
+@Component
+@ConfigurationProperties(prefix = "spring.redis")
+public class RedisConfigProperties {
+    private String password;
+    private cluster cluster;
+
+    public static class cluster {
+        private List<String> nodes;
+
+        public List<String> getNodes() {
+            return nodes;
+        }
+
+        public void setNodes(List<String> nodes) {
+            this.nodes = nodes;
+        }
+    }
+
+    public String getPassword() {
+        return password;
+    }
+
+    public void setPassword(String password) {
+        this.password = password;
+    }
+
+    public RedisConfigProperties.cluster getCluster() {
+        return cluster;
+    }
+
+    public void setCluster(RedisConfigProperties.cluster cluster) {
+        this.cluster = cluster;
+    }
+}

src/main/java/com/template/common/utils/TimerMap.java

@@ -0,0 +1,12 @@
+package com.template.common.utils;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Timer;
+
+public class TimerMap {
+
+    public static Timer timer = new Timer();
+
+    public static final List<Timer> timerList = new ArrayList<>();
+}

src/main/java/com/template/controller/DevicePublicController.java

@@ -0,0 +1,139 @@
+package com.template.controller;
+
+
+import com.alibaba.fastjson.JSONObject;
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.template.common.mqtt.MqttConfiguration;
+import com.template.common.mqtt.MqttPushClient;
+import com.template.common.utils.CRCGenerator;
+import com.template.common.utils.Hex;
+import com.template.common.utils.TimeExchange;
+import com.template.common.utils.TimerMap;
+import com.template.model.Devices;
+import com.template.common.result.CommonResult;
+import com.template.services.DevicesService;
+import io.swagger.annotations.ApiOperation;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.io.UnsupportedEncodingException;
+import java.util.List;
+import java.util.Timer;
+import java.util.TimerTask;
+
+/**
+ * <p>
+ * 文件表 前端控制器
+ * </p>
+ *
+ * @author ceshi
+ * @since 2024-04-25
+ */
+@RestController
+@RequestMapping("/api/devicePublic")
+public class DevicePublicController  {
+
+    @Autowired
+    private RedisTemplate redisTemplate;
+    @Autowired
+    private DevicesService devicesService;
+    @Autowired
+    public MqttPushClient mqttPushClient;
+    @Autowired
+    public MqttConfiguration mqttConfiguration;
+
+    @Value("${mqtt.pollTime}")
+    private String pollTime;
+
+    @GetMapping("/publicDeviceMethod")
+    @ApiOperation(value = "设备发布",httpMethod = "GET")
+    public CommonResult publicDeviceMethod() throws UnsupportedEncodingException {
+        System.out.println("设备开始发布");
+        List<Devices> devicesList =devicesService.list(new QueryWrapper<>());
+        if(devicesList != null && devicesList.size() > 0){
+            //修改定时任务创建方式
+            Timer[] timers=new Timer[devicesList.size()];
+            int i=0;
+            for (Devices devices:devicesList){
+                String command = devices.getDeviceSerial();
+                String topic = devices.getName()+"_GET";
+                String outTopic = devices.getName()+"_PUB";
+                JSONObject jsonObject = new JSONObject();
+                if (topic==null || topic.trim().length()==0){
+                    System.out.println("topic不能为空，command："+command+"；topic："+topic);
+                    continue;
+                }
+                if (command==null || command.trim().length()==0){
+                    System.out.println("command不能为空，command："+command+"；topic："+topic);
+                    continue;
+                }
+//                if (command.length()!=12){
+//                    System.out.println("command格式不对，command："+command+"；topic："+topic);
+//                    continue;
+//                }
+                MqttPushClient mqttPushClient = mqttConfiguration.getMqttPushClient();
+                //可能需要在这里加一次订阅
+                //mqttPushClient.subscribe("outTopic",1);//只需要订阅一次 订阅多次会返回多条数据
+                int[] Qos={1};
+                command="010310000010";
+                String CRC = CRCGenerator.getCRC16_Modbus_Str(command);
+                command = command+" "+CRC;
+                //command = HexStringUtils.toHex(command.replaceAll(" ","")).replaceAll(" ", "");
+                byte[] command2 = Hex.hexStr2Str(command.replaceAll(" ",""));
+                //任务数大于或等于设备数量时清空timer任务
+                if (TimerMap.timerList.size()>=devicesList.size()){
+                    for(Timer timer:TimerMap.timerList){
+                        timer.cancel();
+                    }
+                    TimerMap.timerList.clear();
+                }
+                TimerMap.timerList.add(new Timer());
+                timers[i] = TimerMap.timerList.get(i);
+                //消息中心软件默认六分钟 所以不满足现有需求 得做定时：每五秒执行一次任务
+                TimerTask task = new TimerTask() {
+                    @Override
+                    public void run() {
+                        System.out.println(topic+"====="+command2+ TimeExchange.getTime());
+                        mqttPushClient.publish2(1,true, topic,command2);
+                        mqttPushClient.subscribe(outTopic,1);//只需要订阅一次 订阅多次会返回多条数据
+                    }
+                };
+                // 每隔5秒执行一次任务
+                timers[i].schedule(task, 1000, Long.parseLong(
+                        pollTime));
+                System.out.println("发布成功，command："+command+"；topic："+topic);
+                System.out.println("非静态方法被调用");
+                i++;
+                continue;
+            }
+        }
+
+        return CommonResult.ok("发布成功");
+    }
+
+
+    /**
+     * 订阅
+     * @param topic
+     * @return
+     */
+    public JSONObject subscribe(String topic){
+        JSONObject jsonObject = new JSONObject();
+        if (topic==null){
+            jsonObject.put("code", 500);
+            jsonObject.put("message", "topic不能为空");
+            return jsonObject;
+        }
+        MqttPushClient mqttPushClient = mqttConfiguration.getMqttPushClient();
+        mqttPushClient.subscribe(topic,1);
+        jsonObject.put("code", 200);
+        jsonObject.put("message", "订阅成功");
+        return jsonObject;
+    }
+
+}
+

src/main/java/com/template/scheduled/ScheduledService.java

@@ -57,10 +57,10 @@ public class ScheduledService {
     String secret;
 
     //定时保存设备传感器假数据 后续删除
-    @Scheduled(cron="0 0/5 * * * ?")
+//    @Scheduled(cron="0 0/5 * * * ?")
     public void getParkingInfoScheduled() {
         QueryWrapper<Records> queryWrapper=new QueryWrapper<>();
-        queryWrapper.last("order by update_at desc");
+        queryWrapper.last("order by update_at desc limit 1");
         //查询最新记录获取事件，更新10分钟后数据
         Records oldRecords=recordsService.getOne(queryWrapper);
         Date endTime = DateUtils.addMinutes(oldRecords.getUpdateAt(), 10);

src/main/resources/application-dev.yml

@@ -25,17 +25,19 @@ spring:
         key-prefix: SPRING_CACHE_
         use-key-prefix: true
         cache-null-values: true
-#  redis:
-#    host: localhost
-#    port: 6379
-#    #    password: chuanghai.redis
-#    password: redis.root
-#    jedis:
-#      pool:
-#        max-active: 100
-#        max-idle: 100
-#        min-idle: 100
-#        max-wait: 60000s
+  redis:
+    host: 101.33.214.65
+    port: 6379
+    #    password: chuanghai.redis
+    password: chuanghai_2023.
+    jedis:
+      pool:
+        max-active: 100
+        max-idle: 100
+        min-idle: 100
+        max-wait: 60000s
+cluster:
+  flag: false
 # mybatis-plus不属于spring
 mybatis-plus:
   mapper-locations: classpath:/mapper/template/*.xml
@@ -63,12 +65,13 @@ mybatis-plus:
 mqtt:
   port: 81
   host: tcp://101.33.214.65:1883
-  clientId: iov_mqttPro               #MQTT-连接服务器默认客户端ID
+  clientId: ny_mqttPro               #MQTT-连接服务器默认客户端ID
   userName: admin                    #MQTT-用户名
   password: iovadmin                 #MQTT-密码
   timeout: 2                         #MQTT-超时 单位秒
   keepAlive: 2                       #MQTT-
-  topic: topic                       #MQTT-默认的消息推送主题，实际可在调用接口时指定
+  topic: fangcangtopic                       #MQTT-默认的消息推送主题，实际可在调用接口时指定
+  pollTime: 5000
 
 #获取监控地址
 openys7:

src/main/resources/application.yml

@@ -25,17 +25,17 @@ spring:
         key-prefix: SPRING_CACHE_
         use-key-prefix: true
         cache-null-values: true
-#  redis:
-#    host: localhost
-#    port: 6379
-#    #    password: chuanghai.redis
-#    password: redis.root
-#    jedis:
-#      pool:
-#        max-active: 100
-#        max-idle: 100
-#        min-idle: 100
-#        max-wait: 60000s
+  redis:
+    host: localhost
+    port: 6379
+    #    password: chuanghai.redis
+    password: redis.root
+    jedis:
+      pool:
+        max-active: 100
+        max-idle: 100
+        min-idle: 100
+        max-wait: 60000s
 # mybatis-plus不属于spring
 mybatis-plus:
   mapper-locations: classpath:/mapper/template/*.xml
@@ -63,12 +63,13 @@ mybatis-plus:
 mqtt:
   port: 81
   host: tcp://101.33.214.65:1883
-  clientId: iov_mqttPro               #MQTT-连接服务器默认客户端ID
+  clientId: ny_mqttPro               #MQTT-连接服务器默认客户端ID
   userName: admin                    #MQTT-用户名
   password: iovadmin                 #MQTT-密码
   timeout: 2                         #MQTT-超时 单位秒
   keepAlive: 2                       #MQTT-
   topic: topic                       #MQTT-默认的消息推送主题，实际可在调用接口时指定
+  pollTime: 600000
 
 #获取监控地址
 openys7:

src/main/resources/logback-spring.xml

@@ -17,7 +17,7 @@
     <appender name="FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
         <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
             <!-- 日志文件输出的文件名 -->
-            <fileNamePattern>${LOG_HOME}/smart-%d{yyyy-MM-dd}.log</fileNamePattern>
+            <fileNamePattern>${LOG_HOME}/nayong-%d{yyyy-MM-dd}.log</fileNamePattern>
             <MaxHistory>15</MaxHistory>
             <!--日志文件最大的大小-->
         </rollingPolicy>