esp8266和dht11温湿度传感器 数据采集mqtt入库完成

src/main/java/com/rehome/mqttclienttemperature/AdminClientTemperatureApplication.java

@@ -1,6 +1,7 @@
 package com.rehome.mqttclienttemperature;
 
 
+import com.rehome.mqttclienttemperature.service.TemperatureEspService;
 import com.rehome.mqttclienttemperature.service.TemperatureService;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.BeansException;
@@ -34,6 +35,13 @@ public class AdminClientTemperatureApplication implements CommandLineRunner, App
     @Resource
     private TemperatureService temperatureService;
 
+    /**
+     后台接口自动导入
+     */
+    @Resource
+    private TemperatureEspService temperatureEspService;
+
+
     public static void main(String[] args) {
         SpringApplication.run(AdminClientTemperatureApplication.class, args);
     }
@@ -51,13 +59,21 @@ public class AdminClientTemperatureApplication implements CommandLineRunner, App
             log.info(entry.getKey());//demo1Controller
         }
         try {
-            if (temperatureService != null) {
+//            if (temperatureService != null) {
+//                log.info("------------------------");
+//                log.info("TemperatureController is not empty");
+//                MqttRSAClient client = new MqttRSAClient();
+//                client.start(temperatureService);
+//            }else {
+//                log.info("temperatureService is empty");
+//            }
+            if (temperatureEspService != null) {
                 log.info("------------------------");
-                log.info("TemperatureController is not empty");
+                log.info("temperatureEspService is not empty");
-                MqttRSAClient client = new MqttRSAClient();
+                MqttDianDengTechClient client = new MqttDianDengTechClient();
-                client.start(temperatureService);
+                client.start(temperatureEspService);
             }else {
-                log.info("temperatureService is empty");
+                log.info("temperatureEspService is empty");
             }
         } catch (Exception ex) {
             ex.printStackTrace();

src/main/java/com/rehome/mqttclienttemperature/MqttDianDengTechClient.java

@@ -0,0 +1,159 @@
+package com.rehome.mqttclienttemperature;
+
+
+
+import com.rehome.mqttclienttemperature.service.TemperatureEspService;
+import com.rehome.mqttclienttemperature.service.TemperatureService;
+import com.rehome.mqttclienttemperature.utils.MqttSSLSocketFactory;
+import com.rehome.mqttclienttemperature.utils.UUIDUtil;
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.paho.client.mqttv3.*;
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
+import javax.net.ssl.SSLSocketFactory;
+import java.io.InputStream;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.Timer;
+import java.util.TimerTask;
+
+
+@Slf4j
+public class MqttDianDengTechClient {
+    /**
+     * 代理服务器ip地址
+     */
+    public static final String MQTT_BROKER_HOST = "ssl://47.242.184.139:8883";
+
+    /**
+     * 客户端唯一标识
+     */
+    public static String MQTT_CLIENT_ID = "AppServer_temperature_APP_server_02";
+
+    /**
+     *帐号
+     */
+    public static String USERNAME = "admin";
+    /**
+     * 密码
+     */
+    public static String PASSWORD = "publish452131wW452131wW$";
+    /**
+     * 订阅标识
+     */
+    public static String TOPIC_FILTER = "/device/esp8266/001";
+
+    private volatile static MqttClient mqttClient;
+    private static MqttConnectOptions options;
+    private static int qos = 2;
+
+    //定时器
+    private Timer timer;
+
+    public MqttDianDengTechClient(){
+        try {
+            MQTT_CLIENT_ID = UUIDUtil.getUUID();
+            // host为主机名，clientid即连接MQTT的客户端ID，一般以客户端唯一标识符表示，
+            // MemoryPersistence设置clientid的保存形式，默认为以内存保存
+            mqttClient = new MqttClient(MQTT_BROKER_HOST, MQTT_CLIENT_ID, new MemoryPersistence());
+            // 配置参数信息
+            options = new MqttConnectOptions();
+            // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，
+            // 这里设置为true表示每次连接到服务器都以新的身份连接
+            options.setCleanSession(false);
+            // 设置用户名
+            options.setUserName(USERNAME);
+            // 设置密码
+            options.setPassword(PASSWORD.toCharArray());
+            // 设置超时时间 单位为秒
+            options.setConnectionTimeout(10);
+            // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
+            options.setKeepAliveInterval(20);
+            //断线重连
+            options.setAutomaticReconnect(true);
+            //mqtt服务器端单双向加密
+//            InputStream caCrtFile = this.getClass().getResourceAsStream("/ssl/my_root_ca.crt");
+//            InputStream crtFile = this.getClass().getResourceAsStream("/ssl/client.crt");
+//            InputStream keyFile = this.getClass().getResourceAsStream("/ssl/client.key");
+//            String password = "";
+            InputStream caCrtFile = this.getClass().getResourceAsStream("/ssl/my_root_ca.crt");
+            InputStream crtFile = this.getClass().getResourceAsStream("/ssl/client.crt");
+            InputStream keyFile = this.getClass().getResourceAsStream("/ssl/client.key");
+            String password = "";
+            SSLSocketFactory socketFactory = MqttSSLSocketFactory.getTwoDirSocketFactory(caCrtFile,crtFile,keyFile,password);
+            options.setSocketFactory(socketFactory);
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+    public void start(TemperatureEspService temperatureEspService) {
+        try {
+            // 连接
+            mqttClient.connect(options);
+            // 订阅
+            mqttClient.subscribe(TOPIC_FILTER,qos);
+            // 设置回调
+            mqttClient.setCallback(new MqttCallbackExtended(){
+
+                @Override
+                public void connectionLost(Throwable throwable) {
+                    log.info("connectionLost");
+                    try {
+                        mqttClient.reconnect();
+                    } catch (MqttException e) {
+                        e.printStackTrace();
+                    }
+                }
+
+                @Override
+                public void messageArrived(String s, MqttMessage mqttMessage) {
+                    String strData = new String(mqttMessage.getPayload());
+                    log.info("topic:"+s);
+                    log.info("Qos:"+mqttMessage.getQos());
+                    log.info("message RSA:"+strData);
+                    if(strData!=null&&strData.length()>=11){
+                        String[] strDataTemperature = strData.split(" ");
+                        temperatureEspService.saveTemperature(strDataTemperature[0],strDataTemperature[1],s);
+                    }
+                }
+
+                @Override
+                public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
+                    log.info("deliveryComplete---------"+ iMqttDeliveryToken.isComplete());
+                }
+
+                @Override
+                public void connectComplete(boolean b, String s) {
+                    //连接成功后调用
+                    try {
+                        mqttClient.subscribe(TOPIC_FILTER,qos);//具体订阅代码
+                    } catch (MqttException e) {
+                        e.printStackTrace();
+                    }
+                }
+            });
+            timer = new Timer();
+            timer.schedule(new TimerTask() {
+                public void run() {
+                    log.info("-------设定要指定任务--------");
+                    try {
+                        //判断拦截状态，这里注意一下，如果没有这个判断，是非常坑的
+                        if (!mqttClient.isConnected()) {
+                            log.info("***** 没有连接到服务器 *****");
+                            log.info("***** client to connect *****");
+                            // 重新连接
+                            mqttClient.connect(options);
+                        }
+                        if (mqttClient.isConnected()) {//连接成功，跳出连接
+                            log.info("***** connect success *****");
+                        }
+                    } catch (MqttException e1) {
+                        e1.printStackTrace();
+                    }
+                }
+            }, 10000,10000);
+            // 设定指定的时间time,此处为10000毫秒
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+}

src/main/java/com/rehome/mqttclienttemperature/controller/TemperatureEspController.java

@@ -36,9 +36,9 @@ public class TemperatureEspController {
      */
     @GetMapping("/saveEspTemperature")
     private String save(@Param("temperature")String temperature, @Param("humidity")String humidity) {
-        log.info(temperature);
+//        log.info(temperature);
-        log.info(humidity);
+//        log.info(humidity);
-        this.temperatureEspService.saveTemperature(temperature,humidity);
+//        this.temperatureEspService.saveTemperature(temperature,humidity);
         return "success";
     }
 }

src/main/java/com/rehome/mqttclienttemperature/service/impl/TemperatureEspServiceImpl.java

@@ -86,7 +86,7 @@ public class TemperatureEspServiceImpl implements TemperatureEspService {
 
                 if(topic!=null){
                     temperatureEsp.setTopic(topic);
-                    if(topic.equals("/device/BF9D7939ND2NBM8WVDK0SO5D/r")){
+                    if(topic.equals("/device/esp8266/001")){
                         temperatureEsp.setLocationDesc("广东省珠海市高新区唐家湾镇南方软件园B2栋4楼珠海瑞洪智能系统工程有限公司");
                     }
                     if(topic.equals("WifiSHT/7C87CE9F5CBF/SHT20")){