App消息推送调整

pom文件提交
1、便携式设备台账调整
2026-04-07 09:15:55 +08:00 · 2026-04-01 20:30:47 +08:00 · 2026-04-01 20:30:16 +08:00 · 2026-03-26 18:52:02 +08:00 · 2026-02-11 14:10:05 +08:00 · 2026-01-20 16:26:17 +08:00
26 changed files with 676 additions and 236 deletions
--- a/iot-access/access-api/src/main/java/com/njcn/access/param/DevAccessParam.java
+++ b/iot-access/access-api/src/main/java/com/njcn/access/param/DevAccessParam.java
@@ -5,7 +5,6 @@ import lombok.Data;

 import javax.validation.constraints.NotNull;
 import java.io.Serializable;
-import java.math.BigDecimal;
 import java.util.List;

 /**
@@ -18,6 +17,11 @@ import java.util.List;
@Data
 public class DevAccessParam implements Serializable {

+    private static final long serialVersionUID = 1L;
+
+    @ApiModelProperty("工程id")
+    private String engineeringId;
+
    @ApiModelProperty("项目id")
    @NotNull(message = "项目id不能为空")
    private String projectId;
--- a/iot-access/access-api/src/main/java/com/njcn/access/utils/FileCommonUtils.java
+++ b/iot-access/access-api/src/main/java/com/njcn/access/utils/FileCommonUtils.java
@@ -51,6 +51,7 @@ public class FileCommonUtils {
     */
    public void cleanRedisData(String nDid, String fileName) {
        redisUtil.deleteKeysByString("downloadFilePath:"+ nDid);
+        redisUtil.deleteKeysByString("isWeb:"+ nDid);
        redisUtil.delete("fileDowning:"+nDid);
        redisUtil.delete("fileCheck" + nDid + fileName);
        redisUtil.delete(AppRedisKey.FILE_PART_TIME.concat(fileName));
--- a/iot-access/access-boot/src/main/java/com/njcn/access/controller/CsDeviceController.java
+++ b/iot-access/access-boot/src/main/java/com/njcn/access/controller/CsDeviceController.java
@@ -120,4 +120,14 @@ public class CsDeviceController extends BaseController {
        return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, "success", methodDescribe);
    }

+    @OperateInfo(info = LogEnum.BUSINESS_COMMON)
+    @PostMapping("/autoPortableLedger")
+    @ApiOperation("调整便携式设备的台账信息")
+    @ReturnMsg
+    public HttpResult<String> autoPortableLedger(){
+        String methodDescribe = getMethodDescribe("autoPortableLedger");
+        csDeviceService.autoPortableLedger();
+        return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, "success", methodDescribe);
+    }
+
 }
--- a/iot-access/access-boot/src/main/java/com/njcn/access/handler/MqttMessageHandler.java
+++ b/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 这边由于接收文件数据时间跨度会很长，途中有其他请求进来会中断之前的程序，目前是记录中断的位置，等处理完成再继续请求接收文件
--- a/iot-access/access-boot/src/main/java/com/njcn/access/mapper/OverlimitMapper.java
+++ b/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> {
+//
+//}
--- a/iot-access/access-boot/src/main/java/com/njcn/access/service/ICsDeviceService.java
+++ b/iot-access/access-boot/src/main/java/com/njcn/access/service/ICsDeviceService.java
@@ -54,4 +54,6 @@ public interface ICsDeviceService {
     * @param nDid 设备识别码
     */
    void wlAccess(String nDid);
+
+    String autoPortableLedger();
 }
--- a/iot-access/access-boot/src/main/java/com/njcn/access/service/ICsLedgerService.java
+++ b/iot-access/access-boot/src/main/java/com/njcn/access/service/ICsLedgerService.java
@@ -24,4 +24,6 @@ public interface ICsLedgerService extends IService<CsLedger> {
     */
    CsLedger addLedgerTree(CsLedgerParam csLedgerParam);

+    void updatePortableLedger(String engineeringId, String projectId);
+
 }
--- a/iot-access/access-boot/src/main/java/com/njcn/access/service/impl/AskDeviceDataServiceImpl.java
+++ b/iot-access/access-boot/src/main/java/com/njcn/access/service/impl/AskDeviceDataServiceImpl.java
@@ -87,6 +87,7 @@ public class AskDeviceDataServiceImpl implements AskDeviceDataService {
    public boolean downloadFile(String nDid, String name, Integer size, String fileCheck) {
        boolean result = true;
        try {
+            redisUtil.saveByKeyWithExpire("isWeb:"+nDid,name,30L);
            redisUtil.saveByKeyWithExpire("fileDowning:"+nDid,"fileDowning",300L);
            redisUtil.saveByKey("fileCheck"+nDid+name,fileCheck);
            Object object = getDeviceMid(nDid);
@@ -104,6 +105,7 @@ public class AskDeviceDataServiceImpl implements AskDeviceDataService {
            redisUtil.delete("fileDowning:"+nDid);
            redisUtil.delete("fileCheck"+nDid+name);
            redisUtil.delete("fileDownUserId"+nDid+name);
+            redisUtil.deleteKeysByString("isWeb:"+ nDid);
            throw new BusinessException(AlgorithmResponseEnum.FILE_DOWNLOAD_ERROR);
        }
        return result;
--- a/iot-access/access-boot/src/main/java/com/njcn/access/service/impl/CsDevModelServiceImpl.java
+++ b/iot-access/access-boot/src/main/java/com/njcn/access/service/impl/CsDevModelServiceImpl.java
@@ -133,8 +133,11 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
                    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);
@@ -964,7 +967,7 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
        if(CollectionUtil.isNotEmpty(setList)) {
            csDataSetService.addList(setList);
            setList.forEach(item->{
-                if (Objects.equals(item.getName(),"统计数据")) {
+                if (Objects.equals(item.getName(),"Ds$Pqd$Stat$01")) {
                    redisUtil.saveByKeyWithExpire("setId:" + pId,item.getId(),30L);
                }
            });
@@ -1271,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 = "录波记录位置";
--- a/iot-access/access-boot/src/main/java/com/njcn/access/service/impl/CsDeviceServiceImpl.java
+++ b/iot-access/access-boot/src/main/java/com/njcn/access/service/impl/CsDeviceServiceImpl.java
@@ -22,9 +22,7 @@ import com.njcn.common.pojo.enums.response.CommonResponseEnum;
 import com.njcn.common.pojo.exception.BusinessException;
 import com.njcn.csdevice.api.*;
 import com.njcn.csdevice.enums.AlgorithmResponseEnum;
-import com.njcn.csdevice.pojo.param.CsDevModelRelationAddParm;
-import com.njcn.csdevice.pojo.param.CsLedgerParam;
-import com.njcn.csdevice.pojo.param.CsLineParam;
+import com.njcn.csdevice.pojo.param.*;
 import com.njcn.csdevice.pojo.po.*;
 import com.njcn.csdevice.pojo.vo.CsEquipmentDeliveryVO;
 import com.njcn.redis.pojo.enums.AppRedisKey;
@@ -33,12 +31,14 @@ import com.njcn.system.api.DicDataFeignClient;
 import com.njcn.system.api.DictTreeFeignClient;
 import com.njcn.system.enums.DicDataEnum;
 import com.njcn.system.pojo.po.SysDicTreePO;
+import com.njcn.user.api.UserFeignClient;
+import com.njcn.user.enums.AppRoleEnum;
+import com.njcn.user.pojo.vo.UserVO;
 import com.njcn.web.utils.RequestUtil;
 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;

@@ -81,6 +81,10 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
    private final ChannelObjectUtil channelObjectUtil;
    private final CsLineFeignClient csLineFeignClient;
    private final DataSetFeignClient dataSetFeignClient;
+    private final CsMarketDataFeignClient csMarketDataFeignClient;
+    private final UserFeignClient userFeignClient;
+    private final EngineeringFeignClient engineeringFeignClient;
+    private final AppProjectFeignClient appProjectFeignClient;

    @Override
    @Transactional(rollbackFor = {Exception.class})
@@ -131,15 +135,16 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
            throw new BusinessException(AccessResponseEnum.MISSING_CLIENT);
        }
        //5.判断当前流程是否是合法的
-        if (csEquipmentDeliveryVO.getProcess() > type){
-            logDto.setResult(0);
-            logDto.setFailReason(AccessResponseEnum.PROCESS_SAME_ERROR.getMessage());
-            throw new BusinessException(AccessResponseEnum.PROCESS_SAME_ERROR);
-        } else if (csEquipmentDeliveryVO.getProcess() < type){
-            logDto.setResult(0);
-            logDto.setFailReason(AccessResponseEnum.PROCESS_MISSING_ERROR.getMessage());
-            throw new BusinessException(AccessResponseEnum.PROCESS_MISSING_ERROR);
-        }
+        //note(重要说明) 这边流程原先是三个阶段，在实际应用中嫌麻烦，简化为一个流程
+//        if (csEquipmentDeliveryVO.getProcess() > type){
+//            logDto.setResult(0);
+//            logDto.setFailReason(AccessResponseEnum.PROCESS_SAME_ERROR.getMessage());
+//            throw new BusinessException(AccessResponseEnum.PROCESS_SAME_ERROR);
+//        } else if (csEquipmentDeliveryVO.getProcess() < type){
+//            logDto.setResult(0);
+//            logDto.setFailReason(AccessResponseEnum.PROCESS_MISSING_ERROR.getMessage());
+//            throw new BusinessException(AccessResponseEnum.PROCESS_MISSING_ERROR);
+//        }
        //6.询问设备支持的主题信息
        //将支持的主题入库
        askTopic(nDid);
@@ -297,9 +302,9 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
            //8.删除redis监测点模板信息
            redisUtil.delete(AppRedisKey.MODEL + devAccessParam.getNDid());
            redisUtil.delete(AppRedisKey.LINE + devAccessParam.getNDid());
-            //存储日志
+            //9.存储日志
            csLogsFeignClient.addUserLog(logDto);
-            //存储设备调试日志表
+            //10.存储设备调试日志表
            CsEquipmentProcessPO csEquipmentProcess = new CsEquipmentProcessPO();
            csEquipmentProcess.setDevId(devAccessParam.getNDid());
            csEquipmentProcess.setOperator(RequestUtil.getUserIndex());
@@ -310,6 +315,18 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
                csEquipmentProcess.setStatus(1);
            }
            processFeignClient.add(csEquipmentProcess);
+            //11.这里会出现工程用户接入设备时，如果当前工程用户并没有关注，接入之后应该将用户和工程关联起来
+            List<UserVO> users = userFeignClient.getUserVOByIdList(Collections.singletonList(RequestUtil.getUserIndex())).getData();
+            if (CollectionUtil.isNotEmpty(users)) {
+                UserVO userVO = users.get(0);
+                if (CollectionUtil.isNotEmpty(userVO.getRoleCode()) && userVO.getRoleCode().contains(AppRoleEnum.ENGINEERING_USER.getCode())) {
+                    csMarketDataFeignClient.insertData(userVO.getId(), devAccessParam.getEngineeringId());
+                }
+            }
+            //12.如果设备接入，发现接入的工程项目和之前预设的工程项目不一致，则需要更新原来的预设，使用接入的工程项目
+            if (!Objects.equals(vo.getAssociatedEngineering(),devAccessParam.getEngineeringId()) || !Objects.equals(vo.getAssociatedProject(),devAccessParam.getProjectId())) {
+                equipmentFeignClient.updateLedger(devAccessParam.getNDid(), devAccessParam.getEngineeringId(), devAccessParam.getProjectId());
+            }
        } catch (Exception e) {
            logDto.setResult(0);
            logDto.setFailReason(e.getMessage());
@@ -328,7 +345,13 @@ 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);
+//        }
+        param.setProcess(4);
        csEquipmentDeliveryService.devResetFactory(param);
        //清除关系表
        QueryWrapper<CsLedger> csLedgerQueryWrapper = new QueryWrapper<>();
@@ -374,12 +397,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,15 +404,24 @@ 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.录入装置台账信息
+            //note 1、这边发现便携式设备注册时，如果没有工程 项目，后期特殊处理非常的麻烦，这边接入时，先查询工程 项目，如果没有则创建；如果存在则直接使用；
+            //note 2、查询之前已经接入过的便携式设备，如果存在修改台账信息，添加工程、项目
+            String projectId = this.autoPortableLedger();
+            //新增便携式设备
            CsLedgerParam csLedgerParam = new CsLedgerParam();
            csLedgerParam.setId(vo.getId());
-            csLedgerParam.setPid("0");
+            csLedgerParam.setPid(projectId);
            csLedgerParam.setName(vo.getName());
            csLedgerParam.setLevel(2);
            csLedgerParam.setSort(0);
@@ -441,6 +467,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);
@@ -475,6 +504,37 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
        this.devAccessAskTemplate(nDid,version,1);
    }

+    @Override
+    public String autoPortableLedger() {
+        CsEngineeringPO csEngineeringPO = engineeringFeignClient.getEngineeringByName("便携式工程").getData();
+        if (csEngineeringPO == null) {
+            //新增便携式工程
+            CsEngineeringAddParm param = new CsEngineeringAddParm();
+            param.setName("便携式工程");
+            param.setProvince("320000000000");
+            param.setCity("320100000000");
+            param.setDescription("便携式工程");
+            param.setSort(Integer.MAX_VALUE);
+            csEngineeringPO = engineeringFeignClient.addEngineering(param).getData();
+        }
+
+        AppProjectPO csProjectPO = appProjectFeignClient.getProjectByName("便携式项目").getData();
+        if (csProjectPO == null) {
+            //新增便携式项目
+            AppProjectAddParm param = new AppProjectAddParm();
+            param.setEngineeringId(csEngineeringPO.getId());
+            param.setName("便携式项目");
+            param.setArea("园区");
+            param.setDescription("便携式项目");
+            param.setTopoIds(Collections.singletonList("99ed9b9c8cf9007cc4d2ac4c7127b7e4"));
+            param.setSort(Integer.MAX_VALUE);
+            csProjectPO = appProjectFeignClient.addAppProject(param).getData();
+        }
+        //修改已存在的便携式设备
+        csLedgerService.updatePortableLedger(csEngineeringPO.getId(),csProjectPO.getId());
+        return csProjectPO.getId();
+    }
+
    private void checkDeviceStatus(String nDid) {
        DeviceLogDTO logDto = createLogDto("当前设备"+nDid+"状态判断");
        CsEquipmentDeliveryVO csEquipmentDeliveryVO = csEquipmentDeliveryService.queryEquipmentBynDid(nDid);
--- a/iot-access/access-boot/src/main/java/com/njcn/access/service/impl/CsEquipmentDeliveryServiceImpl.java
+++ b/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()
--- a/iot-access/access-boot/src/main/java/com/njcn/access/service/impl/CsLedgerServiceImpl.java
+++ b/iot-access/access-boot/src/main/java/com/njcn/access/service/impl/CsLedgerServiceImpl.java
@@ -1,5 +1,7 @@
 package com.njcn.access.service.impl;

+import cn.hutool.core.collection.CollectionUtil;
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
 import com.njcn.access.mapper.CsLedgerMapper;
 import com.njcn.access.service.ICsLedgerService;
@@ -8,8 +10,11 @@ import com.njcn.csdevice.pojo.po.CsLedger;
 import lombok.AllArgsConstructor;
 import org.springframework.beans.BeanUtils;
 import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;

+import java.util.List;
 import java.util.Objects;
+import java.util.stream.Collectors;

 /**
 * <p>
@@ -43,4 +48,35 @@ public class CsLedgerServiceImpl extends ServiceImpl<CsLedgerMapper, CsLedger> i
        return csLedger;
    }

+    @Override
+    @Transactional(rollbackFor = {Exception.class})
+    public void updatePortableLedger(String engineeringId, String projectId) {
+        //先查询有没有历史的便携式设备
+        List<CsLedger> portableDevices = this.list(
+                new LambdaQueryWrapper<CsLedger>()
+                        .eq(CsLedger::getPid, "0")
+                        .eq(CsLedger::getLevel, 2)
+                        .eq(CsLedger::getState, 1));
+        if (CollectionUtil.isNotEmpty(portableDevices)) {
+            portableDevices.forEach(item->{
+                item.setPid(projectId);
+                item.setPids("0," + engineeringId + "," + projectId);
+            });
+            this.updateBatchById(portableDevices);
+            //获取监测点id
+            List<String> devList = portableDevices.stream().map(CsLedger::getId).collect(Collectors.toList());
+            LambdaQueryWrapper<CsLedger> queryWrapper = new LambdaQueryWrapper<>();
+            queryWrapper.in(CsLedger::getPid, devList).eq(CsLedger::getState,1).eq(CsLedger::getLevel,3);
+            List<CsLedger> pointList = this.list(queryWrapper);
+            if (CollectionUtil.isNotEmpty(pointList)) {
+                pointList.forEach(item->{
+                    String pidS = item.getPids();
+                    String devPid = pidS.split(",")[1];
+                    item.setPids("0," + engineeringId + "," + projectId + "," + devPid);
+                });
+                this.updateBatchById(pointList);
+            }
+        }
+    }
+
 }
--- a/iot-analysis/analysis-rt/rt-boot/src/main/java/com/njcn/rt/service/impl/RtServiceImpl.java
+++ b/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)) {
@@ -225,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")));
@@ -237,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")));
@@ -281,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;
    }

--- a/iot-analysis/analysis-stat/stat-boot/src/main/java/com/njcn/stat/service/impl/StatServiceImpl.java
+++ b/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)
@@ -155,6 +160,7 @@ public class StatServiceImpl implements IStatService {
                csLineLatestDataFeignClient.addData(csLineLatestData);
            }
        }
+        System.gc();
    }

    /**
@@ -180,7 +186,7 @@ public class StatServiceImpl implements IStatService {
                }
            }
        }
-        redisUtil.saveByKey(AppRedisKey.LINE_POSITION+id,map);
+        redisUtil.saveByKeyWithExpire(AppRedisKey.LINE_POSITION+id,map,30L);
    }

    /**
@@ -216,9 +222,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 +252,4 @@ public class StatServiceImpl implements IStatService {
        }
        return urlList;
    }
-
 }
--- a/iot-analysis/analysis-zl-event/zl-event-api/src/main/java/com/njcn/zlevent/pojo/constant/ZlConstant.java
+++ b/iot-analysis/analysis-zl-event/zl-event-api/src/main/java/com/njcn/zlevent/pojo/constant/ZlConstant.java
@@ -41,6 +41,9 @@ public interface ZlConstant {
     */
    String EVT_PARAM_TM = "Evt_Param_Tm";

-
+    /**
+     * 幅值
+     */
+    String EVT_PARAM_VVADEPTH = "Evt_Param_VVaDepth";

 }
--- a/iot-analysis/analysis-zl-event/zl-event-boot/pom.xml
+++ b/iot-analysis/analysis-zl-event/zl-event-boot/pom.xml
@@ -94,6 +94,11 @@
            <artifactId>common-oss</artifactId>
            <version>${project.version}</version>
        </dependency>
+        <dependency>
+            <groupId>com.njcn</groupId>
+            <artifactId>event-common</artifactId>
+            <version>1.0.0</version>
+        </dependency>
    </dependencies>

    <build>
--- a/iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/impl/CsEventServiceImpl.java
+++ b/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());
        }
--- a/iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/impl/CsWaveAnalysisServiceImpl.java
+++ b/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);
    }
 }
--- a/iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/impl/EventServiceImpl.java
+++ b/iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/impl/EventServiceImpl.java
@@ -2,7 +2,9 @@ package com.njcn.zlevent.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.dynamic.datasource.annotation.DSTransactional;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.njcn.common.pojo.exception.BusinessException;
 import com.njcn.csdevice.api.CsLineFeignClient;
@@ -14,6 +16,8 @@ import com.njcn.csdevice.pojo.po.CsLinePO;
 import com.njcn.csdevice.pojo.po.WlRecord;
 import com.njcn.csdevice.pojo.vo.CsEquipmentDeliveryVO;
 import com.njcn.csharmonic.pojo.po.CsEventPO;
+import com.njcn.event.common.mapper.WlRmpEventDetailMapper;
+import com.njcn.event.pojo.po.RmpEventDetailPO;
 import com.njcn.influx.pojo.constant.InfluxDBTableConstant;
 import com.njcn.influx.utils.InfluxDbUtils;
 import com.njcn.mq.message.AppEventMessage;
@@ -27,7 +31,6 @@ import com.njcn.system.enums.DicDataEnum;
 import com.njcn.system.pojo.dto.EpdDTO;
 import com.njcn.system.pojo.po.DictData;
 import com.njcn.zlevent.pojo.constant.ZlConstant;
-import com.njcn.zlevent.pojo.po.CsEventLogs;
 import com.njcn.zlevent.service.ICsEventLogsService;
 import com.njcn.zlevent.service.ICsEventService;
 import com.njcn.zlevent.service.IEventService;
@@ -49,6 +52,7 @@ import java.time.ZoneId;
 import java.time.format.DateTimeFormatter;
 import java.util.*;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;

 /**
 * 类的介绍：
@@ -72,9 +76,10 @@ public class EventServiceImpl implements IEventService {
    private final ICsEventLogsService csEventLogsService;
    private final SendEventUtils sendEventUtils;
    private final WlRecordFeignClient wlRecordFeignClient;
+    private final WlRmpEventDetailMapper wlRmpEventDetailMapper;

    @Override
-    @Transactional(rollbackFor = Exception.class)
+    @DSTransactional
    public void analysis(AppEventMessage appEventMessage) {
        List<CsEventPO> list1 = new ArrayList<>();
        List<String> records = new ArrayList<String>();
@@ -96,12 +101,21 @@ public class EventServiceImpl implements IEventService {
        try {
            if (devModel) {
                if (Objects.equals(appEventMessage.getDid(),1)){
-                    lineId = new Gson().fromJson(String.valueOf(redisUtil.getObjectByKey(AppRedisKey.LINE_POSITION+appEventMessage.getId())), Map.class).get("0").toString();
+                    Object object = new Gson().fromJson(String.valueOf(redisUtil.getObjectByKey(AppRedisKey.LINE_POSITION+appEventMessage.getId())), Map.class).get("0");
+                    if (ObjectUtil.isNotNull(object)) {
+                        lineId = object.toString();
+                    }
                } else if (Objects.equals(appEventMessage.getDid(),2)){
-                    lineId = new Gson().fromJson(String.valueOf(redisUtil.getObjectByKey(AppRedisKey.LINE_POSITION+appEventMessage.getId())), Map.class).get(appEventMessage.getMsg().getClDid().toString()).toString();
+                    Object object = new Gson().fromJson(String.valueOf(redisUtil.getObjectByKey(AppRedisKey.LINE_POSITION+appEventMessage.getId())), Map.class).get(appEventMessage.getMsg().getClDid().toString());
+                    if (ObjectUtil.isNotNull(object)) {
+                        lineId = object.toString();
+                    }
                }
            } else {
-                lineId = new Gson().fromJson(String.valueOf(redisUtil.getObjectByKey(AppRedisKey.LINE_POSITION+appEventMessage.getId())), Map.class).get(appEventMessage.getMsg().getClDid().toString()).toString();
+                Object object = new Gson().fromJson(String.valueOf(redisUtil.getObjectByKey(AppRedisKey.LINE_POSITION+appEventMessage.getId())), Map.class).get(appEventMessage.getMsg().getClDid().toString());
+                if (ObjectUtil.isNotNull(object)) {
+                    lineId = object.toString();
+                }
            }

            //处理事件数据
@@ -111,9 +125,9 @@ public class EventServiceImpl implements IEventService {
                //判断事件是否存在，如果存在则不处理
                LambdaQueryWrapper<CsEventPO> queryWrapper = new LambdaQueryWrapper<>();
                queryWrapper.eq(CsEventPO::getDeviceId,po.getId())
-                        .eq(CsEventPO::getTag,tag)
+                        .eq(CsEventPO::getTag,item.getName())
                        .eq(CsEventPO::getStartTime,eventTime)
-                        .eq(CsEventPO::getLineId,lineId);
+                        .eq(ObjectUtil.isNotNull(lineId),CsEventPO::getLineId,lineId);
                List<CsEventPO> eventList = csEventService.list(queryWrapper);
                if (CollectionUtil.isEmpty(eventList)) {
                    id = IdUtil.fastSimpleUUID();
@@ -150,6 +164,12 @@ public class EventServiceImpl implements IEventService {
                            if (Objects.equals(param.getName(),ZlConstant.EVT_PARAM_TM)){
                                csEvent.setPersistTime(Double.parseDouble(param.getData().toString()));
                            }
+                            if (Objects.equals(param.getName(),ZlConstant.EVT_PARAM_VVADEPTH)) {
+                                csEvent.setAmplitude(Double.parseDouble(param.getData().toString()));
+                            }
+                            if (Objects.equals(param.getName(),"Evt_Param_Phase")) {
+                                csEvent.setPhase(param.getData().toString());
+                            }
                            fields.put(param.getName(),param.getData());
                        }
                        //只有治理型号的设备有监测位置
@@ -169,20 +189,16 @@ public class EventServiceImpl implements IEventService {
                        records.add(batchPoints.lineProtocol());
                    }
                    list1.add(csEvent);
-                    //事件处理日志库
-                    CsEventLogs csEventLogs = new CsEventLogs();
-                    csEventLogs.setLineId(lineId);
-                    csEventLogs.setDeviceId(po.getId());
-                    csEventLogs.setStartTime(timeFormat(item.getDataTimeSec(),item.getDataTimeUSec()));
-                    csEventLogs.setTag(item.getName());
-                    csEventLogs.setStatus(1);
-                    csEventLogs.setTime(LocalDateTime.now());
-                    csEventLogsService.save(csEventLogs);
                }
            }
            //cs_event入库
            if (CollectionUtil.isNotEmpty(list1)){
                csEventService.saveBatch(list1);
+                //同步数据到 r_mp_event_detail 只有暂态事件再同步
+                List<CsEventPO> filterList = list1.stream().filter(csEvent -> Objects.equals(csEvent.getType(), 0)).collect(Collectors.toList());
+                if (CollectionUtil.isNotEmpty(filterList)) {
+                    filterList.forEach(this::insertEvent);
+                }
                //推送事件逻辑处理 && cs_event_user入库
                for (AppEventMessage.DataArray item : dataArray) {
                    sendEventUtils.sendUser(1,item.getType(),po.getId(),item.getName(),eventTime,id,po.getNdid());
@@ -193,18 +209,64 @@ public class EventServiceImpl implements IEventService {
                influxDbUtils.batchInsert(influxDbUtils.getDbName(), "", InfluxDB.ConsistencyLevel.ALL, TimeUnit.MILLISECONDS, records);
            }
        } catch (Exception e) {
-            CsEventLogs csEventLogs = new CsEventLogs();
-            csEventLogs.setLineId(lineId);
-            csEventLogs.setDeviceId(po.getId());
-            csEventLogs.setStartTime(eventTime);
-            csEventLogs.setTag(tag);
-            csEventLogs.setStatus(0);
-            csEventLogs.setTime(LocalDateTime.now());
-            csEventLogs.setRemark(e.getMessage());
-            csEventLogsService.save(csEventLogs);
+            log.error("事件入库异常：{}",e.getMessage());
        }
    }

+    public void insertEvent(CsEventPO item) {
+        RmpEventDetailPO rmpEventDetailPo = new RmpEventDetailPO();
+        rmpEventDetailPo.setEventId(item.getId());
+        rmpEventDetailPo.setMeasurementPointId(item.getLineId());
+        rmpEventDetailPo.setStartTime(item.getStartTime());
+        rmpEventDetailPo.setEventType(getEventType(item.getTag()));
+        rmpEventDetailPo.setFeatureAmplitude(item.getAmplitude());
+        rmpEventDetailPo.setDuration(item.getPersistTime());
+        rmpEventDetailPo.setEventDescribe(getTag(item.getTag()));
+        rmpEventDetailPo.setDealFlag(0);
+        rmpEventDetailPo.setFileFlag(0);
+        rmpEventDetailPo.setPhase(item.getPhase());
+        wlRmpEventDetailMapper.insert(rmpEventDetailPo);
+    }
+
+    public String getEventType(String tag) {
+        switch (tag) {
+            case "Evt_Sys_DipStr":
+                DictData dip = dicDataFeignClient.getDicDataByCode(DicDataEnum.VOLTAGE_DIP.getCode()).getData();
+                tag = dip.getId();
+                break;
+            case "Evt_Sys_SwlStr":
+                DictData rise = dicDataFeignClient.getDicDataByCode(DicDataEnum.VOLTAGE_RISE.getCode()).getData();
+                tag = rise.getId();
+                break;
+            case "Evt_Sys_IntrStr":
+                DictData interruptions = dicDataFeignClient.getDicDataByCode(DicDataEnum.SHORT_INTERRUPTIONS.getCode()).getData();
+                tag = interruptions.getId();
+                break;
+            default:
+                tag = "Un_Know";
+                break;
+        }
+        return tag;
+    }
+
+    public String getTag(String tag) {
+        switch (tag) {
+            case "Evt_Sys_DipStr":
+                tag = DicDataEnum.VOLTAGE_DIP.getCode();
+                break;
+            case "Evt_Sys_SwlStr":
+                tag = DicDataEnum.VOLTAGE_RISE.getCode();
+                break;
+            case "Evt_Sys_IntrStr":
+                tag = DicDataEnum.SHORT_INTERRUPTIONS.getCode();
+                break;
+            default:
+                tag = "Un_Know";
+                break;
+        }
+        return tag;
+    }
+
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void getPortableData(AppEventMessage appEventMessage) {
@@ -265,7 +327,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 +351,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 +432,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);
    }

    /**
--- a/iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/impl/FileServiceImpl.java
+++ b/iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/impl/FileServiceImpl.java
@@ -31,7 +31,6 @@ import com.njcn.oss.constant.OssPath;
 import com.njcn.oss.utils.FileStorageUtil;
 import com.njcn.redis.pojo.enums.AppRedisKey;
 import com.njcn.redis.utils.RedisUtil;
-import com.njcn.user.api.UserFeignClient;
 import com.njcn.zlevent.param.CsEventParam;
 import com.njcn.zlevent.pojo.dto.FileInfoDto;
 import com.njcn.zlevent.pojo.dto.FileStreamDto;
@@ -270,7 +269,7 @@ public class FileServiceImpl implements IFileService {
                        + ",nowStep:" + appFileMessage.getMsg().getFrameCurr()
                        + ",userId:" + userIndex
                        +"}";
-                publisher.send("/Web/Progress/" + appFileMessage.getId(), new Gson().toJson(json), 1, false);
+                publisher.send("/Web/Progress/" + appFileMessage.getId(), new Gson().toJson(json), 2, false);
                if (!Objects.isNull(filePath)){
                    redisUtil.saveByKeyWithExpire("downloadFilePath:" + appFileMessage.getId() + appFileMessage.getMsg().getName(),filePath,60L);
                }
@@ -280,6 +279,7 @@ public class FileServiceImpl implements IFileService {
            //2.缓存了判断收到的报文个数是否和总个数一致，一致则解析文件；不一致则更新缓存
            //3.超时判断: 30s未收到相关文件信息，核查文件个数，看丢失哪些帧，重新请求
            else {
+                Object isWeb = redisUtil.getObjectByKey("isWeb:" + appFileMessage.getId());
                redisUtil.saveByKey("handleEvent:" + appFileMessage.getId(),"doing");
                if (appFileMessage.getMsg().getFrameTotal() == 1){
                    //解析文件入库
@@ -289,14 +289,20 @@ public class FileServiceImpl implements IFileService {
                    csEventLogs.setNowStep(1);
                    csEventLogs.setAllStep(1);
                    csEventLogs.setIsAll(1);
-                    //更新文件信息
-                    csWaveService.updateCsWave(fileName);
-                    //波形文件关联事件
-                    filePath = filePath.replaceAll(GeneralConstant.CFG,"").replaceAll(GeneralConstant.DAT,"");
-                    List<String> eventList = correlateEvents(fileInfoDto,filePath,fileName);
-                    //波形文件解析成图片
-                    if (CollectionUtil.isNotEmpty(eventList) && devModel){
-                        eventList.forEach(wavePicFeignClient::getWavePics);
+                    if (Objects.isNull(isWeb)) {
+                        //更新文件信息
+                        csWaveService.updateCsWave(fileName);
+                        //波形文件关联事件
+                        filePath = filePath.replaceAll(GeneralConstant.CFG,"").replaceAll(GeneralConstant.DAT,"");
+                        List<String> eventList = correlateEvents(fileInfoDto,filePath,fileName);
+                        if (CollectionUtil.isNotEmpty(eventList) && devModel){
+                            eventList.forEach(item -> {
+                                //波形文件解析成图片
+                                wavePicFeignClient.getWavePics(item);
+                                //同步更新r_mp_event_detail，将波形路径录入
+                                wavePicFeignClient.updateEventById(item);
+                            });
+                        }
                    }
                    //解析完删除、处理缓存
                    removeInfoUtils.deleteEventInfo(appFileMessage.getId(),fileName);
@@ -335,14 +341,20 @@ public class FileServiceImpl implements IFileService {
                                csEventLogs.setAllStep(appFileMessage.getMsg().getFrameTotal());
                                csEventLogs.setIsAll(1);
                                log.info("当前文件 {} 帧,这是第 {} 帧报文,全部收到,解析成功!", appFileMessage.getMsg().getFrameTotal(), appFileMessage.getMsg().getFrameCurr());
-                                //修改文件信息
-                                csWaveService.updateCsWave(fileName);
-                                //波形文件关联事件
-                                filePath = filePath.replaceAll(GeneralConstant.CFG, "").replaceAll(GeneralConstant.DAT, "");
-                                List<String> eventList = correlateEvents(fileInfoDto, filePath, fileName);
-                                //波形文件解析成图片
-                                if (CollectionUtil.isNotEmpty(eventList) && devModel) {
-                                    eventList.forEach(wavePicFeignClient::getWavePics);
+                                if (Objects.isNull(isWeb)) {
+                                    //修改文件信息
+                                    csWaveService.updateCsWave(fileName);
+                                    //波形文件关联事件
+                                    filePath = filePath.replaceAll(GeneralConstant.CFG, "").replaceAll(GeneralConstant.DAT, "");
+                                    List<String> eventList = correlateEvents(fileInfoDto, filePath, fileName);
+                                    if (CollectionUtil.isNotEmpty(eventList) && devModel){
+                                        eventList.forEach(item -> {
+                                            //波形文件解析成图片
+                                            wavePicFeignClient.getWavePics(item);
+                                            //同步更新r_mp_event_detail，将波形路径录入
+                                            wavePicFeignClient.updateEventById(item);
+                                        });
+                                    }
                                }
                                redisUtil.delete(AppRedisKey.FILE_PART_TIME.concat(appFileMessage.getMsg().getName()));
                                redisUtil.delete(AppRedisKey.FILE_PART.concat(appFileMessage.getMsg().getName()));
--- a/iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/utils/SendEventUtils.java
+++ b/iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/utils/SendEventUtils.java
@@ -90,7 +90,7 @@ public class SendEventUtils {
                eventName = epdFeignClient.findByName(eventName).getData().getShowName();
                switch (type) {
                    case "1":
-                        code = 2;
+                        code = 3;
                        //设备自身事件 不推送给用户，推送给业务管理
                        eventUser = getEventUser(devId,false);
                        if (CollectionUtil.isNotEmpty(eventUser)) {
@@ -108,7 +108,7 @@ public class SendEventUtils {
                                    userList.add(user.getDevCode());
                                }
                                noticeUserDto.setPushClientId(userList);
-                                noticeUserDto.setTitle("设备事件");
+                                noticeUserDto.setTitle("运行事件");
                            }
                        }
                        break;
@@ -132,26 +132,26 @@ public class SendEventUtils {
                            }
                        }
                        break;
-                    case "3":
-                        code = 1;
-                        //稳态事件
-                        eventUser = getEventUser(devId,true);
-                        if (CollectionUtil.isNotEmpty(eventUser)) {
-                            eventUser.forEach(item->{
-                                CsEventUserPO csEventUser = new CsEventUserPO();
-                                csEventUser.setUserId(item);
-                                csEventUser.setStatus(0);
-                                csEventUser.setEventId(id);
-                                result.add(csEventUser);
-                            });
-                            users = getSendUser(eventUser,1);
-                            if (CollectionUtil.isNotEmpty(users)){
-                                devCodeList = users.stream().map(User::getDevCode).distinct().collect(Collectors.toList());
-                                noticeUserDto.setPushClientId(devCodeList);
-                                noticeUserDto.setTitle("稳态事件");
-                            }
-                        }
-                        break;
+//                    case "3":
+//                        code = 1;
+//                        //稳态事件
+//                        eventUser = getEventUser(devId,true);
+//                        if (CollectionUtil.isNotEmpty(eventUser)) {
+//                            eventUser.forEach(item->{
+//                                CsEventUserPO csEventUser = new CsEventUserPO();
+//                                csEventUser.setUserId(item);
+//                                csEventUser.setStatus(0);
+//                                csEventUser.setEventId(id);
+//                                result.add(csEventUser);
+//                            });
+//                            users = getSendUser(eventUser,1);
+//                            if (CollectionUtil.isNotEmpty(users)){
+//                                devCodeList = users.stream().map(User::getDevCode).distinct().collect(Collectors.toList());
+//                                noticeUserDto.setPushClientId(devCodeList);
+//                                noticeUserDto.setTitle("稳态事件");
+//                            }
+//                        }
+//                        break;
                    default:
                        code = 0;
                        break;
@@ -161,62 +161,62 @@ public class SendEventUtils {
                content = devDetailDto.getEngineeringName() + "-" + devDetailDto.getProjectName() + "-" + devDetailDto.getEquipmentName() + "于" + eventTime.format(DatePattern.NORM_DATETIME_MS_FORMATTER) + "发生" + eventName;
                noticeUserDto.setContent(content);
                payload.setType(code);
-                payload.setPath("/pages/message/message?type="+payload.getType());
-                noticeUserDto.setPayload(payload);
-            }
-            //告警处理
-            else if (eventType == 2){
-                switch (type) {
-                    case "1":
-                        //Ⅰ级告警 不推送给用户，推送给业务管理
-                        eventUser = getEventUser(devId,false);
-                        if (CollectionUtil.isNotEmpty(eventUser)) {
-                            eventUser.forEach(item->{
-                                CsEventUserPO csEventUser = new CsEventUserPO();
-                                csEventUser.setUserId(item);
-                                csEventUser.setStatus(0);
-                                csEventUser.setEventId(id);
-                                result.add(csEventUser);
-                            });
-                            users = getSendUser(eventUser,3);
-                            if (CollectionUtil.isNotEmpty(users)){
-                                eventName = epdFeignClient.findByName(eventName).getData().getShowName();
-                                devCodeList = users.stream().map(User::getDevCode).distinct().collect(Collectors.toList());
-                                noticeUserDto.setPushClientId(devCodeList);
-                            }
-                        }
-                        break;
-                    case "2":
-                        eventName = epdFeignClient.findByName(eventName).getData().getShowName();
-                    case "3":
-                        //Ⅱ、Ⅲ级告警推送相关用户及业务管理员
-                        eventUser = getEventUser(devId,true);
-                        if (CollectionUtil.isNotEmpty(eventUser)) {
-                            eventUser.forEach(item->{
-                                CsEventUserPO csEventUser = new CsEventUserPO();
-                                csEventUser.setUserId(item);
-                                csEventUser.setStatus(0);
-                                csEventUser.setEventId(id);
-                                result.add(csEventUser);
-                            });
-                            users = getSendUser(eventUser,3);
-                            if (CollectionUtil.isNotEmpty(users)){
-                                devCodeList = users.stream().map(User::getDevCode).distinct().collect(Collectors.toList());
-                                noticeUserDto.setPushClientId(devCodeList);
-                            }
-                        }
-                        break;
-                    default:
-                        break;
-                }
-                noticeUserDto.setTitle("告警事件");
-                DevDetailDTO devDetailDto = csLedgerFeignclient.queryDevDetail(devId).getData();
-                content = devDetailDto.getEngineeringName() + "-" + devDetailDto.getProjectName() + "-" + devDetailDto.getEquipmentName() + "于" + eventTime.format(DatePattern.NORM_DATETIME_MS_FORMATTER) + "发生告警,告警信息:" + eventName;
-                noticeUserDto.setContent(content);
-                payload.setType(3);
-                payload.setPath("/pages/message/message?type="+payload.getType());
+                payload.setPath("/pages/index/message1?type="+payload.getType());
                noticeUserDto.setPayload(payload);
            }
+//            //告警处理
+//            else if (eventType == 2){
+//                switch (type) {
+//                    case "1":
+//                        //Ⅰ级告警 不推送给用户，推送给业务管理
+//                        eventUser = getEventUser(devId,false);
+//                        if (CollectionUtil.isNotEmpty(eventUser)) {
+//                            eventUser.forEach(item->{
+//                                CsEventUserPO csEventUser = new CsEventUserPO();
+//                                csEventUser.setUserId(item);
+//                                csEventUser.setStatus(0);
+//                                csEventUser.setEventId(id);
+//                                result.add(csEventUser);
+//                            });
+//                            users = getSendUser(eventUser,3);
+//                            if (CollectionUtil.isNotEmpty(users)){
+//                                eventName = epdFeignClient.findByName(eventName).getData().getShowName();
+//                                devCodeList = users.stream().map(User::getDevCode).distinct().collect(Collectors.toList());
+//                                noticeUserDto.setPushClientId(devCodeList);
+//                            }
+//                        }
+//                        break;
+//                    case "2":
+//                        eventName = epdFeignClient.findByName(eventName).getData().getShowName();
+//                    case "3":
+//                        //Ⅱ、Ⅲ级告警推送相关用户及业务管理员
+//                        eventUser = getEventUser(devId,true);
+//                        if (CollectionUtil.isNotEmpty(eventUser)) {
+//                            eventUser.forEach(item->{
+//                                CsEventUserPO csEventUser = new CsEventUserPO();
+//                                csEventUser.setUserId(item);
+//                                csEventUser.setStatus(0);
+//                                csEventUser.setEventId(id);
+//                                result.add(csEventUser);
+//                            });
+//                            users = getSendUser(eventUser,3);
+//                            if (CollectionUtil.isNotEmpty(users)){
+//                                devCodeList = users.stream().map(User::getDevCode).distinct().collect(Collectors.toList());
+//                                noticeUserDto.setPushClientId(devCodeList);
+//                            }
+//                        }
+//                        break;
+//                    default:
+//                        break;
+//                }
+//                noticeUserDto.setTitle("告警事件");
+//                DevDetailDTO devDetailDto = csLedgerFeignclient.queryDevDetail(devId).getData();
+//                content = devDetailDto.getEngineeringName() + "-" + devDetailDto.getProjectName() + "-" + devDetailDto.getEquipmentName() + "于" + eventTime.format(DatePattern.NORM_DATETIME_MS_FORMATTER) + "发生告警,告警信息:" + eventName;
+//                noticeUserDto.setContent(content);
+//                payload.setType(3);
+//                payload.setPath("/pages/message/message?type="+payload.getType());
+//                noticeUserDto.setPayload(payload);
+//            }
            if (CollectionUtil.isNotEmpty(noticeUserDto.getPushClientId())) {
                List<String> filteredList = noticeUserDto.getPushClientId().stream()
                        .filter(Objects::nonNull)
--- a/iot-message/message-boot/src/main/java/com/njcn/message/consumer/BzConsumer.java
+++ b/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);
+    }
+}
--- a/iot-message/message-boot/src/main/java/com/njcn/message/consumer/CldDevRunFlagConsumer.java
+++ b/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);
    }


--- a/iot-message/message-boot/src/main/java/com/njcn/message/consumer/CldEventConsumer.java
+++ b/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
--- a/iot-message/message-boot/src/main/java/com/njcn/message/consumer/CldHeartConsumer.java
+++ b/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
--- a/iot-message/message-boot/src/main/java/com/njcn/message/consumer/UpdateLedgerConsumer.java
+++ b/iot-message/message-boot/src/main/java/com/njcn/message/consumer/UpdateLedgerConsumer.java
@@ -2,6 +2,7 @@ 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;
@@ -13,10 +14,10 @@ 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 org.springframework.transaction.annotation.Transactional;

 import javax.annotation.Resource;
 import java.util.List;
@@ -32,7 +33,9 @@ import java.util.Objects;
@Service
@RocketMQMessageListener(
        topic = BusinessTopic.REPLY_TOPIC,
-        consumerGroup = BusinessTopic.REPLY_TOPIC,
+        consumerGroup = "LEDGER",
+        selectorExpression = "LEDGER",
+        consumeMode = ConsumeMode.ORDERLY,
        consumeThreadNumber = 10,
        enableMsgTrace = true
 )
@@ -47,18 +50,23 @@ public class UpdateLedgerConsumer extends EnhanceConsumerMessageHandler<UpdateLe
    private CsTerminalReplyFeignClient csTerminalReplyFeignClient;

    @Override
-    @Transactional(rollbackFor = Exception.class)
    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) {
-                    csTerminalReplyFeignClient.updateData(updateLedgerMessage.getGuid(),1,item.getDeviceId());
+                    param.setState(1);
                } else {
-                    csTerminalReplyFeignClient.updateData(updateLedgerMessage.getGuid(),2,item.getDeviceId());
+                    param.setState(2);
                }
+                csTerminalReplyFeignClient.updateData(param);
            });
        }
    }
Author	SHA1	Message	Date
xy	292e8b58a5	App消息推送调整	2026-04-07 09:15:55 +08:00
xy	d20c869bad	pom文件提交	2026-04-01 20:30:47 +08:00
xy	71b9bee182	1、便携式设备台账调整 2、事件入库参数优化	2026-04-01 20:30:16 +08:00
xy	f227fe3c3f	暂态事件同步	2026-03-26 18:52:02 +08:00
xy	56c9c69fc9	添加运行状态统计功能	2026-02-11 14:10:05 +08:00
xy	004de8f307	微调	2026-01-20 16:26:17 +08:00
xy	bda31ce52a	微调	2026-01-20 11:12:57 +08:00
xy	4d1af87153	微调	2026-01-15 15:48:50 +08:00
xy	e34b5ba46e	微调	2026-01-14 20:51:14 +08:00
xy	33b9fae9ef	微调	2026-01-14 20:41:24 +08:00
xy	a369ae6160	微调	2026-01-14 20:32:57 +08:00
xy	6714a6f582	微调	2026-01-13 14:42:10 +08:00
xy	2910770be1	基于云前置系统-新增监测设备的告警等级	2026-01-08 18:39:17 +08:00
xy	cd8cf60683	云前置改造	2025-11-14 10:02:52 +08:00
xy	528f376f6d	云前置改造微调	2025-11-04 14:09:51 +08:00
xy	1d29a03a3c	云前置改造-补召功能调整	2025-10-23 09:41:49 +08:00
xy	9ea6a00cb5	无线接入bug调整	2025-10-16 16:35:52 +08:00
xy	c33490c4fc	云前置改造-暂态数据补召功能	2025-10-15 20:51:16 +08:00