中心去掉物联信息生成告警信息逻辑

amos-boot-module/amos-boot-module-biz/amos-boot-module-equip-biz/src/main/java/com/yeejoin/equipmanage/config/EquipmentIotMqttReceiveConfig.java

@@ -140,7 +140,7 @@ public class EquipmentIotMqttReceiveConfig {
         adapter = new MqttPahoMessageDrivenChannelAdapter(clientId + "_inbound", mqttPahoClientFactory(), arr);
         adapter.setCompletionTimeout(completionTimeout);
         adapter.setConverter(new DefaultPahoMessageConverter());
-        adapter.setQos(0);
+        adapter.setQos(2);
         adapter.setOutputChannel(mqttInputChannel());
         return adapter;
     }

amos-boot-module/amos-boot-module-biz/amos-boot-module-equip-biz/src/main/java/com/yeejoin/equipmanage/service/impl/MqttReceiveServiceImpl.java

@@ -409,53 +409,56 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
             String messageTraceId = JSON.toJSONString(messageObj);
             mqttSendGateway.sendToMqtt("influxdb/" + topic.substring(0, endIndex), messageTraceId);
         }
-        EquipmentSpecificVo vo = eqIotCodeList.get(0);
-        topicEntity.setType(vo.getType());
-        topicEntity.setCode(vo.getCode());
 
-        JSONObject json = JSONObject.parseObject(message);
-        Iterator it = json.entrySet().iterator();
-        List<IotDataVO> iotDatalist = new ArrayList<>();
-        String traceId = "";
-        while (it.hasNext()) {
-            IotDataVO iotDataVO = new IotDataVO();
-            Map.Entry<String, Object> entry = (Map.Entry<String, Object>) it.next();
-            String key = entry.getKey();
-            Object value = entry.getValue();
-            iotDataVO.setKey(key);
-            iotDataVO.setValue(value);
-            if ("traceId".equalsIgnoreCase(key)) {
-                traceId = value.toString();
-                continue;
-            }
-            iotDatalist.add(iotDataVO);
-        }
-        if (ObjectUtils.isEmpty(iotDatalist)) {
-            return;
-        }
-        log.info(String.format("收到mqtt消息：%s", message));
-
-        // 发送emq消息转kafka
-        JSONObject jsonObject = new JSONObject();
-        jsonObject.put("topic", topic);
-        jsonObject.put("data", message);
-
-        try {
-            emqKeeper.getMqttClient().publish("emq.iot.created", jsonObject.toString().getBytes(), 1, false);
-        } catch (MqttException e) {
-            log.info(String.format("发送eqm转kafka消息失败：%s", e.getMessage()));
-        }
-
-        if (!StringUtils.isEmpty(traceId)) {
-            String finalTraceId = traceId;
-            List<IotDataVO> collect = iotDatalist.stream().map(x -> {
-                x.setTraceId(finalTraceId);
-                return x;
-            }).collect(Collectors.toList());
-            realTimeDateProcessing(topicEntity, collect, vo);
-        } else {
-            realTimeDateProcessing(topicEntity, iotDatalist, vo);
-        }
+//
+//
+//        EquipmentSpecificVo vo = eqIotCodeList.get(0);
+//        topicEntity.setType(vo.getType());
+//        topicEntity.setCode(vo.getCode());
+//
+//        JSONObject json = JSONObject.parseObject(message);
+//        Iterator it = json.entrySet().iterator();
+//        List<IotDataVO> iotDatalist = new ArrayList<>();
+//        String traceId = "";
+//        while (it.hasNext()) {
+//            IotDataVO iotDataVO = new IotDataVO();
+//            Map.Entry<String, Object> entry = (Map.Entry<String, Object>) it.next();
+//            String key = entry.getKey();
+//            Object value = entry.getValue();
+//            iotDataVO.setKey(key);
+//            iotDataVO.setValue(value);
+//            if ("traceId".equalsIgnoreCase(key)) {
+//                traceId = value.toString();
+//                continue;
+//            }
+//            iotDatalist.add(iotDataVO);
+//        }
+//        if (ObjectUtils.isEmpty(iotDatalist)) {
+//            return;
+//        }
+//        log.info(String.format("收到mqtt消息：%s", message));
+//
+//        // 发送emq消息转kafka
+//        JSONObject jsonObject = new JSONObject();
+//        jsonObject.put("topic", topic);
+//        jsonObject.put("data", message);
+//
+//        try {
+//            emqKeeper.getMqttClient().publish("emq.iot.created", jsonObject.toString().getBytes(), 1, false);
+//        } catch (MqttException e) {
+//            log.info(String.format("发送eqm转kafka消息失败：%s", e.getMessage()));
+//        }
+//
+//        if (!StringUtils.isEmpty(traceId)) {
+//            String finalTraceId = traceId;
+//            List<IotDataVO> collect = iotDatalist.stream().map(x -> {
+//                x.setTraceId(finalTraceId);
+//                return x;
+//            }).collect(Collectors.toList());
+//            realTimeDateProcessing(topicEntity, collect, vo);
+//        } else {
+//            realTimeDateProcessing(topicEntity, iotDatalist, vo);
+//        }
     }
 
     @Override
@@ -1929,16 +1932,16 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 //        equipmentSpecificAlarmLog.setStationName(stationName);
         boolean bool = equipmentSpecificAlarmLogService.save(equipmentSpecificAlarmLog);
         // 同步告警消息给平台
-        if (amosSwitch && bool) {
-            EquipmentSpecificAlarmLog alarmLog = equipmentSpecificAlarmLogService
-                    .getById(equipmentSpecificAlarmLog.getId());
-            new Thread(new Runnable() {
-                @Override
-                public void run() {
-                    syncSystemctlMsg(alarmLog);
-                }
-            }).start();
-        }
+//        if (amosSwitch && bool) {
+//            EquipmentSpecificAlarmLog alarmLog = equipmentSpecificAlarmLogService
+//                    .getById(equipmentSpecificAlarmLog.getId());
+//            new Thread(new Runnable() {
+//                @Override
+//                public void run() {
+//                    syncSystemctlMsg(alarmLog);
+//                }
+//            }).start();
+//        }
         return equipmentSpecificAlarmLog;
     }