mqtt接收数据

huimv-env-platform/huimv-env-manage/pom.xml

@@ -71,6 +71,11 @@
             <groupId>cn.hutool</groupId>
             <artifactId>hutool-all</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.google.code.gson</groupId>
+            <artifactId>gson</artifactId>
+            <version>2.8.3</version>
+        </dependency>
 
         <!-- 排除Tomcat容器 -->
        <!-- <dependency>

huimv-env-platform/huimv-env-manage/src/main/java/com/huimv/env/manage/entity/EnvAmmonia.java

@@ -36,8 +36,8 @@ public class EnvAmmonia implements Serializable {
     private Integer sensorSn;
 
     private BigDecimal ammonia;
-
-    private String askTime;
+    @JsonFormat(pattern = "MM-dd HH:mm:ss",timezone = "GMT+8")
+    private Date askTime;
 
     @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss",timezone = "GMT+8")
     private Date addTime;

huimv-env-platform/huimv-env-manage/src/main/java/com/huimv/env/manage/saas/controller/EnvDeviceEquipmentController.java

@@ -1,7 +1,6 @@
 package com.huimv.env.manage.saas.controller;
 
 
-import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONObject;
 import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
 import com.huimv.env.manage.entity.EnvDeviceRegister;
@@ -57,23 +56,19 @@ public class EnvDeviceEquipmentController {
             String chipId = one.getChipId();
             log.info("-------获取设备编码："+chipId);
             PublishMQTT publishMQTT = new PublishMQTT();
-
             log.info("-----开始配置消息----");
             QueryWrapper<EnvDeviceEquipment> queryWrapper1 = new QueryWrapper<>();
-            queryWrapper1.eq("device_code",deviceCode);
+            queryWrapper1.eq("device_code",deviceCode).orderByDesc("equipment_port");
             List<EnvDeviceEquipment> list = envDeviceEquipmentService.list(queryWrapper1);
-
-            StringBuffer deviceStatus = new StringBuffer("0000000000000000");
+            String deviceStatus = "";
             for (EnvDeviceEquipment envDeviceEquipment1 : list) {
-                Integer equipmentPort = envDeviceEquipment1.getEquipmentPort();
-                Integer openStart = envDeviceEquipment1.getUploadStart();
-                deviceStatus.replace(equipmentPort -1,equipmentPort,openStart+"");
+                deviceStatus = deviceStatus + envDeviceEquipment1.getUploadStart();
             }
-            String s = deviceStatus.toString();
+            String s = deviceStatus;
+            int i = Integer.parseInt(s, 2);
             JSONObject jsonObject  =new JSONObject();
-            jsonObject.put("dev_ctrl",s);
-            log.info("消息配置完成开始发送："+s);
-
+            jsonObject.put("dev_ctrl",i);
+            log.info("消息配置完成开始发送："+i);
             publishMQTT.start("tcp://115.238.57.190:1883",chipId,"admin","admin","huimv_down_"+chipId,jsonObject.toJSONString());
             return new Result(10000,"修改成功",true);
         }catch (Exception e){

huimv-env-platform/huimv-env-manage/src/main/java/com/huimv/env/manage/saas/mqtt/subscribe/ConsumerMQTT.java

@@ -7,7 +7,7 @@ import org.springframework.context.annotation.Configuration;
 @Configuration
 public class ConsumerMQTT {
     public static final String HOST = "tcp://192.168.1.68:1883";
-    private static final String clientid = "subscribe";
+    private static final String clientid = "test001";
     private MqttClient client;
     private MqttConnectOptions options;
 
@@ -39,7 +39,7 @@ public class ConsumerMQTT {
 
             //订阅消息
 //            int[] Qos = {1};//0：最多一次 、1：最少一次 、2：只有一次
-            client.subscribe("test",1);
+            client.subscribe("huimv_up_867699060002277",2);
 
 
         } catch (Exception e) {
@@ -47,4 +47,8 @@ public class ConsumerMQTT {
         }
     }
 
+    public static void main(String[] args) {
+        ConsumerMQTT consumerMQTT = new ConsumerMQTT();
+        consumerMQTT.start();
+    }
 }

huimv-env-platform/huimv-env-manage/src/main/java/com/huimv/env/manage/saas/mqtt/subscribe/PushCallback.java

@@ -1,12 +1,35 @@
 package com.huimv.env.manage.saas.mqtt.subscribe;
 
+import cn.hutool.core.util.ObjectUtil;
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParser;
+import com.huimv.env.common.entity.BaseThreshold;
+import com.huimv.env.common.service.IBaseThresholdService;
+import com.huimv.env.manage.entity.*;
+import com.huimv.env.manage.saas.dao.entity.EnvDeviceEquipment;
+import com.huimv.env.manage.saas.mqtt.publish.PublishMQTT;
+import com.huimv.env.manage.saas.mqtt.utils.SpringUtil;
+import com.huimv.env.manage.saas.service.IEnvDeviceEquipmentService;
+import com.huimv.env.manage.service.*;
+import com.mysql.cj.util.StringUtils;
+import lombok.extern.slf4j.Slf4j;
 import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
 import org.eclipse.paho.client.mqttv3.MqttCallback;
 import org.eclipse.paho.client.mqttv3.MqttMessage;
 import org.springframework.stereotype.Component;
 
+import java.math.BigDecimal;
+import java.text.SimpleDateFormat;
+import java.time.ZonedDateTime;
+import java.util.*;
+
 @Component
 //接收消息回调
+@Slf4j
 public class PushCallback implements MqttCallback {
 
 
@@ -30,16 +53,169 @@ public class PushCallback implements MqttCallback {
         System.out.println("接收消息内容 : " + new String(message.getPayload()));
         System.out.println("开始处理当前数据...");
 
+        String[] split = topic.split("_");
+        String chip = split[2];
+        System.out.println(chip);
         String messages = new String(message.getPayload());
-        String[] split = messages.split(",");
-//        PeopleServiceImpl peopleService = SpringUtil.getBean(PeopleServiceImpl.class);//创建impl实例
-//        People people = new People();
-//        people.setName(split[0]);
-//        people.setAge(Integer.parseInt(split[1]));
-//        people.setSex(split[2]);
-//        peopleService.save(people);
-//        peopleService.addPeople(split);
+        if (messages.equals("sync_time")) {
+            log.info("-----开始配置同步时间----");
+            JSONObject jsonObject1 = new JSONObject();
+            ZonedDateTime ZonedDateTime = java.time.ZonedDateTime.now();
+            Map<String, Object> map = new HashMap<>();
+            map.put("year", ZonedDateTime.getYear());
+            map.put("mon", ZonedDateTime.getMonthValue());
+            map.put("date", ZonedDateTime.getDayOfMonth());
+            map.put("hour", ZonedDateTime.getHour());
+            map.put("min", ZonedDateTime.getMinute());
+            map.put("sec", ZonedDateTime.getSecond());
+            jsonObject1.put("sync_time", map);
+            System.out.println(jsonObject1);
+            try {
+                PublishMQTT publishMQTT = new PublishMQTT();
+                publishMQTT.start("tcp://192.168.1.68:1883", chip, "admin", "admin", "huimv_down_" + chip, jsonObject1.toJSONString());
+            } catch (Exception e) {
+                System.out.println(e);
+            }
+        } else if (messages.equals("alarm_cfg")) {
+            log.info("-----开始配置报警信息----");
+            IEnvDeviceRegisterService registerService = SpringUtil.getBean(IEnvDeviceRegisterService.class);
+            QueryWrapper<EnvDeviceRegister> queryWrapper = new QueryWrapper<>();
+            queryWrapper.eq("chip_id", chip);
+            EnvDeviceRegister register = registerService.getOne(queryWrapper);
+            IBaseThresholdService thresholdService = SpringUtil.getBean(IBaseThresholdService.class);
+            QueryWrapper<BaseThreshold> queryWrapper1 = new QueryWrapper<>();
+            queryWrapper1.eq("farm_code", register.getFarmCode());
+            BaseThreshold threshold = thresholdService.getOne(queryWrapper1);
+
+            JSONObject jsonObject1 = new JSONObject();
+            Map<String, Object> map = new HashMap<>();
+            Map<String, Object> map1 = new HashMap<>();
+            Map<String, Object> map2 = new HashMap<>();
+            Map<String, Object> map3 = new HashMap<>();
+
+            BigDecimal big2 = BigDecimal.valueOf(10);
+            map1.put("en", 1);
+            map1.put("param", 10);
+            map1.put("condition", 0);
+            map2.put("en", 1);
+            map2.put("param", 10);//高温阈值
+            map2.put("condition", threshold.getMaxTemp().multiply(big2));
+            map3.put("en", 1);
+            map3.put("param", 10);
+            map3.put("condition", threshold.getMinTemp().multiply(big2));//低温阈值
+            map.put("power_off", map1);
+            map.put("high_temp", map2);
+            map.put("low_temp", map3);
+            jsonObject1.put("alarm_cfg", map);
+            System.out.println(jsonObject1);
+            PublishMQTT publishMQTT = new PublishMQTT();
+            publishMQTT.start("tcp://192.168.1.68:1883", chip, "admin", "admin", "huimv_down_" + chip, jsonObject1.toJSONString());
+            log.info("----报警配置下发完成---");
+        } else if (messages.equals("get_dev_ctrl")) {
+            JSONObject jsonObject = new JSONObject();
+            log.info("-----开始配置设备控制消息----");
+            QueryWrapper<EnvDeviceRegister> queryWrapper1 = new QueryWrapper<>();
+            queryWrapper1.eq("chip_id", chip);
+            IEnvDeviceRegisterService envDeviceRegisterService = SpringUtil.getBean(IEnvDeviceRegisterService.class);
+            EnvDeviceRegister one = envDeviceRegisterService.getOne(queryWrapper1);
+
+            IEnvDeviceEquipmentService envDeviceEquipmentService = SpringUtil.getBean(IEnvDeviceEquipmentService.class);
+            QueryWrapper<EnvDeviceEquipment> queryWrapper = new QueryWrapper<>();
+            queryWrapper.eq("device_code", one.getDeviceCode()).orderByDesc("equipment_port");
+            List<EnvDeviceEquipment> list = envDeviceEquipmentService.list(queryWrapper);
+            String deviceStatus = "";
+            for (EnvDeviceEquipment envDeviceEquipment1 : list) {
+                deviceStatus = deviceStatus + envDeviceEquipment1.getUploadStart();
+            }
+            String s = deviceStatus;
+            int i = Integer.parseInt(s, 2);
+            jsonObject.put("dev_ctrl", i);
+            log.info("设备配置完成开始发送：" + i);
+            PublishMQTT publishMQTT = new PublishMQTT();
+            publishMQTT.start("tcp://192.168.1.68:1883", chip, "admin", "admin", "huimv_down_" + chip, jsonObject.toJSONString());
+            log.info("---设备控制发送完成---");
+        } else {
+            JSONObject jsonObject = JSON.parseObject(messages);
+            System.out.println("这里是1111：" + jsonObject);
+            if (ObjectUtil.isNotEmpty(jsonObject.get("sensor"))) {
+                log.info("----开始保存温度---");
+                Date date = new Date();
+                IEnvDeviceRegisterService registerService = SpringUtil.getBean(IEnvDeviceRegisterService.class);
+                QueryWrapper<EnvDeviceRegister> queryWrapper = new QueryWrapper<>();
+                queryWrapper.eq("chip_id", chip);
+                EnvDeviceRegister deviceRegister = registerService.getOne(queryWrapper);
+                IEnvTempService tempService = SpringUtil.getBean(IEnvTempService.class);
+                Map map = (Map) jsonObject.get("sensor");
+                Map tempData = (Map) map.get("temp");
+                JSONArray jsonArray = (JSONArray) tempData.get("data");
+                BigDecimal temp = new BigDecimal(Integer.parseInt(jsonArray.get(0).toString()));
+                BigDecimal big2 = BigDecimal.valueOf(10);
+                EnvTemp envTemp = new EnvTemp();
+                envTemp.setDeviceCode(deviceRegister.getDeviceCode());
+                envTemp.setTemp(temp.divide(big2));
+                envTemp.setAskTime(date);
+                envTemp.setAddTime(date);
+                envTemp.setAddDate(date);
+                envTemp.setFarmCode(deviceRegister.getFarmCode());
+                envTemp.setUnitId(deviceRegister.getUnitId());
+                tempService.save(envTemp);
+                log.info("----温度保存成功！----");
+                log.info("-----开始保留湿度----");
+                IEnvHumiService humiService = SpringUtil.getBean(IEnvHumiService.class);
+                Map map1 = (Map) jsonObject.get("sensor");
+                Map humiData = (Map) map1.get("temp");
+                JSONArray jsonArray1 = (JSONArray) humiData.get("data");
+                BigDecimal humi = new BigDecimal(Integer.parseInt(jsonArray1.get(0).toString()));
+                EnvHumi envHumi = new EnvHumi();
+                envHumi.setDeviceCode(deviceRegister.getDeviceCode());
+                envHumi.setHumi(humi.divide(big2));
+                envHumi.setAskTime(date);
+                envHumi.setAddTime(date);
+                envHumi.setAddDate(date);
+                envHumi.setFarmCode(deviceRegister.getFarmCode());
+                envHumi.setUnitId(deviceRegister.getUnitId());
+                humiService.save(envHumi);
+                log.info("----湿度保存成功！----");
+                log.info("----开始保存氨气---");
+                IEnvAmmoniaService ammoniaService = SpringUtil.getBean(IEnvAmmoniaService.class);
+                Map map2 = (Map) jsonObject.get("sensor");
+                Map ammoData = (Map) map2.get("nh3");
+                JSONArray jsonArray2 = (JSONArray) ammoData.get("data");
+                BigDecimal ammo = new BigDecimal(Integer.parseInt(jsonArray2.get(0).toString()));
+                EnvAmmonia envAmmonia = new EnvAmmonia();
+                envAmmonia.setDeviceCode(deviceRegister.getDeviceCode());
+                envAmmonia.setAmmonia(ammo.divide(big2));
+                envAmmonia.setAskTime(date);
+                envAmmonia.setAddTime(date);
+                envAmmonia.setAddDate(date);
+                envAmmonia.setFarmCode(deviceRegister.getFarmCode());
+                envAmmonia.setUnitId(deviceRegister.getUnitId());
+                ammoniaService.save(envAmmonia);
+                log.info("---氨气保存成功！----");
+            }
+
+
+            if (ObjectUtil.isNotEmpty(jsonObject.get("alarm_event"))) {
+                log.info("----报警信息开始保存---");
+                IEnvAlarmTypeService alarmTypeService = SpringUtil.getBean(IEnvAlarmTypeService.class);
+                JSONObject jsonObject1 = (JSONObject) jsonObject.get("alarm_event");
+                System.out.println(jsonObject1);
+                EnvAlarmType envAlarmType = new EnvAlarmType();
+                try {
+                    envAlarmType.setFaultType(jsonObject1.get("type").toString());
+                    envAlarmType.setFaultSn((Integer) jsonObject1.get("sta"));
+                    System.out.println(envAlarmType);
+                } catch (Exception e) {
+                    System.out.println(e);
+                }
+
+                alarmTypeService.save(envAlarmType);
+                log.info("----报警信息保存成功！----");
+            }
+        }
+
         System.out.println("数据处理完成！");
     }
 
+
 }

huimv-env-platform/huimv-env-manage/src/main/java/com/huimv/env/manage/saas/mqtt/uplod/UploadMqttStart.java

@@ -1,11 +1,13 @@
 package com.huimv.env.manage.saas.mqtt.uplod;
 
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
 import com.huimv.env.manage.entity.EnvDeviceRegister;
 import com.huimv.env.manage.service.IEnvDeviceRegisterService;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
 import javax.annotation.PostConstruct;
+import javax.management.Query;
 import java.util.List;
 
 @Component
@@ -15,15 +17,24 @@ public class UploadMqttStart {
     private IEnvDeviceRegisterService deviceRegisterService;
 
     @PostConstruct
-    public void test(){
+    public void test() {
         List<EnvDeviceRegister> list = deviceRegisterService.list();
         for (EnvDeviceRegister deviceRegister : list) {
             String chipId = deviceRegister.getChipId();
             uploadmqtt uploadmqtt = new uploadmqtt();
-            uploadmqtt.setClientid(chipId);
-            uploadmqtt.setTopic("huimv_up_"+chipId);
+            uploadmqtt.setClientid("pub_" + chipId);
+            uploadmqtt.setTopic("huimv_up_" + chipId);
             uploadmqtt.start();
         }
+//        QueryWrapper<EnvDeviceRegister> queryWrapper = new QueryWrapper<>();
+//        queryWrapper.eq("device_code", 122);
+//        EnvDeviceRegister one = deviceRegisterService.getOne(queryWrapper);
+//        String chipId = one.getChipId();
+//        System.out.println("这里拿到了设备编码");
+//            uploadmqtt uploadmqtt = new uploadmqtt();
+//            uploadmqtt.setClientid("pub_"+chipId);
+//            uploadmqtt.setTopic("huimv_up_"+chipId);
+//            uploadmqtt.start();
     }
 
 }