添加运行状态统计功能

iot-access/access-api/src/main/java/com/njcn/access/api/AskDeviceDataFeignClient.java

@@ -35,6 +35,6 @@ public interface AskDeviceDataFeignClient {
     HttpResult<String> askRealData(@RequestParam("nDid") String nDid, @RequestParam("idx") Integer idx, @RequestParam("clDId") Integer clDId);
 
     @PostMapping("/askCldRealData")
-    HttpResult<String> askCldRealData(@RequestParam("devId") String devId, @RequestParam("lineId") String lineId, @RequestParam("nodeId") String nodeId);
+    HttpResult<String> askCldRealData(@RequestParam("devId") String devId, @RequestParam("lineId") String lineId, @RequestParam("nodeId") String nodeId, @RequestParam("idx") Integer idx);
 
 }

iot-access/access-api/src/main/java/com/njcn/access/api/fallback/AskDeviceDataClientFallbackFactory.java

@@ -75,7 +75,7 @@ public class AskDeviceDataClientFallbackFactory implements FallbackFactory<AskDe
             }
 
             @Override
-            public HttpResult<String> askCldRealData(String devId, String lineId, String nodeId) {
+            public HttpResult<String> askCldRealData(String devId, String lineId, String nodeId, Integer idx) {
                 log.error("{}异常，降级处理，异常为：{}","询问云前置实时数据",cause.toString());
                 throw new BusinessException(finalExceptionEnum);
             }

iot-access/access-api/src/main/java/com/njcn/access/enums/AccessResponseEnum.java

@@ -56,7 +56,7 @@ public enum AccessResponseEnum {
     CTRL_DICT_MISSING("A0307","Ctrl字典数据缺失!"),
     WAVE_INFO_MISSING("A0307","波形参数缺失!"),
 
-    MODEL_MISS("A0308","模板信息缺失!"),
+    MODEL_MISS("A0308","询问模板信息超时,设备未响应!"),
     MODEL_VERSION_ERROR("A0308","询问装置模板信息错误"),
     UPLOAD_ERROR("A0308","平台上送文件异常"),
     RELOAD_UPLOAD_ERROR("A0308","平台重新上送文件异常"),

iot-access/access-boot/src/main/java/com/njcn/access/controller/AskDeviceDataController.java

@@ -129,11 +129,12 @@ public class AskDeviceDataController extends BaseController {
     @ApiImplicitParams({
             @ApiImplicitParam(name = "devId", value = "装置id"),
             @ApiImplicitParam(name = "lineId", value = "监测点id"),
-            @ApiImplicitParam(name = "nodeId", value = "前置id")
+            @ApiImplicitParam(name = "nodeId", value = "前置id"),
+            @ApiImplicitParam(name = "idx", value = "数据集编号")
     })
-    public HttpResult<String> askCldRealData(@RequestParam("devId") String devId, @RequestParam("lineId") String lineId, @RequestParam("nodeId") String nodeId){
+    public HttpResult<String> askCldRealData(@RequestParam("devId") String devId, @RequestParam("lineId") String lineId, @RequestParam("nodeId") String nodeId, @RequestParam("idx") Integer idx){
         String methodDescribe = getMethodDescribe("askCldRealData");
-        askDeviceDataService.askCldRealData(devId,lineId,nodeId);
+        askDeviceDataService.askCldRealData(devId,lineId,nodeId,idx);
         return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, null, methodDescribe);
     }
 

iot-access/access-boot/src/main/java/com/njcn/access/handler/MqttMessageHandler.java

@@ -15,7 +15,6 @@ import com.github.tocrhz.mqtt.publisher.MqttPublisher;
 import com.njcn.access.enums.AccessEnum;
 import com.njcn.access.enums.AccessResponseEnum;
 import com.njcn.access.enums.TypeEnum;
-import com.njcn.access.mapper.OverlimitMapper;
 import com.njcn.access.pojo.RspDataDto;
 import com.njcn.access.pojo.dto.*;
 import com.njcn.access.pojo.dto.file.FileDto;
@@ -34,6 +33,7 @@ import com.njcn.csdevice.api.*;
 import com.njcn.csdevice.pojo.dto.PqsCommunicateDto;
 import com.njcn.csdevice.pojo.param.CsLineParam;
 import com.njcn.csdevice.pojo.po.*;
+import com.njcn.device.biz.mapper.OverLimitWlMapper;
 import com.njcn.device.biz.pojo.po.Overlimit;
 import com.njcn.device.biz.utils.COverlimitUtil;
 import com.njcn.mq.message.AppAutoDataMessage;
@@ -69,6 +69,8 @@ import java.util.stream.Collectors;
 
 import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
 
+;
+
 /**
  * @author hongawen
  * @version 1.0.0
@@ -96,7 +98,7 @@ public class MqttMessageHandler {
     private final CsLineFeignClient csLineFeignClient;
     private final DevCapacityFeignClient devCapacityFeignClient;
     private final EquipmentFeignClient equipmentFeignClient;
-    private final OverlimitMapper overlimitMapper;
+    private final OverLimitWlMapper overLimitWlMapper;
     private final ChannelObjectUtil channelObjectUtil;
     private final WaveFeignClient waveFeignClient;
     private final RtFeignClient rtFeignClient;
@@ -380,32 +382,27 @@ public class MqttMessageHandler {
                                     if (Objects.equals(res.getDid(),1)){
                                         log.info("{},设备数据应答--->更新治理监测点信息和设备容量", nDid);
                                         List<CsDevCapacityPO> list3 = new ArrayList<>();
-                                        devInfo.forEach(item->{
-                                            //1.更新治理监测点信息
-                                            CsLineParam csLineParam = new CsLineParam();
-                                            if (Objects.equals(item.getClDid(),0)){
-                                                csLineParam.setLineId(nDid.concat("0"));
+                                        boolean hasZeroClDid = devInfo.stream().anyMatch(item -> item.getClDid() == 0);
+                                        //治理设备
+                                        if (hasZeroClDid) {
+                                            devInfo.forEach(item->{
+                                                if (Objects.equals(item.getClDid(),0)){
+                                                    updateLineInfo(nDid,item);
+                                                }
                                                 //2.录入各个模块设备容量
                                                 CsDevCapacityPO csDevCapacity = new CsDevCapacityPO();
                                                 csDevCapacity.setLineId(nDid.concat("0"));
                                                 csDevCapacity.setCldid(item.getClDid());
                                                 csDevCapacity.setCapacity(Objects.isNull(item.getCapacityA())?0.0:item.getCapacityA());
                                                 list3.add(csDevCapacity);
-                                            } else {
-                                                csLineParam.setLineId(nDid.concat(item.getClDid().toString()));
-                                            }
-                                            csLineParam.setVolGrade(item.getVolGrade());
-                                            csLineParam.setPtRatio(item.getPtRatio());
-                                            csLineParam.setCtRatio(item.getCtRatio());
-                                            csLineParam.setConType(item.getConType());
-                                            csLineParam.setLineInterval(item.getStatCycle());
-                                            csLineFeignClient.updateLine(csLineParam);
-                                            //生成监测点限值
-                                            Overlimit overlimit = COverlimitUtil.globalAssemble(item.getVolGrade().floatValue(),10f,10f,10f,0,0);
-                                            overlimit.setId(nDid.concat(item.getClDid().toString()));
-                                            overlimitMapper.deleteById(nDid.concat(item.getClDid().toString()));
-                                            overlimitMapper.insert(overlimit);
-                                        });
+                                            });
+                                        }
+                                        //其余设备
+                                        else {
+                                            devInfo.forEach(item->{
+                                                updateLineInfo(nDid,item);
+                                            });
+                                        }
                                         if (CollectionUtil.isNotEmpty(list3)) {
                                             devCapacityFeignClient.addList(list3);
                                             //3.更新设备模块个数
@@ -416,14 +413,7 @@ public class MqttMessageHandler {
                                         logDto.setOperate(nDid + "更新电网侧、负载侧监测点信息");
                                         //1.更新电网侧、负载侧监测点相关信息
                                         devInfo.forEach(item->{
-                                            CsLineParam csLineParam = new CsLineParam();
-                                            csLineParam.setLineId(nDid.concat(item.getClDid().toString()));
-                                            csLineParam.setVolGrade(item.getVolGrade());
-                                            csLineParam.setPtRatio(item.getPtRatio());
-                                            csLineParam.setCtRatio(item.getCtRatio());
-                                            csLineParam.setConType(item.getConType());
-                                            csLineParam.setLineInterval(item.getStatCycle());
-                                            csLineFeignClient.updateLine(csLineParam);
+                                            updateLineInfo(nDid,item);
                                         });
                                     }
                                 }
@@ -468,6 +458,22 @@ public class MqttMessageHandler {
         }
     }
 
+    public void updateLineInfo(String nDid,RspDataDto.LdevInfo item) {
+        CsLineParam csLineParam = new CsLineParam();
+        csLineParam.setLineId(nDid.concat(item.getClDid().toString()));
+        csLineParam.setVolGrade(item.getVolGrade());
+        csLineParam.setPtRatio(item.getPtRatio());
+        csLineParam.setCtRatio(item.getCtRatio());
+        csLineParam.setConType(item.getConType());
+        csLineParam.setLineInterval(item.getStatCycle());
+        csLineFeignClient.updateLine(csLineParam);
+        //生成监测点限值
+        Overlimit overlimit = COverlimitUtil.globalAssemble(item.getVolGrade().floatValue(),10f,10f,10f,0,0);
+        overlimit.setId(nDid.concat(item.getClDid().toString()));
+        overLimitWlMapper.deleteById(nDid.concat(item.getClDid().toString()));
+        overLimitWlMapper.insert(overlimit);
+    }
+
     /**
      * 装置心跳 && 主动数据上送
      * fixme 这边由于接收文件数据时间跨度会很长，途中有其他请求进来会中断之前的程序，目前是记录中断的位置，等处理完成再继续请求接收文件

iot-access/access-boot/src/main/java/com/njcn/access/mapper/OverlimitMapper.java

@@ -1,21 +1,21 @@
-package com.njcn.access.mapper;
-
-
-import com.baomidou.dynamic.datasource.annotation.DS;
-import com.baomidou.mybatisplus.core.mapper.BaseMapper;
-import com.njcn.device.biz.pojo.po.Overlimit;
-import org.apache.ibatis.annotations.Mapper;
-
-
-/**
- * <p>
- *  Mapper 接口
- * </p>
- *
- * @author xy
- */
-@DS("sjzx")
-@Mapper
-public interface OverlimitMapper extends BaseMapper<Overlimit> {
-
-}
+//package com.njcn.access.mapper;
+//
+//
+//import com.baomidou.dynamic.datasource.annotation.DS;
+//import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+//import com.njcn.device.biz.pojo.po.Overlimit;
+//import org.apache.ibatis.annotations.Mapper;
+//
+//
+///**
+// * <p>
+// *  Mapper 接口
+// * </p>
+// *
+// * @author xy
+// */
+//@DS("sjzx")
+//@Mapper
+//public interface OverlimitMapper extends BaseMapper<Overlimit> {
+//
+//}

iot-access/access-boot/src/main/java/com/njcn/access/service/AskDeviceDataService.java

@@ -22,5 +22,5 @@ public interface AskDeviceDataService {
      */
     void askRealData(String nDid, Integer idx, Integer size);
 
-    void askCldRealData(String devId, String lineId, String nodeId);
+    void askCldRealData(String devId, String lineId, String nodeId, Integer idx);
 }

iot-access/access-boot/src/main/java/com/njcn/access/service/impl/AskDeviceDataServiceImpl.java

@@ -213,7 +213,7 @@ public class AskDeviceDataServiceImpl implements AskDeviceDataService {
     }
 
     @Override
-    public void askCldRealData(String devId, String lineId, String nodeId) {
+    public void askCldRealData(String devId, String lineId, String nodeId, Integer idx) {
         RealDataMessage realDataMessage = new RealDataMessage();
         realDataMessage.setDevSeries(devId);
         int lastDigit = Character.getNumericValue(lineId.charAt(lineId.length() - 1));
@@ -221,6 +221,7 @@ public class AskDeviceDataServiceImpl implements AskDeviceDataService {
         realDataMessage.setRealData(true);
         realDataMessage.setSoeData(true);
         realDataMessage.setLimit(20);
+        realDataMessage.setIdx(idx);
         realDataMessageTemplate.sendMember(realDataMessage,nodeId);
     }
 

iot-access/access-boot/src/main/java/com/njcn/access/service/impl/CsDevModelServiceImpl.java

@@ -2,6 +2,7 @@ package com.njcn.access.service.impl;
 
 import cn.hutool.core.collection.CollectionUtil;
 import cn.hutool.core.util.IdUtil;
+import cn.hutool.core.util.ObjectUtil;
 import com.alibaba.nacos.shaded.com.google.gson.Gson;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.github.tocrhz.mqtt.publisher.MqttPublisher;
@@ -21,11 +22,13 @@ import com.njcn.access.utils.CRC32Utils;
 import com.njcn.access.utils.JsonUtil;
 import com.njcn.common.pojo.dto.DeviceLogDTO;
 import com.njcn.common.pojo.exception.BusinessException;
-import com.njcn.csdevice.api.CsLogsFeignClient;
-import com.njcn.csdevice.api.DevModelFeignClient;
+import com.njcn.csdevice.api.*;
 import com.njcn.csdevice.enums.AlgorithmResponseEnum;
 import com.njcn.csdevice.pojo.param.CsDevModelAddParm;
-import com.njcn.csdevice.pojo.po.*;
+import com.njcn.csdevice.pojo.po.CsDataArray;
+import com.njcn.csdevice.pojo.po.CsDataSet;
+import com.njcn.csdevice.pojo.po.CsDevModelPO;
+import com.njcn.csdevice.pojo.po.CsEquipmentDeliveryPO;
 import com.njcn.oss.constant.OssPath;
 import com.njcn.oss.utils.FileStorageUtil;
 import com.njcn.redis.pojo.enums.AppRedisKey;
@@ -46,7 +49,6 @@ import org.springframework.web.multipart.MultipartFile;
 
 import java.sql.Date;
 import java.util.*;
-import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
 
 /**
@@ -78,6 +80,9 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
     private final MqttPublisher publisher;
     private final ICsTopicService csTopicService;
     private final RedisUtil redisUtil;
+    private final EquipmentFeignClient eequipmentFeignClient;
+    private final DevModelRelationFeignClient devModelRelationFeignClient;
+    private final CsLineFeignClient csLineFeignClient;
 
     @Override
     @Transactional(rollbackFor = {Exception.class})
@@ -115,6 +120,24 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
             analysisDataSet(templateDto,csDevModelPo.getId());
             //3.录入监测点模板表(记录当前模板有几个监测点，治理类型的模板目前规定1个监测点，电能质量模板根据逻辑子设备来)
             addCsLineModel(templateDto,csDevModelPo.getId());
+            //4.如果是云前置的模板录入,重新录入完需要将模板关系信息重新录入
+            if (Objects.equals(devType,DicDataEnum.DEV_CLD.getCode())) {
+                List<CsEquipmentDeliveryPO> list = eequipmentFeignClient.getAll().getData();
+                if (CollectionUtil.isNotEmpty(list)) {
+                    List<String> devList = list.stream()
+                            .filter(item -> Objects.equals(item.getDevAccessMethod(),"CLD"))
+                            .map(CsEquipmentDeliveryPO::getId)
+                            .collect(Collectors.toList());
+                    devModelRelationFeignClient.updateDataByList(devList,csDevModelPo.getId());
+                    Object object = redisUtil.getObjectByKey("setId:" + csDevModelPo.getId());
+                    if (ObjectUtil.isNotNull(object)) {
+                        csLineFeignClient.updateDataByList(devList,csDevModelPo.getId(),object.toString());
+                    }
+
+                }
+            }
+            //5.清空模板缓存
+            redisUtil.deleteKeysByString(AppRedisKey.DEV_MODEL);
             csLogsFeignClient.addUserLog(logDto);
         } catch (Exception e) {
             logDto.setResult(0);
@@ -943,6 +966,11 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
         }
         if(CollectionUtil.isNotEmpty(setList)) {
             csDataSetService.addList(setList);
+            setList.forEach(item->{
+                if (Objects.equals(item.getName(),"Ds$Pqd$Stat$01")) {
+                    redisUtil.saveByKeyWithExpire("setId:" + pId,item.getId(),30L);
+                }
+            });
         }
         if(CollectionUtil.isNotEmpty(arrayList)) {
             csDataArrayService.addList(arrayList);
@@ -1246,19 +1274,26 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
                 showName = "APF模块8数据模板";
                 break;
             case "Ds$Pqd$Stat$01":
-                if (Objects.equals(code, DicDataEnum.CONNECT_DEV.getCode()) || Objects.isNull(code)){
+                if (Objects.equals(code, DicDataEnum.CONNECT_DEV.getCode())){
                     showName = "电网侧数据模板";
                 } else if (Objects.equals(code, DicDataEnum.PORTABLE.getCode())){
                     showName = "监测1#数据模板";
+                } else {
+                    showName = "统计数据";
                 }
                 break;
             case "Ds$Pqd$Stat$02":
-                if (Objects.equals(code, DicDataEnum.CONNECT_DEV.getCode()) || Objects.isNull(code)){
+                if (Objects.equals(code, DicDataEnum.CONNECT_DEV.getCode())){
                     showName = "负载侧数据模板";
                 } else if (Objects.equals(code, DicDataEnum.PORTABLE.getCode())){
                     showName = "监测2#数据模板";
+                } else {
+                    showName = "统计数据";
                 }
                 break;
+            case "Ds$Pqd$Rt$01":
+                showName = "实时数据";
+                break;
             //波形参数名称
             case "Wave_Param_Position":
                 showName = "录波记录位置";

iot-access/access-boot/src/main/java/com/njcn/access/service/impl/CsDeviceServiceImpl.java

@@ -38,7 +38,6 @@ import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
@@ -328,7 +327,12 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
         param.setNDid(nDid);
         param.setStatus(1);
         param.setRunStatus(1);
-        param.setProcess(2);
+        boolean isConnectDev = DicDataEnum.CONNECT_DEV.getCode().equals(dictTreeFeignClient.queryById(csEquipmentDeliveryVO.getDevType()).getData().getCode());
+        if (isConnectDev) {
+            param.setProcess(2);
+        } else {
+            param.setProcess(4);
+        }
         csEquipmentDeliveryService.devResetFactory(param);
         //清除关系表
         QueryWrapper<CsLedger> csLedgerQueryWrapper = new QueryWrapper<>();
@@ -374,12 +378,6 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
     @Transactional(rollbackFor = Exception.class)
     public String wlDevRegister(String nDid) {
         String result = "fail";
-        // 设备状态判断
-        checkDeviceStatus(nDid);
-        // 询问设备支持的主题信息，并将支持的主题入库
-        askAndStoreTopics(nDid);
-        // MQTT询问装置用的模板，并判断库中是否存在模板
-        checkDeviceModel(nDid);
         // 根据模板接入设备
         DeviceLogDTO logDto = new DeviceLogDTO();
         logDto.setUserName(RequestUtil.getUserNickname());
@@ -387,9 +385,14 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
         logDto.setOperate("便携式设备"+nDid+"注册、接入");
         logDto.setResult(1);
         try {
+            // 设备状态判断
+            checkDeviceStatus(nDid);
+            // 询问设备支持的主题信息，并将支持的主题入库
+            askAndStoreTopics(nDid);
+            Thread.sleep(2000);
+            // MQTT询问装置用的模板，并判断库中是否存在模板
+            checkDeviceModel(nDid);
             Thread.sleep(2000);
-            //获取版本
-            String version = csTopicService.getVersion(nDid);
             CsEquipmentDeliveryVO vo = equipmentFeignClient.queryEquipmentByndid(nDid).getData();
             List<CsLinePO> csLinePoList = new ArrayList<>();
             //1.录入装置台账信息
@@ -441,6 +444,9 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
             csDevModelRelationService.addDevModelRelation(csDevModelRelationAddParm);
             //5.发起自动接入请求
             Thread.sleep(2000);
+            //先获取版本
+//            String version = csTopicService.getVersion(nDid);
+            String version = "V1";
             devAccessAskTemplate(nDid,version,1);
             //6.修改流程,便携式设备接入成功即为实际环境
             csEquipmentDeliveryService.updateProcessBynDid(nDid,4);

iot-access/access-boot/src/main/java/com/njcn/access/service/impl/CsEquipmentDeliveryServiceImpl.java

@@ -1,7 +1,6 @@
 package com.njcn.access.service.impl;
 
 import cn.hutool.core.collection.CollUtil;
-import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
 import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
@@ -18,7 +17,10 @@ import lombok.RequiredArgsConstructor;
 import org.springframework.beans.BeanUtils;
 import org.springframework.stereotype.Service;
 
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
 import java.util.stream.Collectors;
 
 /**
@@ -151,7 +153,6 @@ public class CsEquipmentDeliveryServiceImpl extends ServiceImpl<CsEquipmentDeliv
         QueryWrapper<CsEquipmentDeliveryPO> wrapper = new QueryWrapper<>();
         wrapper.select("DISTINCT CONCAT(node_id, node_process) as concatenated");
         wrapper.eq("usage_status", 1);
-        wrapper.eq("run_status", 2);
         wrapper.isNotNull("node_id");
         return baseMapper.selectObjs(wrapper)
                 .stream()

iot-analysis/analysis-rt/rt-boot/src/main/java/com/njcn/rt/service/impl/RtServiceImpl.java

@@ -88,7 +88,7 @@ public class RtServiceImpl implements IRtService {
             long timestamp = item.getDataTimeSec() - 8*3600;
             baseRealDataSet.setDataTime(getTime(timestamp));
             publisher.send("/Web/RealData/" + lineId, new Gson().toJson(baseRealDataSet), 1, false);
-        } else if (dataSet.getName().contains("实时数据")) {
+        } else if (dataSet.getName().contains("实时数据") || dataSet.getName().contains("Ds$Pqd$Rt$01")) {
             //用户Id
             Object redisObject = redisUtil.getObjectByKey("rtDataUserId:"+lineId);
             if (ObjectUtil.isNotNull(redisObject)) {
@@ -102,12 +102,13 @@ public class RtServiceImpl implements IRtService {
                     baseRealDataSet.setDataLevel(dataSet.getDataLevel());
                     long timestamp = item.getDataTimeSec();
                     baseRealDataSet.setDataTime(getTime(timestamp));
-                    publisher.send("/Web/RealData/" + lineId, new Gson().toJson(baseRealDataSet), 1, false);
+                    publisher.send("/Web/RealData/" + userId, new Gson().toJson(baseRealDataSet), 1, false);
                 });
             }
         }
         //fixme 目前实时数据只有基础数据和谐波数据，后期拓展，这边需要再判断
         else {
+            long timestamp;
             //用户Id
             String userId = redisUtil.getObjectByKey("rtDataUserId:"+lineId).toString();
             HarmRealDataSet harmRealDataSet = harmData(dataArrayList,item,dataSet.getDataLevel(),po.getCtRatio());
@@ -116,7 +117,11 @@ public class RtServiceImpl implements IRtService {
             harmRealDataSet.setPt(po.getPtRatio().floatValue());
             harmRealDataSet.setCt(po.getCtRatio().floatValue());
             harmRealDataSet.setDataLevel(dataSet.getDataLevel());
-            long timestamp = item.getDataTimeSec() - 8*3600;
+            if (ObjectUtil.isNotNull(po.getLineNo())) {
+                timestamp = item.getDataTimeSec();
+            } else {
+                timestamp = item.getDataTimeSec() - 8*3600;
+            }
             harmRealDataSet.setDataTime(getTime(timestamp));
             publisher.send("/Web/RealData/" + lineId, new Gson().toJson(harmRealDataSet), 1, false);
         }
@@ -220,9 +225,15 @@ public class RtServiceImpl implements IRtService {
             baseRealDataSet.setVRmsC(FloatUtils.get2Float(map.get("Pq_RmsLUCA")));
         }
         //基波电压幅值
-        baseRealDataSet.setV1A(FloatUtils.get2Float(map.get("Pq_RmsFundUA")));
-        baseRealDataSet.setV1B(FloatUtils.get2Float(map.get("Pq_RmsFundUB")));
-        baseRealDataSet.setV1C(FloatUtils.get2Float(map.get("Pq_RmsFundUC")));
+        if (conType == 0) {
+            baseRealDataSet.setV1A(FloatUtils.get2Float(map.get("Pq_RmsFundUA")));
+            baseRealDataSet.setV1B(FloatUtils.get2Float(map.get("Pq_RmsFundUB")));
+            baseRealDataSet.setV1C(FloatUtils.get2Float(map.get("Pq_RmsFundUC")));
+        } else {
+            baseRealDataSet.setV1A(FloatUtils.get2Float(map.get("Pq_RmsFundLUAB")));
+            baseRealDataSet.setV1B(FloatUtils.get2Float(map.get("Pq_RmsFundLUBC")));
+            baseRealDataSet.setV1C(FloatUtils.get2Float(map.get("Pq_RmsFundLUCA")));
+        }
         //电流有效值
         baseRealDataSet.setIRmsA(FloatUtils.get2Float(map.get("Pq_RmsIA")));
         baseRealDataSet.setIRmsB(FloatUtils.get2Float(map.get("Pq_RmsIB")));
@@ -232,21 +243,39 @@ public class RtServiceImpl implements IRtService {
         baseRealDataSet.setI1B(FloatUtils.get2Float(map.get("Pq_RmsFundIB")));
         baseRealDataSet.setI1C(FloatUtils.get2Float(map.get("Pq_RmsFundIC")));
         //电压偏差
-        baseRealDataSet.setVDevA(FloatUtils.get2Float(map.get("Pq_UDevA")));
-        baseRealDataSet.setVDevB(FloatUtils.get2Float(map.get("Pq_UDevB")));
-        baseRealDataSet.setVDevC(FloatUtils.get2Float(map.get("Pq_UDevC")));
+        if (conType == 0) {
+            baseRealDataSet.setVDevA(FloatUtils.get2Float(map.get("Pq_UDevA")));
+            baseRealDataSet.setVDevB(FloatUtils.get2Float(map.get("Pq_UDevB")));
+            baseRealDataSet.setVDevC(FloatUtils.get2Float(map.get("Pq_UDevC")));
+        } else {
+            baseRealDataSet.setVDevA(FloatUtils.get2Float(map.get("Pq_LUDevAB")));
+            baseRealDataSet.setVDevB(FloatUtils.get2Float(map.get("Pq_LUDevBC")));
+            baseRealDataSet.setVDevC(FloatUtils.get2Float(map.get("Pq_LUDevCA")));
+        }
         //基波电压相位
-        baseRealDataSet.setV1AngA(FloatUtils.get2Float(map.get("Pq_FundUAngA")));
-        baseRealDataSet.setV1AngB(FloatUtils.get2Float(map.get("Pq_FundUAngB")));
-        baseRealDataSet.setV1AngC(FloatUtils.get2Float(map.get("Pq_FundUAngC")));
+        if (conType == 0) {
+            baseRealDataSet.setV1AngA(FloatUtils.get2Float(map.get("Pq_FundUAngA")));
+            baseRealDataSet.setV1AngB(FloatUtils.get2Float(map.get("Pq_FundUAngB")));
+            baseRealDataSet.setV1AngC(FloatUtils.get2Float(map.get("Pq_FundUAngC")));
+        } else {
+            baseRealDataSet.setV1AngA(FloatUtils.get2Float(map.get("Pq_FundLUAngAB")));
+            baseRealDataSet.setV1AngB(FloatUtils.get2Float(map.get("Pq_FundLUAngBC")));
+            baseRealDataSet.setV1AngC(FloatUtils.get2Float(map.get("Pq_FundLUAngCA")));
+        }
         //基波电流相位
         baseRealDataSet.setI1AngA(FloatUtils.get2Float(map.get("Pq_FundIAngA")));
         baseRealDataSet.setI1AngB(FloatUtils.get2Float(map.get("Pq_FundIAngB")));
         baseRealDataSet.setI1AngC(FloatUtils.get2Float(map.get("Pq_FundIAngC")));
         //电压总谐波畸变率
-        baseRealDataSet.setVThdA(FloatUtils.get2Float(map.get("Pq_ThdUA")));
-        baseRealDataSet.setVThdB(FloatUtils.get2Float(map.get("Pq_ThdUB")));
-        baseRealDataSet.setVThdC(FloatUtils.get2Float(map.get("Pq_ThdUC")));
+        if (conType == 0) {
+            baseRealDataSet.setVThdA(FloatUtils.get2Float(map.get("Pq_ThdUA")));
+            baseRealDataSet.setVThdB(FloatUtils.get2Float(map.get("Pq_ThdUB")));
+            baseRealDataSet.setVThdC(FloatUtils.get2Float(map.get("Pq_ThdUC")));
+        } else {
+            baseRealDataSet.setVThdA(FloatUtils.get2Float(map.get("Pq_ThdLUAB")));
+            baseRealDataSet.setVThdB(FloatUtils.get2Float(map.get("Pq_ThdLUBC")));
+            baseRealDataSet.setVThdC(FloatUtils.get2Float(map.get("Pq_ThdLUCA")));
+        }
         //电流总谐波畸变率
         baseRealDataSet.setIThdA(FloatUtils.get2Float(map.get("Pq_ThdIA")));
         baseRealDataSet.setIThdB(FloatUtils.get2Float(map.get("Pq_ThdIB")));
@@ -276,10 +305,10 @@ public class RtServiceImpl implements IRtService {
         baseRealDataSet.setPfC(FloatUtils.get2Float(map.get("Pq_PFC")));
         baseRealDataSet.setPfTot(FloatUtils.get2Float(map.get("Pq_TotPFM")));
         //基波功率因数
-        baseRealDataSet.setDpfA(FloatUtils.get2Float(map.get("Pq_DPFA")));
-        baseRealDataSet.setDpfB(FloatUtils.get2Float(map.get("Pq_DPFB")));
-        baseRealDataSet.setDpfC(FloatUtils.get2Float(map.get("Pq_DPFC")));
-        baseRealDataSet.setDpfTot(FloatUtils.get2Float(map.get("Pq_TotDPFM")));
+        baseRealDataSet.setDpfA(FloatUtils.get2Float(map.get("Pq_DFA")));
+        baseRealDataSet.setDpfB(FloatUtils.get2Float(map.get("Pq_DFB")));
+        baseRealDataSet.setDpfC(FloatUtils.get2Float(map.get("Pq_DFC")));
+        baseRealDataSet.setDpfTot(FloatUtils.get2Float(map.get("Pq_TotDFM")));
         return baseRealDataSet;
     }
 
@@ -323,6 +352,8 @@ public class RtServiceImpl implements IRtService {
                 }
             } else if (Objects.equals(item.getHarmName(),"Pq_RmsFundU")) {
                 harmRealDataSet.setData1(FloatUtils.get2Float(item.getData()));
+            } else if (Objects.equals(item.getHarmName(),"Pq_ThdU")) {
+                harmRealDataSet.setData1(FloatUtils.get2Float(item.getData()));
             } else {
                 String numberStr = item.getHarmName().substring(item.getHarmName().lastIndexOf('_') + 1);
                 String fieldName = "data" + numberStr;

iot-analysis/analysis-stat/stat-boot/src/main/java/com/njcn/stat/service/impl/StatServiceImpl.java

@@ -57,6 +57,11 @@ public class StatServiceImpl implements IStatService {
     private final RedisUtil redisUtil;
     private final ChannelObjectUtil channelObjectUtil;
     private final CsLineLatestDataFeignClient csLineLatestDataFeignClient;
+    private static final Map<String, String> PHASE_MAPPING = new HashMap<String, String>() {{
+        put("AB", "A");
+        put("BC", "B");
+        put("CA", "C");
+    }};
 
     @Override
     @Transactional(rollbackFor = Exception.class)
@@ -180,7 +185,7 @@ public class StatServiceImpl implements IStatService {
                 }
             }
         }
-        redisUtil.saveByKey(AppRedisKey.LINE_POSITION+id,map);
+        redisUtil.saveByKeyWithExpire(AppRedisKey.LINE_POSITION+id,map,30L);
     }
 
     /**
@@ -216,9 +221,12 @@ public class StatServiceImpl implements IStatService {
             Map<String, String> tags = new HashMap<>();
             tags.put(InfluxDBTableConstant.LINE_ID,lineId);
             tags.put(InfluxDBTableConstant.PHASIC_TYPE,dataArrayList.get(i).getPhase());
+            //todo 不清楚之前为啥要修改相别，这边按字典配置相别无法查询到数据，先改回来
+            //tags.put(InfluxDBTableConstant.PHASIC_TYPE,Objects.isNull(PHASE_MAPPING.get(dataArrayList.get(i).getPhase()))?dataArrayList.get(i).getPhase():PHASE_MAPPING.get(dataArrayList.get(i).getPhase()));
             tags.put(InfluxDBTableConstant.VALUE_TYPE,statMethod);
             tags.put(InfluxDBTableConstant.CL_DID,clDid.toString());
             tags.put(InfluxDBTableConstant.PROCESS,process.toString());
+            tags.put(InfluxDBTableConstant.QUALITY_FLAG,"0");
             Map<String,Object> fields = new HashMap<>();
             //这边特殊处理，如果数据为3.14159，则将数据置为null
             fields.put(dataArrayList.get(i).getName(),Objects.equals(floats.get(i),3.14159f) ? null:floats.get(i));
@@ -243,5 +251,4 @@ public class StatServiceImpl implements IStatService {
         }
         return urlList;
     }
-
 }

iot-analysis/analysis-zl-event/zl-event-api/src/main/java/com/njcn/zlevent/api/EventFeignClient.java

@@ -1,16 +1,10 @@
 package com.njcn.zlevent.api;
 
-import com.njcn.common.pojo.annotation.OperateInfo;
 import com.njcn.common.pojo.constant.ServerInfo;
-import com.njcn.common.pojo.enums.common.LogEnum;
-import com.njcn.common.pojo.enums.response.CommonResponseEnum;
 import com.njcn.common.pojo.response.HttpResult;
-import com.njcn.common.utils.HttpResultUtil;
 import com.njcn.mq.message.AppEventMessage;
 import com.njcn.mq.message.CldLogMessage;
 import com.njcn.zlevent.api.fallback.EventClientFallbackFactory;
-import io.swagger.annotations.ApiImplicitParam;
-import io.swagger.annotations.ApiOperation;
 import org.springframework.cloud.openfeign.FeignClient;
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestBody;

iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/impl/CsEventServiceImpl.java

@@ -11,10 +11,10 @@ import lombok.AllArgsConstructor;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-import java.util.Objects;
+import java.text.SimpleDateFormat;
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.*;
 import java.util.stream.Collectors;
 
 /**
@@ -33,12 +33,19 @@ public class CsEventServiceImpl extends ServiceImpl<CsEventMapper, CsEventPO> im
     @Transactional(rollbackFor = Exception.class)
     public List<String> updateCsEvent(CsEventParam csEventParam) {
         List<String> eventList = new ArrayList<>();
+
+        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSSSSS");
+        LocalDateTime dateTime = LocalDateTime.parse(csEventParam.getStartTime(), formatter);
+        // 减去1毫秒
+        LocalDateTime newDateTime = dateTime.minusNanos(1000000);
+        String startTime = newDateTime.format(formatter);
+
         //1.将波形文件关联事件
         LambdaUpdateWrapper<CsEventPO> lambdaUpdateWrapper = new LambdaUpdateWrapper<>();
         lambdaUpdateWrapper.set(CsEventPO::getWavePath,csEventParam.getPath()).eq(CsEventPO::getLineId,csEventParam.getLineId())
                           .eq(CsEventPO::getDeviceId,csEventParam.getDeviceId())
                           .in(CsEventPO::getType, Arrays.asList(0,1))
-                          .between(CsEventPO::getStartTime,csEventParam.getStartTime(),csEventParam.getEndTime());
+                          .between(CsEventPO::getStartTime,startTime,csEventParam.getEndTime());
         if (Objects.nonNull(csEventParam.getLocation())) {
             lambdaUpdateWrapper.eq(CsEventPO::getLocation, csEventParam.getLocation());
         }

iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/impl/CsWaveAnalysisServiceImpl.java

@@ -184,6 +184,6 @@ public class CsWaveAnalysisServiceImpl implements ICsWaveAnalysisService {
                 map.put(2,item.getLineId());
             }
         }
-        redisUtil.saveByKeyWithExpire(AppRedisKey.LINE_POSITION+id,map,600L);
+        redisUtil.saveByKeyWithExpire(AppRedisKey.LINE_POSITION+id,map,30L);
     }
 }

iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/impl/EventServiceImpl.java

@@ -265,7 +265,7 @@ public class EventServiceImpl implements IEventService {
         po.setStartTime(LocalDateTime.parse(cldLogMessage.getTime(), DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
         po.setTag(cldLogMessage.getLog());
         po.setClDid(1);
-        po.setLevel(3);
+        po.setLevel(channelLevel(cldLogMessage.getGrade()));
         po.setProcess(4);
         po.setCode(cldLogMessage.getCode());
         //前置告警
@@ -289,6 +289,29 @@ public class EventServiceImpl implements IEventService {
         csEventService.save(po);
     }
 
+    /**
+     * 处理告警等级
+     */
+    public int channelLevel(String grade) {
+        int result;
+        switch (grade) {
+            case "DEBUG":
+                result = 4;
+                break;
+            case "WARN":
+                result = 6;
+                break;
+            case "ERROR":
+                result = 7;
+                break;
+            default:
+                result = 5;
+                break;
+        }
+        return result;
+    }
+
+
     /**
      * 处理电压
      * @param vol
@@ -347,7 +370,7 @@ public class EventServiceImpl implements IEventService {
                 map.put(2,item.getLineId());
             }
         }
-        redisUtil.saveByKeyWithExpire(AppRedisKey.LINE_POSITION+id,map,600L);
+        redisUtil.saveByKeyWithExpire(AppRedisKey.LINE_POSITION+id,map,30L);
     }
 
     /**

iot-message/message-boot/src/main/java/com/njcn/message/consumer/BzConsumer.java

@@ -0,0 +1,148 @@
+package com.njcn.message.consumer;
+
+import com.njcn.csdevice.api.CsTerminalReplyFeignClient;
+import com.njcn.csdevice.param.IcdBzReplyParam;
+import com.njcn.middle.rocket.constant.EnhanceMessageConstant;
+import com.njcn.middle.rocket.handler.EnhanceConsumerMessageHandler;
+import com.njcn.mq.constant.BusinessTopic;
+import com.njcn.mq.constant.MessageStatus;
+import com.njcn.mq.message.BzMessage;
+import com.njcn.redis.pojo.enums.AppRedisKey;
+import com.njcn.redis.pojo.enums.RedisKeyEnum;
+import com.njcn.redis.utils.RedisUtil;
+import com.njcn.system.api.RocketMqLogFeignClient;
+import com.njcn.system.pojo.po.RocketmqMsgErrorLog;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.spring.annotation.ConsumeMode;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.Resource;
+import java.util.Objects;
+
+/**
+ * 类的介绍：接收前置响应台账更新相关信息
+ *
+ * @author xuyang
+ * @version 1.0.0
+ * @createTime 2023/8/11 15:32
+ */
+@Service
+@RocketMQMessageListener(
+        topic = BusinessTopic.REPLY_TOPIC,
+        consumerGroup = "RECALL",
+        selectorExpression = "RECALL",
+        consumeMode = ConsumeMode.ORDERLY,
+        consumeThreadNumber = 10,
+        enableMsgTrace = true
+)
+@Slf4j
+public class BzConsumer extends EnhanceConsumerMessageHandler<BzMessage> implements RocketMQListener<BzMessage> {
+
+    @Resource
+    private RedisUtil redisUtil;
+    @Resource
+    private RocketMqLogFeignClient rocketMqLogFeignClient;
+    @Resource
+    private CsTerminalReplyFeignClient csTerminalReplyFeignClient;
+
+    @Override
+    public void handleMessage(BzMessage message) {
+        log.info("分发至补召响应处理程序");
+        //收到消息修改(cs_terminal_reply)
+        IcdBzReplyParam param = new IcdBzReplyParam();
+        param.setId(message.getGuid());
+        param.setDeviceId(message.getTerminalId());
+        param.setLineId(message.getMonitorId());
+        param.setCode(message.getCode());
+        param.setMsg(message.getResult());
+        if (param.getCode() == 200) {
+            param.setState(1);
+        } else {
+            param.setState(2);
+        }
+        csTerminalReplyFeignClient.updateBzData(param);
+    }
+
+    /***
+     * 通过redis分布式锁判断当前消息所处状态
+     * 1、null 查不到该key的数据，属于第一次消费，放行
+     * 2、fail 上次消息消费时发生异常，放行
+     * 3、being processed 正在处理，打回去
+     * 4、success 最近72小时消费成功，避免重复消费，打回去
+     */
+    @Override
+    public boolean filter(BzMessage message) {
+        String keyStatus = redisUtil.getStringByKey(AppRedisKey.RMQ_CONSUME_KEY.concat(message.getKey()));
+        if (Objects.isNull(keyStatus) || keyStatus.equalsIgnoreCase(MessageStatus.FAIL)) {
+            redisUtil.saveByKeyWithExpire(AppRedisKey.RMQ_CONSUME_KEY.concat(message.getKey()), MessageStatus.BEING_PROCESSED, 60L);
+            return false;
+        }
+        return true;
+    }
+
+    /**
+     * 消费成功，缓存到redis72小时，避免重复消费
+     */
+    @Override
+    protected void consumeSuccess(BzMessage message) {
+        redisUtil.saveByKeyWithExpire(AppRedisKey.RMQ_CONSUME_KEY.concat(message.getKey()), MessageStatus.SUCCESS, RedisKeyEnum.ROCKET_MQ_KEY.getTime());
+    }
+
+    /**
+     * 发生异常时，进行错误信息入库保存
+     * 默认没有实现类，子类可以实现该方法，调用feign接口进行入库保存
+     */
+    @Override
+    protected void saveExceptionMsgLog(BzMessage message, String identity, Exception exception) {
+        redisUtil.saveByKeyWithExpire(AppRedisKey.RMQ_CONSUME_KEY.concat(message.getKey()), MessageStatus.FAIL, RedisKeyEnum.ROCKET_MQ_KEY.getTime());
+        RocketmqMsgErrorLog rocketmqMsgErrorLog = new RocketmqMsgErrorLog();
+        rocketmqMsgErrorLog.setMsgKey(message.getKey());
+        rocketmqMsgErrorLog.setResource(message.getSource());
+        if (identity.equalsIgnoreCase(EnhanceMessageConstant.IDENTITY_SINGLE)) {
+            //数据库字段配置长度200，避免插入失败，大致分析异常原因
+            String exceptionMsg = exception.getMessage();
+            if(exceptionMsg.length() > 200){
+                exceptionMsg = exceptionMsg.substring(0,180);
+            }
+            rocketmqMsgErrorLog.setRecord(exceptionMsg);
+            //如果是当前消息重试的则略过
+            if(!message.getSource().startsWith(EnhanceMessageConstant.RETRY_PREFIX)){
+                //单次消费异常
+                rocketMqLogFeignClient.add(rocketmqMsgErrorLog);
+            }
+        } else {
+            rocketmqMsgErrorLog.setRecord("重试消费" + super.getMaxRetryTimes() + "次，依旧消费失败。");
+            //重试N次后，依然消费异常
+            rocketMqLogFeignClient.add(rocketmqMsgErrorLog);
+        }
+    }
+
+    /***
+     * 处理失败后，是否重试
+     * 一般开启
+     */
+    @Override
+    protected boolean isRetry() {
+        return true;
+    }
+
+
+    /***
+     * 消费失败是否抛出异常，抛出异常后就不再消费了
+     */
+    @Override
+    protected boolean throwException() {
+        return false;
+    }
+
+
+    /***
+     * 调用父类handler处理消息的元信息
+     */
+    @Override
+    public void onMessage(BzMessage message) {
+        super.dispatchMessage(message);
+    }
+}

iot-message/message-boot/src/main/java/com/njcn/message/consumer/CldDevRunFlagConsumer.java

@@ -12,6 +12,7 @@ import com.njcn.redis.utils.RedisUtil;
 import com.njcn.system.api.RocketMqLogFeignClient;
 import com.njcn.system.pojo.po.RocketmqMsgErrorLog;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.spring.annotation.ConsumeMode;
 import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
 import org.apache.rocketmq.spring.core.RocketMQListener;
 import org.springframework.stereotype.Service;
@@ -31,6 +32,7 @@ import java.util.Objects;
         topic = BusinessTopic.DEVICE_RUN_FLAG_TOPIC,
         consumerGroup = BusinessTopic.DEVICE_RUN_FLAG_TOPIC,
         consumeThreadNumber = 10,
+        consumeMode = ConsumeMode.ORDERLY,
         enableMsgTrace = true
 )
 @Slf4j
@@ -47,7 +49,7 @@ public class CldDevRunFlagConsumer extends EnhanceConsumerMessageHandler<CldDevi
     protected void handleMessage(CldDeviceRunFlagMessage cldDeviceRunFlagMessage) {
         log.info("分发至翻转设备状态");
         int status = Objects.equals(cldDeviceRunFlagMessage.getStatus(),"0") ? 1 : 2;
-        equipmentFeignClient.flipCldDevStatus(cldDeviceRunFlagMessage.getId(), status);
+        equipmentFeignClient.flipCldDevStatus(cldDeviceRunFlagMessage.getDate(),cldDeviceRunFlagMessage.getId(), status);
     }
 
 

iot-message/message-boot/src/main/java/com/njcn/message/consumer/CldEventConsumer.java

@@ -12,6 +12,7 @@ import com.njcn.system.api.RocketMqLogFeignClient;
 import com.njcn.system.pojo.po.RocketmqMsgErrorLog;
 import com.njcn.zlevent.api.EventFeignClient;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.spring.annotation.ConsumeMode;
 import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
 import org.apache.rocketmq.spring.core.RocketMQListener;
 import org.springframework.stereotype.Service;
@@ -31,6 +32,7 @@ import java.util.Objects;
         topic = BusinessTopic.LOG_TOPIC,
         consumerGroup = BusinessTopic.LOG_TOPIC,
         consumeThreadNumber = 10,
+        consumeMode = ConsumeMode.ORDERLY,
         enableMsgTrace = true
 )
 @Slf4j

iot-message/message-boot/src/main/java/com/njcn/message/consumer/CldHeartConsumer.java

@@ -11,6 +11,7 @@ import com.njcn.redis.utils.RedisUtil;
 import com.njcn.system.api.RocketMqLogFeignClient;
 import com.njcn.system.pojo.po.RocketmqMsgErrorLog;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.spring.annotation.ConsumeMode;
 import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
 import org.apache.rocketmq.spring.core.RocketMQListener;
 import org.springframework.stereotype.Service;
@@ -30,6 +31,7 @@ import java.util.Objects;
         topic = BusinessTopic.HEART_BEAT_TOPIC,
         consumerGroup = BusinessTopic.HEART_BEAT_TOPIC,
         consumeThreadNumber = 10,
+        consumeMode = ConsumeMode.ORDERLY,
         enableMsgTrace = true
 )
 @Slf4j

iot-message/message-boot/src/main/java/com/njcn/message/consumer/RealDataConsumer.java

@@ -47,6 +47,8 @@ public class RealDataConsumer extends EnhanceConsumerMessageHandler<AppAutoDataM
     @Override
     protected void handleMessage(AppAutoDataMessage appAutoDataMessage) {
         log.info("分发至实时数据");
+        String lineId = appAutoDataMessage.getId() + appAutoDataMessage.getMsg().getClDid();
+        redisUtil.saveByKeyWithExpire("devResponse:" + lineId,200,5L);
         rtFeignClient.analysis(appAutoDataMessage);
     }
 

iot-message/message-boot/src/main/java/com/njcn/message/consumer/UpdateLedgerConsumer.java

@@ -0,0 +1,155 @@
+package com.njcn.message.consumer;
+
+import cn.hutool.core.util.ObjectUtil;
+import com.njcn.csdevice.api.CsTerminalReplyFeignClient;
+import com.njcn.csdevice.param.IcdBzReplyParam;
+import com.njcn.middle.rocket.constant.EnhanceMessageConstant;
+import com.njcn.middle.rocket.handler.EnhanceConsumerMessageHandler;
+import com.njcn.mq.constant.BusinessTopic;
+import com.njcn.mq.constant.MessageStatus;
+import com.njcn.mq.message.UpdateLedgerMessage;
+import com.njcn.redis.pojo.enums.AppRedisKey;
+import com.njcn.redis.pojo.enums.RedisKeyEnum;
+import com.njcn.redis.utils.RedisUtil;
+import com.njcn.system.api.RocketMqLogFeignClient;
+import com.njcn.system.pojo.po.RocketmqMsgErrorLog;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.spring.annotation.ConsumeMode;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.Resource;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * 类的介绍：接收前置响应台账更新相关信息
+ *
+ * @author xuyang
+ * @version 1.0.0
+ * @createTime 2023/8/11 15:32
+ */
+@Service
+@RocketMQMessageListener(
+        topic = BusinessTopic.REPLY_TOPIC,
+        consumerGroup = "LEDGER",
+        selectorExpression = "LEDGER",
+        consumeMode = ConsumeMode.ORDERLY,
+        consumeThreadNumber = 10,
+        enableMsgTrace = true
+)
+@Slf4j
+public class UpdateLedgerConsumer extends EnhanceConsumerMessageHandler<UpdateLedgerMessage> implements RocketMQListener<UpdateLedgerMessage> {
+
+    @Resource
+    private RedisUtil redisUtil;
+    @Resource
+    private RocketMqLogFeignClient rocketMqLogFeignClient;
+    @Resource
+    private CsTerminalReplyFeignClient csTerminalReplyFeignClient;
+
+    @Override
+    public void handleMessage(UpdateLedgerMessage updateLedgerMessage) {
+        log.info("分发至更新台账响应处理程序");
+        //收到消息修改(cs_terminal_reply)
+        List<UpdateLedgerMessage.HandleData> data = updateLedgerMessage.getData();
+        if (ObjectUtil.isNotEmpty(data)) {
+            data.forEach(item->{
+                IcdBzReplyParam param = new IcdBzReplyParam();
+                param.setId(updateLedgerMessage.getGuid());
+                param.setDeviceId(item.getDeviceId());
+                param.setCode(item.getCode());
+                param.setMsg(item.getResult());
+                if (item.getCode() == 200) {
+                    param.setState(1);
+                } else {
+                    param.setState(2);
+                }
+                csTerminalReplyFeignClient.updateData(param);
+            });
+        }
+    }
+
+    /***
+     * 通过redis分布式锁判断当前消息所处状态
+     * 1、null 查不到该key的数据，属于第一次消费，放行
+     * 2、fail 上次消息消费时发生异常，放行
+     * 3、being processed 正在处理，打回去
+     * 4、success 最近72小时消费成功，避免重复消费，打回去
+     */
+    @Override
+    public boolean filter(UpdateLedgerMessage message) {
+        String keyStatus = redisUtil.getStringByKey(AppRedisKey.RMQ_CONSUME_KEY.concat(message.getKey()));
+        if (Objects.isNull(keyStatus) || keyStatus.equalsIgnoreCase(MessageStatus.FAIL)) {
+            redisUtil.saveByKeyWithExpire(AppRedisKey.RMQ_CONSUME_KEY.concat(message.getKey()), MessageStatus.BEING_PROCESSED, 60L);
+            return false;
+        }
+        return true;
+    }
+
+    /**
+     * 消费成功，缓存到redis72小时，避免重复消费
+     */
+    @Override
+    protected void consumeSuccess(UpdateLedgerMessage message) {
+        redisUtil.saveByKeyWithExpire(AppRedisKey.RMQ_CONSUME_KEY.concat(message.getKey()), MessageStatus.SUCCESS, RedisKeyEnum.ROCKET_MQ_KEY.getTime());
+    }
+
+    /**
+     * 发生异常时，进行错误信息入库保存
+     * 默认没有实现类，子类可以实现该方法，调用feign接口进行入库保存
+     */
+    @Override
+    protected void saveExceptionMsgLog(UpdateLedgerMessage message, String identity, Exception exception) {
+        redisUtil.saveByKeyWithExpire(AppRedisKey.RMQ_CONSUME_KEY.concat(message.getKey()), MessageStatus.FAIL, RedisKeyEnum.ROCKET_MQ_KEY.getTime());
+        RocketmqMsgErrorLog rocketmqMsgErrorLog = new RocketmqMsgErrorLog();
+        rocketmqMsgErrorLog.setMsgKey(message.getKey());
+        rocketmqMsgErrorLog.setResource(message.getSource());
+        if (identity.equalsIgnoreCase(EnhanceMessageConstant.IDENTITY_SINGLE)) {
+            //数据库字段配置长度200，避免插入失败，大致分析异常原因
+            String exceptionMsg = exception.getMessage();
+            if(exceptionMsg.length() > 200){
+                exceptionMsg = exceptionMsg.substring(0,180);
+            }
+            rocketmqMsgErrorLog.setRecord(exceptionMsg);
+            //如果是当前消息重试的则略过
+            if(!message.getSource().startsWith(EnhanceMessageConstant.RETRY_PREFIX)){
+                //单次消费异常
+                rocketMqLogFeignClient.add(rocketmqMsgErrorLog);
+            }
+        } else {
+            rocketmqMsgErrorLog.setRecord("重试消费" + super.getMaxRetryTimes() + "次，依旧消费失败。");
+            //重试N次后，依然消费异常
+            rocketMqLogFeignClient.add(rocketmqMsgErrorLog);
+        }
+    }
+
+
+    /***
+     * 处理失败后，是否重试
+     * 一般开启
+     */
+    @Override
+    protected boolean isRetry() {
+        return true;
+    }
+
+
+    /***
+     * 消费失败是否抛出异常，抛出异常后就不再消费了
+     */
+    @Override
+    protected boolean throwException() {
+        return false;
+    }
+
+
+    /***
+     * 调用父类handler处理消息的元信息
+     */
+    @Override
+    public void onMessage(UpdateLedgerMessage message) {
+        super.dispatchMessage(message);
+    }
+}

pom.xml

@@ -31,16 +31,47 @@
   <description>物联网交互模块</description>
 
   <properties>
-    <!--中间件目标地址-->
-    <middle.server.url>192.168.1.22</middle.server.url>
-    <!--微服务模块发布地址-->
-    <service.server.url>127.0.0.1</service.server.url>
-    <!--docker仓库地址-->
-    <docker.server.url>192.168.1.22</docker.server.url>
-    <!--nacos的ip:port-->
+
+    <!--103本地-->
+<!--    <middle.server.url>192.168.1.103</middle.server.url>-->
+<!--    <service.server.url>192.168.1.126</service.server.url>-->
+<!--    <docker.server.url>192.168.1.103</docker.server.url>-->
+<!--    <nacos.url>${middle.server.url}:18848</nacos.url>-->
+<!--    <nacos.namespace>72972c43-3c20-4452-a261-66624e17da97</nacos.namespace>-->
+
+    <!--103线上-->
+    <middle.server.url>192.168.1.103</middle.server.url>
+    <service.server.url>192.168.1.103</service.server.url>
+    <docker.server.url>192.168.1.103</docker.server.url>
     <nacos.url>${middle.server.url}:18848</nacos.url>
-    <!--服务器发布内容为空-->
-    <nacos.namespace>415a1c87-33aa-47bd-8e25-13cc456c87ed</nacos.namespace>
+    <nacos.namespace></nacos.namespace>
+
+<!--    <middle.server.url>192.168.1.22</middle.server.url>-->
+<!--    <service.server.url>192.168.1.126</service.server.url>-->
+<!--    <docker.server.url>192.168.1.22</docker.server.url>-->
+<!--    <nacos.url>${middle.server.url}:18848</nacos.url>-->
+<!--    <nacos.namespace>b0b0dedf-baa9-407f-bef6-988b9e0a640d</nacos.namespace>-->
+
+    <!--浙江现场-->
+<!--    <middle.server.url>192.168.4.151</middle.server.url>-->
+<!--    <service.server.url>192.168.4.151</service.server.url>-->
+<!--    <docker.server.url>192.168.4.151</docker.server.url>-->
+<!--    <nacos.url>${middle.server.url}:18848</nacos.url>-->
+<!--    <nacos.namespace></nacos.namespace>-->
+
+    <!--102-->
+<!--    <middle.server.url>192.168.1.102</middle.server.url>-->
+<!--    <service.server.url>127.0.0.1</service.server.url>-->
+<!--    <docker.server.url>192.168.1.102</docker.server.url>-->
+<!--    <nacos.url>${middle.server.url}:18848</nacos.url>-->
+<!--    <nacos.namespace>c208a65e-1578-4372-b7c0-97fecd323fe6</nacos.namespace>-->
+
+<!--    <middle.server.url>192.168.1.27</middle.server.url>-->
+<!--    <service.server.url>127.0.0.1</service.server.url>-->
+<!--    <docker.server.url>192.168.1.27</docker.server.url>-->
+<!--    <nacos.url>${middle.server.url}:18848</nacos.url>-->
+<!--    <nacos.namespace></nacos.namespace>-->
+
     <!--sentinel:port-->
     <sentinel.url>${middle.server.url}:8080</sentinel.url>
     <!--网关地址，主要用于配置swagger中认证token-->