feat(device): 添加谐波计划线路客户端并优化统计缓存键

- 引入 CsHarmonicPlanLineFeignClient 客户端依赖
- 在 CsDeviceServiceImpl 中注入谐波计划线路客户端实例
- 删除监测点稳态指标告警的默认指标配置功能
- 重构统计服务中的缓存键生成逻辑，移除统计方法参数

iot-access/access-api/pom.xml

@@ -20,6 +20,11 @@
     </properties>
 
     <dependencies>
+        <dependency>
+            <groupId>com.njcn</groupId>
+            <artifactId>common-mq</artifactId>
+            <version>${project.version}</version>
+        </dependency>
         <dependency>
             <groupId>com.github.tocrhz</groupId>
             <artifactId>mqtt-spring-boot-starter</artifactId>

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

@@ -0,0 +1,22 @@
+package com.njcn.access.api;
+
+import com.njcn.access.api.fallback.CsHeartbeatClientFallbackFactory;
+import com.njcn.common.pojo.constant.ServerInfo;
+import com.njcn.common.pojo.response.HttpResult;
+import com.njcn.mq.message.HeartbeatTimeoutMessage;
+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;
+
+/**
+ * @author xy
+ */
+@FeignClient(value = ServerInfo.ACCESS_BOOT, path = "/heartbeat", fallbackFactory = CsHeartbeatClientFallbackFactory.class,contextId = "heartbeat")
+public interface CsHeartbeatFeignClient {
+
+    @PostMapping("/handleHeartbeat")
+    @ApiOperation("处理物联设备心跳")
+    HttpResult<String> handleHeartbeat(@RequestBody HeartbeatTimeoutMessage message);
+
+}

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

@@ -0,0 +1,35 @@
+package com.njcn.access.api.fallback;
+
+import com.njcn.access.api.CsHeartbeatFeignClient;
+import com.njcn.common.pojo.enums.response.CommonResponseEnum;
+import com.njcn.common.pojo.exception.BusinessException;
+import com.njcn.common.pojo.response.HttpResult;
+import com.njcn.mq.message.HeartbeatTimeoutMessage;
+import feign.hystrix.FallbackFactory;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+/**
+ * @author xy
+ */
+@Slf4j
+@Component
+public class CsHeartbeatClientFallbackFactory implements FallbackFactory<CsHeartbeatFeignClient> {
+    @Override
+    public CsHeartbeatFeignClient create(Throwable cause) {
+        //判断抛出异常是否为解码器抛出的业务异常
+        Enum<?> exceptionEnum = CommonResponseEnum.SERVICE_FALLBACK;
+        if (cause.getCause() instanceof BusinessException) {
+            BusinessException businessException = (BusinessException) cause.getCause();
+        }
+        Enum<?> finalExceptionEnum = exceptionEnum;
+        return new CsHeartbeatFeignClient() {
+
+            @Override
+            public HttpResult<String> handleHeartbeat(HeartbeatTimeoutMessage message) {
+                log.error("{}异常，降级处理，异常为：{}","处理物联设备心跳数据异常",cause.toString());
+                throw new BusinessException(finalExceptionEnum);
+            }
+        };
+    }
+}

iot-access/access-api/src/main/java/com/njcn/access/param/DevAccessParam.java

@@ -72,6 +72,9 @@ public class DevAccessParam implements Serializable {
         @ApiModelProperty("中心点纬度")
         @NotNull(message = "中心点纬度不能为空")
         private Double lat;
+
+        @ApiModelProperty("拓扑图指标id")
+        private String target;
     }
 
 }

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

@@ -0,0 +1,48 @@
+package com.njcn.access.controller;
+
+import com.njcn.access.service.ICsHeartService;
+import com.njcn.common.pojo.annotation.OperateInfo;
+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.HeartbeatTimeoutMessage;
+import com.njcn.web.controller.BaseController;
+import io.swagger.annotations.Api;
+import io.swagger.annotations.ApiImplicitParam;
+import io.swagger.annotations.ApiOperation;
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+import springfox.documentation.annotations.ApiIgnore;
+
+/**
+ * 类的介绍：
+ *
+ * @author xuyang
+ * @version 1.0.0
+ * @createTime 2023/9/6 11:07
+ */
+@Slf4j
+@RestController
+@RequestMapping("/heartbeat")
+@Api(tags = "心跳")
+@AllArgsConstructor
+@ApiIgnore
+public class CsHeartController extends BaseController {
+
+    private final ICsHeartService csHeartService;
+
+    @OperateInfo(info = LogEnum.BUSINESS_COMMON)
+    @PostMapping("/handleHeartbeat")
+    @ApiOperation("处理物联设备心跳")
+    @ApiImplicitParam(name = "message", value = "message", required = true)
+    public HttpResult<String> handleHeartbeat(@RequestBody HeartbeatTimeoutMessage message){
+        String methodDescribe = getMethodDescribe("handleHeartbeat");
+        csHeartService.handleHeartbeat(message);
+        return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, null, methodDescribe);
+    }
+}

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

@@ -22,10 +22,7 @@ import com.njcn.access.pojo.dto.file.FileRedisDto;
 import com.njcn.access.pojo.param.ReqAndResParam;
 import com.njcn.access.pojo.po.CsLineModel;
 import com.njcn.access.pojo.po.CsTopic;
-import com.njcn.access.service.ICsDeviceOnlineLogsService;
-import com.njcn.access.service.ICsEquipmentDeliveryService;
-import com.njcn.access.service.ICsLineModelService;
-import com.njcn.access.service.ICsTopicService;
+import com.njcn.access.service.*;
 import com.njcn.access.utils.ChannelObjectUtil;
 import com.njcn.common.pojo.dto.DeviceLogDTO;
 import com.njcn.common.pojo.exception.BusinessException;
@@ -59,7 +56,6 @@ import org.springframework.transaction.annotation.Transactional;
 import javax.validation.ConstraintViolation;
 import javax.validation.Validator;
 import java.nio.charset.StandardCharsets;
-import java.time.Instant;
 import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
 import java.time.format.DateTimeFormatterBuilder;
@@ -103,6 +99,7 @@ public class MqttMessageHandler {
     private final WaveFeignClient waveFeignClient;
     private final RtFeignClient rtFeignClient;
     private final CsCommunicateFeignClient csCommunicateFeignClient;
+    private final IHeartbeatService heartbeatService;
     @Autowired
     Validator validator;
 
@@ -328,7 +325,7 @@ public class MqttMessageHandler {
                         //更新电网侧、负载侧监测点信息
                         askDevData(nDid,version,3,(res.getMid()+1));
                         //接入后系统重置装置心跳
-                        redisUtil.saveByKeyWithExpire("MQTT:" + nDid, Instant.now().toEpochMilli(),180L);
+                        heartbeatService.receiveHeartbeat(nDid);
                         //修改redis的mid
                         redisUtil.saveByKey(AppRedisKey.DEVICE_MID + nDid,1);
                         //接入成功标识
@@ -349,7 +346,6 @@ public class MqttMessageHandler {
                     if (!Objects.isNull(rspDataDto.getDataType())) {
                         switch (rspDataDto.getDataType()){
                             case 1:
-                                log.info("{},设备数据应答--->更新设备软件信息", nDid);
                                 logDto.setOperate(nDid + "更新设备软件信息");
                                 RspDataDto.SoftInfo softInfo = JSON.parseObject(JSON.toJSONString(rspDataDto.getDataArray()), RspDataDto.SoftInfo.class);
                                 //记录设备软件信息
@@ -380,7 +376,6 @@ public class MqttMessageHandler {
                                 List<RspDataDto.LdevInfo> devInfo = JSON.parseArray(JSON.toJSONString(rspDataDto.getDataArray()), RspDataDto.LdevInfo.class);
                                 if (CollectionUtil.isNotEmpty(devInfo)){
                                     if (Objects.equals(res.getDid(),1)){
-                                        log.info("{},设备数据应答--->更新治理监测点信息和设备容量", nDid);
                                         List<CsDevCapacityPO> list3 = new ArrayList<>();
                                         boolean hasZeroClDid = devInfo.stream().anyMatch(item -> item.getClDid() == 0);
                                         //治理设备
@@ -409,7 +404,6 @@ public class MqttMessageHandler {
                                             equipmentFeignClient.updateModuleNumber(nDid,(devInfo.size()-1));
                                         }
                                     } else if (Objects.equals(res.getDid(),2)) {
-                                        log.info("{},设备数据应答--->更新电网侧、负载侧监测点信息", nDid);
                                         logDto.setOperate(nDid + "更新电网侧、负载侧监测点信息");
                                         //1.更新电网侧、负载侧监测点相关信息
                                         devInfo.forEach(item->{
@@ -419,14 +413,12 @@ public class MqttMessageHandler {
                                 }
                                 break;
                             case 15:
-                                log.info("{}模块{}:处理实时数据", nDid, rspDataDto.getClDid());
                                 JSONObject jsonObject = JSONObject.parseObject(JSON.toJSONString(res));
                                 AppAutoDataMessage appAutoDataMessage = JSONObject.toJavaObject(jsonObject, AppAutoDataMessage.class);
                                 appAutoDataMessage.setId(nDid);
                                 rtFeignClient.apfRtAnalysis(appAutoDataMessage);
                                 break;
                             case 48:
-                                log.info("询问装置项目列表");
                                 logDto.setUserName("运维管理员");
                                 logDto.setOperate("监测点:" + (nDid + rspDataDto.getClDid()) + "询问项目列表");
                                 List<RspDataDto.ProjectInfo> projectInfoList = JSON.parseArray(JSON.toJSONString(rspDataDto.getDataArray()), RspDataDto.ProjectInfo.class);
@@ -439,7 +431,6 @@ public class MqttMessageHandler {
                     }
                     break;
                 case 4663:
-                    log.info("装置操作应答");
                     if (Objects.equals(res.getCode(),AccessEnum.SUCCESS.getCode())){
                         String key4 = AppRedisKey.CONTROL + nDid;
                         redisUtil.saveByKeyWithExpire(key4,"success",10L);
@@ -492,33 +483,37 @@ public class MqttMessageHandler {
         //响应请求
         switch (res.getType()){
             case 4865:
-                //设置心跳时间，超时改为掉线
-                redisUtil.saveByKeyWithExpire("MQTT:" + nDid, Instant.now().toEpochMilli(),180L);
+                heartbeatService.receiveHeartbeat(nDid);
                 //有心跳，则将装置改成在线
                 //csEquipmentDeliveryService.updateRunStatusBynDid(nDid,AccessEnum.ONLINE.getCode());
-                //处理心跳
-                ReqAndResDto.Res reqAndResParam = new ReqAndResDto.Res();
-                reqAndResParam.setMid(res.getMid());
-                reqAndResParam.setDid(0);
-                reqAndResParam.setPri(AccessEnum.FIRST_CHANNEL.getCode());
-                reqAndResParam.setType(Integer.parseInt(TypeEnum.TYPE_29.getCode()));
-                reqAndResParam.setCode(200);
-                //fixme 前置处理的时间应该是UTC时间，所以需要加8小时。
-                String json = "{Time:"+(System.currentTimeMillis()/1000+8*3600)+"}";
-                net.sf.json.JSONObject jsonObject = net.sf.json.JSONObject.fromObject(json);
-                reqAndResParam.setMsg(jsonObject);
-                publisher.send("/Dev/DataRsp/"+version+"/"+nDid,gson.toJson(reqAndResParam),1,false);
-                //处理业务逻辑
-                Object object = res.getMsg();
-                if (!Objects.isNull(object)){
-                    List<String> abnormalList = new ArrayList<>();
-                    if (object instanceof ArrayList<?>){
-                        abnormalList.addAll((List<String>) object);
+                //处理心跳 判断设备是否接入，如果设备已经接入则响应，不然忽略
+                CsEquipmentDeliveryPO po = equipmentFeignClient.findDevByNDid(nDid).getData();
+                if (Objects.nonNull(po)) {
+                    if (po.getUsageStatus() == 1 && po.getRunStatus() == 2 && po.getStatus() == 3) {
+                        ReqAndResDto.Res reqAndResParam = new ReqAndResDto.Res();
+                        reqAndResParam.setMid(res.getMid());
+                        reqAndResParam.setDid(0);
+                        reqAndResParam.setPri(AccessEnum.FIRST_CHANNEL.getCode());
+                        reqAndResParam.setType(Integer.parseInt(TypeEnum.TYPE_29.getCode()));
+                        reqAndResParam.setCode(200);
+                        //fixme 前置处理的时间应该是UTC时间，所以需要加8小时。
+                        String json = "{Time:"+(System.currentTimeMillis()/1000+8*3600)+"}";
+                        net.sf.json.JSONObject jsonObject = net.sf.json.JSONObject.fromObject(json);
+                        reqAndResParam.setMsg(jsonObject);
+                        publisher.send("/Dev/DataRsp/"+version+"/"+nDid,gson.toJson(reqAndResParam),1,false);
+                        //处理业务逻辑
+                        Object object = res.getMsg();
+                        if (!Objects.isNull(object)){
+                            List<String> abnormalList = new ArrayList<>();
+                            if (object instanceof ArrayList<?>){
+                                abnormalList.addAll((List<String>) object);
+                            }
+                            //todo APF设备不存在逻辑设备掉线的情况，网关下的设备会存在
+                            abnormalList.forEach(item->{
+                                System.out.println("异常设备ID："+item);
+                            });
+                        }
                     }
-                    //todo APF设备不存在逻辑设备掉线的情况，网关下的设备会存在
-                    abnormalList.forEach(item->{
-                        System.out.println("异常设备ID："+item);
-                    });
                 }
                 break;
             case 4866:
@@ -531,15 +526,12 @@ public class MqttMessageHandler {
                     response.setPri(AccessEnum.FIRST_CHANNEL.getCode());
                     response.setType(Integer.parseInt(TypeEnum.TYPE_15.getCode()));
                     response.setCode(200);
-                    log.info("应答事件:{}", new Gson().toJson(response));
                     publisher.send("/Dev/DataRsp/"+version+"/"+nDid,new Gson().toJson(response),1,false);
                 }
                 //判断事件类型
                 switch (dataDto.getMsg().getDataAttr()) {
                     //暂态事件、录波处理、工程信息
                     case 0:
-                        log.info("{}处理事件", nDid);
-                        //log.info(nDid + "事件报文为:" + new String(message.getPayload(), StandardCharsets.UTF_8));
                         EventDto eventDto = gson.fromJson(new String(message.getPayload(), StandardCharsets.UTF_8), EventDto.class);
                         JSONObject jsonObject0 = JSONObject.parseObject(JSON.toJSONString(eventDto));
                         AppEventMessage appEventMessage = JSONObject.toJavaObject(jsonObject0, AppEventMessage.class);
@@ -548,7 +540,6 @@ public class MqttMessageHandler {
                         break;
                     //实时数据
                     case 1:
-                        log.info("{}处理实时数据", nDid);
                         JSONObject jsonObject2 = JSONObject.parseObject(JSON.toJSONString(dataDto));
                         AppAutoDataMessage appAutoDataMessage = JSONObject.toJavaObject(jsonObject2, AppAutoDataMessage.class);
                         appAutoDataMessage.setId(nDid);
@@ -559,9 +550,6 @@ public class MqttMessageHandler {
                         JSONObject jsonObject3 = JSONObject.parseObject(JSON.toJSONString(dataDto));
                         AppAutoDataMessage appAutoDataMessage2 = JSONObject.toJavaObject(jsonObject3, AppAutoDataMessage.class);
                         appAutoDataMessage2.setId(nDid);
-                        appAutoDataMessage2.getMsg().getDataArray().forEach(item->{
-                            log.info("{}处理统计数据{}", nDid, item.getDataAttr());
-                        });
                         appAutoDataMessageTemplate.sendMember(appAutoDataMessage2);
                         break;
                     default:
@@ -593,7 +581,6 @@ public class MqttMessageHandler {
         //响应请求
         switch (fileDto.getType()){
             case 4657:
-                log.info("获取文件信息{}", fileDto);
                 if (Objects.equals(fileDto.getCode(),AccessEnum.SUCCESS.getCode())) {
                     String key = AppRedisKey.PROJECT_INFO + nDid;
                     if (Objects.isNull(fileDto.getMsg().getType())) {
@@ -626,7 +613,6 @@ public class MqttMessageHandler {
                 }
                 break;
             case 4658:
-                log.info("获取文件流信息");
                 FileRedisDto dto = new FileRedisDto();
                 dto.setCode(fileDto.getCode());
                 redisUtil.saveByKeyWithExpire(AppRedisKey.DOWNLOAD + fileDto.getMsg().getName() + fileDto.getMid(),dto,60L);

iot-access/access-boot/src/main/java/com/njcn/access/listener/RedisKeyExpirationListener.java

@@ -1,31 +1,14 @@
 package com.njcn.access.listener;
 
-import cn.hutool.core.collection.CollectionUtil;
-import cn.hutool.core.date.DatePattern;
 import cn.hutool.core.util.ObjectUtil;
 import com.alibaba.nacos.shaded.com.google.gson.Gson;
 import com.github.tocrhz.mqtt.publisher.MqttPublisher;
-import com.njcn.access.enums.AccessEnum;
-import com.njcn.access.pojo.dto.NoticeUserDto;
-import com.njcn.access.service.ICsDeviceOnlineLogsService;
 import com.njcn.access.service.ICsEquipmentDeliveryService;
-import com.njcn.access.service.impl.CsDeviceServiceImpl;
-import com.njcn.access.utils.MqttUtil;
 import com.njcn.access.utils.RedisSetUtil;
 import com.njcn.access.utils.SendMessageUtil;
-import com.njcn.common.pojo.dto.DeviceLogDTO;
 import com.njcn.csdevice.api.*;
-import com.njcn.csdevice.param.DeviceMessageParam;
-import com.njcn.csdevice.pojo.dto.DevDetailDTO;
-import com.njcn.csdevice.pojo.dto.PqsCommunicateDto;
-import com.njcn.csdevice.pojo.po.CsEquipmentDeliveryPO;
-import com.njcn.redis.pojo.enums.AppRedisKey;
 import com.njcn.redis.utils.RedisUtil;
 import com.njcn.rt.pojo.dto.BaseRealDataSet;
-import com.njcn.user.api.AppInfoSetFeignClient;
-import com.njcn.user.api.AppUserFeignClient;
-import com.njcn.user.api.UserFeignClient;
-import com.njcn.user.pojo.po.User;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.springframework.data.redis.connection.Message;
@@ -34,12 +17,7 @@ import org.springframework.data.redis.listener.RedisMessageListenerContainer;
 import org.springframework.stereotype.Component;
 
 import javax.annotation.Resource;
-import java.time.LocalDateTime;
-import java.time.format.DateTimeFormatter;
-import java.util.List;
-import java.util.Objects;
 import java.util.Set;
-import java.util.stream.Collectors;
 
 /**
  * @author xy
@@ -53,24 +31,12 @@ public class RedisKeyExpirationListener extends KeyExpirationEventMessageListene
     @Resource
     private ICsEquipmentDeliveryService csEquipmentDeliveryService;
     @Resource
-    private CsDeviceServiceImpl csDeviceService;
-    @Resource
     private CsLogsFeignClient csLogsFeignClient;
     @Resource
-    private ICsDeviceOnlineLogsService onlineLogsService;
-    @Resource
-    private MqttUtil mqttUtil;
-    @Resource
     private CsLedgerFeignClient csLedgerFeignclient;
     @Resource
     private EquipmentFeignClient equipmentFeignClient;
     @Resource
-    private AppUserFeignClient appUserFeignClient;
-    @Resource
-    private CsDeviceUserFeignClient csDeviceUserFeignClient;
-    @Resource
-    private UserFeignClient userFeignClient;
-    @Resource
     private RedisUtil redisUtil;
     @Resource
     private SendMessageUtil sendMessageUtil;
@@ -81,20 +47,12 @@ public class RedisKeyExpirationListener extends KeyExpirationEventMessageListene
     @Resource
     private RedisSetUtil redisSetUtil;
     @Resource
-    private AppInfoSetFeignClient appInfoSetFeignClient;
-    @Resource
     private DeviceMessageFeignClient deviceMessageFeignClient;
 
-
-    private final Object lock = new Object();
-
     public RedisKeyExpirationListener(RedisMessageListenerContainer listenerContainer) {
         super(listenerContainer);
     }
 
-    //最大告警次数
-    private static int MAX_WARNING_TIMES = 0;
-
     /**
      * 针对redis数据失效事件，进行数据处理
      * 注意message.toString()可以获取失效的key
@@ -106,10 +64,10 @@ public class RedisKeyExpirationListener extends KeyExpirationEventMessageListene
         }
         //判断失效的key是否为MQTT消费端存入的
         String expiredKey = message.toString();
-        if(expiredKey.startsWith("MQTT:")){
-            String nDid = expiredKey.split(":")[1];
-            executeMainTask(nDid);
-        }
+//        if(expiredKey.startsWith("MQTT:")){
+//            String nDid = expiredKey.split(":")[1];
+//            executeMainTask(nDid);
+//        }
         if(expiredKey.startsWith("cldRtDataOverTime:")){
             String lineId = expiredKey.split(":")[1];
             Object redisObject = redisUtil.getObjectByKey("rtDataUserId:"+lineId);
@@ -124,84 +82,77 @@ public class RedisKeyExpirationListener extends KeyExpirationEventMessageListene
                 });
             }
         }
-        //云前置设备心跳丢失处理
-//        if(expiredKey.startsWith(RedisKeyEnum.CLD_HEART_BEAT_KEY.getKey())){
-//            String node = expiredKey.split(":")[1];
-//            String nodeId = node.substring(0, node.length() - 1);
-//            int processNo = Integer.parseInt(node.substring(node.length() - 1));
-//            equipmentFeignClient.updateCldDevStatus(nodeId,processNo);
+    }
+
+//    //主任务
+//    //1.装置心跳断连
+//    //2.MQTT客户端不在线
+//    private void executeMainTask(String nDid) {
+//        log.info("{}->装置离线", nDid);
+//        DeviceLogDTO logDto = new DeviceLogDTO();
+//        logDto.setUserName("运维管理员");
+//        logDto.setLoginName("njcnyw");
+//        //装置下线
+//        csEquipmentDeliveryService.updateRunStatusBynDid(nDid, AccessEnum.OFFLINE.getCode());
+//        //装置调整为注册状态
+//        csEquipmentDeliveryService.updateStatusBynDid(nDid,AccessEnum.REGISTERED.getCode(),null,null);
+//        logDto.setOperate(nDid +"装置离线");
+//        sendMessage(nDid);
+//        //记录装置掉线时间
+//        PqsCommunicateDto dto = new PqsCommunicateDto();
+//        dto.setTime(LocalDateTime.now().format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_PATTERN)));
+//        dto.setDevId(nDid);
+//        dto.setType(0);
+//        dto.setDescription("通讯中断");
+//        csCommunicateFeignClient.insertion(dto);
+//        csLogsFeignClient.addUserLog(logDto);
+//        //清空缓存
+//        redisUtil.deleteKeysByString(AppRedisKey.LINE_POSITION+nDid);
+//    }
+
+//    //判断设备型号发送数据
+//    private void sendMessage(String nDid) {
+//        boolean devModel = equipmentFeignClient.judgeDevModel(nDid).getData();
+//        if (devModel) {
+//            NoticeUserDto dto = sendOffLine(nDid);
+//            if (CollectionUtil.isNotEmpty(dto.getPushClientId())) {
+//                sendMessageUtil.sendEventToUser(dto);
+//                addLogs(dto);
+//            }
 //        }
-    }
+//    }
 
-    //主任务
-    //1.装置心跳断连
-    //2.MQTT客户端不在线
-    private void executeMainTask(String nDid) {
-        log.info("{}->装置离线", nDid);
-        DeviceLogDTO logDto = new DeviceLogDTO();
-        logDto.setUserName("运维管理员");
-        logDto.setLoginName("njcnyw");
-        //装置下线
-        csEquipmentDeliveryService.updateRunStatusBynDid(nDid, AccessEnum.OFFLINE.getCode());
-        //装置调整为注册状态
-        csEquipmentDeliveryService.updateStatusBynDid(nDid,AccessEnum.REGISTERED.getCode(),null,null);
-        logDto.setOperate(nDid +"装置离线");
-        sendMessage(nDid);
-        //记录装置掉线时间
-        PqsCommunicateDto dto = new PqsCommunicateDto();
-        dto.setTime(LocalDateTime.now().format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_PATTERN)));
-        dto.setDevId(nDid);
-        dto.setType(0);
-        dto.setDescription("通讯中断");
-        csCommunicateFeignClient.insertion(dto);
-        csLogsFeignClient.addUserLog(logDto);
-        //清空缓存
-        redisUtil.deleteKeysByString(AppRedisKey.LINE_POSITION+nDid);
-    }
-
-    //判断设备型号发送数据
-    private void sendMessage(String nDid) {
-        boolean devModel = equipmentFeignClient.judgeDevModel(nDid).getData();
-        if (devModel) {
-            NoticeUserDto dto = sendOffLine(nDid);
-            if (CollectionUtil.isNotEmpty(dto.getPushClientId())) {
-                sendMessageUtil.sendEventToUser(dto);
-                addLogs(dto);
-            }
-        }
-    }
-
-    //掉线通知
-    private NoticeUserDto sendOffLine(String nDid) {
-        NoticeUserDto dto = new NoticeUserDto();
-        dto.setTitle("设备离线");
-        CsEquipmentDeliveryPO po = equipmentFeignClient.findDevByNDid(nDid).getData();
-        DevDetailDTO devDetailDto = csLedgerFeignclient.queryDevDetail(po.getId()).getData();
-        DateTimeFormatter fmt = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
-        LocalDateTime localDateTime = LocalDateTime.now();
-        String dateStr = localDateTime.format(fmt);
-        String content = String.format(devDetailDto.getEngineeringName() + "-" + devDetailDto.getProjectName() + "-" + devDetailDto.getEquipmentName() + "于" + dateStr + "离线");
-        dto.setContent(content);
-        //获取设备关联的用户
-        List<String> eventUser = deviceMessageFeignClient.getEventUserByDeviceId(po.getId(),true).getData();
-        DeviceMessageParam param1 = new DeviceMessageParam();
-        param1.setUserList(eventUser);
-        param1.setEventType(2);
-        //获取打开推送的用户
-        List<User> users = deviceMessageFeignClient.getSendUserByType(param1).getData();
-        if (CollectionUtil.isNotEmpty(users)){
-            dto.setPushClientId(
-                    users.stream().filter(Objects::nonNull).map(User::getDevCode).filter(StringUtils::isNotBlank).distinct().collect(Collectors.toList()));
-        }
-        return dto;
-    }
-
-    //日志记录
-    private void addLogs(NoticeUserDto noticeUserDto) {
-        DeviceLogDTO dto = new DeviceLogDTO();
-        dto.setUserName("运维管理员");
-        dto.setLoginName("njcnyw");
-        dto.setOperate(noticeUserDto.getContent());
-        csLogsFeignClient.addUserLog(dto);
-    }
+//    //掉线通知
+//    private NoticeUserDto sendOffLine(String nDid) {
+//        NoticeUserDto dto = new NoticeUserDto();
+//        dto.setTitle("设备离线");
+//        CsEquipmentDeliveryPO po = equipmentFeignClient.findDevByNDid(nDid).getData();
+//        DevDetailDTO devDetailDto = csLedgerFeignclient.queryDevDetail(po.getId()).getData();
+//        DateTimeFormatter fmt = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
+//        LocalDateTime localDateTime = LocalDateTime.now();
+//        String dateStr = localDateTime.format(fmt);
+//        String content = String.format(devDetailDto.getEngineeringName() + "-" + devDetailDto.getProjectName() + "-" + devDetailDto.getEquipmentName() + "于" + dateStr + "离线");
+//        dto.setContent(content);
+//        //获取设备关联的用户
+//        List<String> eventUser = deviceMessageFeignClient.getEventUserByDeviceId(po.getId(),true).getData();
+//        DeviceMessageParam param1 = new DeviceMessageParam();
+//        param1.setUserList(eventUser);
+//        param1.setEventType(2);
+//        //获取打开推送的用户
+//        List<User> users = deviceMessageFeignClient.getSendUserByType(param1).getData();
+//        if (CollectionUtil.isNotEmpty(users)){
+//            dto.setPushClientId(
+//                    users.stream().filter(Objects::nonNull).map(User::getDevCode).filter(StringUtils::isNotBlank).distinct().collect(Collectors.toList()));
+//        }
+//        return dto;
+//    }
+//
+//    //日志记录
+//    private void addLogs(NoticeUserDto noticeUserDto) {
+//        DeviceLogDTO dto = new DeviceLogDTO();
+//        dto.setUserName("运维管理员");
+//        dto.setLoginName("njcnyw");
+//        dto.setOperate(noticeUserDto.getContent());
+//        csLogsFeignClient.addUserLog(dto);
+//    }
 }

iot-access/access-boot/src/main/java/com/njcn/access/runner/AutoAccessTimer.java

@@ -161,74 +161,4 @@ public class AutoAccessTimer implements ApplicationRunner {
         }
     }
 
-//    @Override
-//    public void run(ApplicationArguments args) {
-//        if (scheduler.isShutdown() || scheduler.isTerminated()) {
-//            scheduler = Executors.newScheduledThreadPool(1);
-//        }
-//        Runnable task = () -> {
-//            log.info("轮询定时任务执行中!");
-//            List<CsEquipmentDeliveryPO> list = csEquipmentDeliveryService.getOfflineDev();
-//            if (CollUtil.isNotEmpty(list)) {
-//                ExecutorService executor = Executors.newFixedThreadPool(10);
-//                // 将任务平均分配给10个子列表
-//                List<List<CsEquipmentDeliveryPO>> subLists = CollUtil.split(list, 10);
-//                // 创建一个ExecutorService来处理这些任务
-//                List<Future<Void>> futures = new ArrayList<>();
-//                for (List<CsEquipmentDeliveryPO> subList : subLists) {
-//                    futures.add(executor.submit(() -> {
-//                        try {
-//                            accessDev(subList);
-//                        } catch (Exception e) {
-//                            log.error("处理设备子列表异常，但继续处理其他任务", e);
-//                        }
-//                        return null;
-//                    }));
-//                }
-//                // 等待所有任务完成
-//                for (Future<Void> future : futures) {
-//                    try {
-//                        future.get();
-//                    } catch (InterruptedException e) {
-//                        Thread.currentThread().interrupt();
-//                        log.error("任务被中断", e);
-//                    } catch (ExecutionException e) {
-//                        log.error("任务执行异常", e.getCause());
-//                    } catch (Exception e) {
-//                        log.error("系统异常", e.getCause());
-//                    }
-//                }
-//                // 关闭ExecutorService
-//                executor.shutdown();
-//            }
-//        };
-//        //第一次执行的时间为120s，然后在前一个任务执行完毕后，等待120s再执行下一个任务
-//        scheduler.scheduleWithFixedDelay(task, AUTO_TIME, AUTO_TIME, TimeUnit.SECONDS);
-//    }
-//
-//    public void accessDev(List<CsEquipmentDeliveryPO> list) {
-//        if (CollUtil.isNotEmpty(list)) {
-//            try {
-//                list.forEach(item -> {
-//                    System.out.println(Thread.currentThread().getName() + ": auto : nDid : " + item.getNdid());
-//                    //判断设备类型 便携式设备需要特殊处理 未注册的要先注册、再接入;已注册的直接重新接入
-//                    String code = dictTreeFeignClient.queryById(item.getDevType()).getData().getCode();
-//                    if (Objects.equals(code, DicDataEnum.PORTABLE.getCode()) && Objects.equals(item.getStatus(), 1)) {
-//                        //csDeviceService.wlDevRegister(item.getNdid());
-//                        log.info("请先手动注册、接入");
-//                    } else {
-//                        String version = csTopicService.getVersion(item.getNdid());
-//                        if (Objects.isNull(version)) {
-//                            version = "V1";
-//                        }
-//                        csDeviceService.autoAccess(item.getNdid(), version, 1);
-//                    }
-//                    redisUtil.saveByKey(AppRedisKey.DEVICE_MID + item.getNdid(), 1);
-//                });
-//            } catch (Exception e) {
-//                log.error(e.getMessage());
-//            }
-//        }
-//    }
-
 }

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

@@ -65,6 +65,11 @@ public interface ICsEquipmentDeliveryService extends IService<CsEquipmentDeliver
      */
     List<CsEquipmentDeliveryPO> getOnlineDev();
 
+    /**
+     * 获取启用、系统在线、MQTT接入的装置
+     */
+    List<CsEquipmentDeliveryPO> getUseOnlineDevice();
+
     /**
      * 获取离线、启用、客户端在线的装置
      */

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

@@ -0,0 +1,12 @@
+package com.njcn.access.service;
+
+import com.njcn.mq.message.HeartbeatTimeoutMessage;
+
+/**
+ * @author xy
+ */
+public interface ICsHeartService {
+
+    void handleHeartbeat(HeartbeatTimeoutMessage message);
+
+}

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

@@ -28,5 +28,5 @@ public interface ICsTopicService extends IService<CsTopic> {
      */
     String getVersion(String nDid);
 
-
+    void deleteByNDid(String nDid);
 }

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

@@ -0,0 +1,8 @@
+package com.njcn.access.service;
+
+public interface IHeartbeatService {
+
+    void receiveHeartbeat(String nDid);
+
+    Boolean isHeartbeatUpdated(String nDid, Long sendTime);
+}

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

@@ -399,7 +399,7 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
                                 eleEpdPqdParam.setStatMethod(apf.getStatMethod());
                                 eleEpdPqdParam.setDataType(id);
                                 if (Objects.isNull(apf.getPhase())){
-                                    eleEpdPqdParam.setPhase("M");
+                                    eleEpdPqdParam.setPhase("T");
                                 } else {
                                     eleEpdPqdParam.setPhase(apf.getPhase());
                                 }
@@ -433,7 +433,7 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
                                 eleEpdPqdParam.setEventType(evt.getEventType());
                                 eleEpdPqdParam.setDataType(id);
                                 if (Objects.isNull(evt.getPhase())){
-                                    eleEpdPqdParam.setPhase("M");
+                                    eleEpdPqdParam.setPhase("T");
                                 } else {
                                     eleEpdPqdParam.setPhase(evt.getPhase());
                                 }
@@ -494,7 +494,7 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
                                 //告警code，到时候推送给用户告警码+事件时间
                                 eleEpdPqdParam.setDefaultValue(alm.getCode());
                                 if (Objects.isNull(alm.getPhase())){
-                                    eleEpdPqdParam.setPhase("M");
+                                    eleEpdPqdParam.setPhase("T");
                                 } else {
                                     eleEpdPqdParam.setPhase(alm.getPhase());
                                 }
@@ -521,7 +521,7 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
                                 eleEpdPqdParam.setTranRule(sts.getTranRule());
                                 eleEpdPqdParam.setDataType(id);
                                 if (Objects.isNull(sts.getPhase())){
-                                    eleEpdPqdParam.setPhase("M");
+                                    eleEpdPqdParam.setPhase("T");
                                 } else {
                                     eleEpdPqdParam.setPhase(sts.getPhase());
                                 }
@@ -553,7 +553,7 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
                                 eleEpdPqdParam.setDefaultValue(parm.getDefaultValue());
                                 eleEpdPqdParam.setDataType(id);
                                 if (Objects.isNull(parm.getPhase())){
-                                    eleEpdPqdParam.setPhase("M");
+                                    eleEpdPqdParam.setPhase("T");
                                 } else {
                                     eleEpdPqdParam.setPhase(parm.getPhase());
                                 }
@@ -582,7 +582,7 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
                                 eleEpdPqdParam.setUnit(set.getUnit());
                                 eleEpdPqdParam.setDataType(id);
                                 if (Objects.isNull(set.getPhase())){
-                                    eleEpdPqdParam.setPhase("M");
+                                    eleEpdPqdParam.setPhase("T");
                                 } else {
                                     eleEpdPqdParam.setPhase(set.getPhase());
                                 }
@@ -614,7 +614,7 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
                                 eleEpdPqdParam.setMinNum(ctrl.getMinNum());
                                 eleEpdPqdParam.setDataType(id);
                                 if (Objects.isNull(ctrl.getPhase())){
-                                    eleEpdPqdParam.setPhase("M");
+                                    eleEpdPqdParam.setPhase("T");
                                 } else {
                                     eleEpdPqdParam.setPhase(ctrl.getPhase());
                                 }
@@ -638,7 +638,7 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
                                 eleEpdPqdParam.setSort(epd.getIdx());
                                 eleEpdPqdParam.setType(epd.getType());
                                 if (Objects.isNull(epd.getPhase())){
-                                    eleEpdPqdParam.setPhase("M");
+                                    eleEpdPqdParam.setPhase("T");
                                 } else {
                                     eleEpdPqdParam.setPhase(epd.getPhase());
                                 }
@@ -674,7 +674,7 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
                                 eleEpdPqdParam.setSort(pqd.getIdx());
                                 eleEpdPqdParam.setType(pqd.getType());
                                 if (Objects.isNull(pqd.getPhase())){
-                                    eleEpdPqdParam.setPhase("M");
+                                    eleEpdPqdParam.setPhase("T");
                                 } else {
                                     eleEpdPqdParam.setPhase(pqd.getPhase());
                                 }
@@ -710,7 +710,7 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
                                 eleEpdPqdParam.setSort(bmd.getIdx());
                                 eleEpdPqdParam.setType(bmd.getType());
                                 if (Objects.isNull(bmd.getPhase())){
-                                    eleEpdPqdParam.setPhase("M");
+                                    eleEpdPqdParam.setPhase("T");
                                 } else {
                                     eleEpdPqdParam.setPhase(bmd.getPhase());
                                 }
@@ -741,7 +741,7 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
                                 eleEpdPqdParam.setTranFlag(di.getTranFlag());
                                 eleEpdPqdParam.setTranRule(di.getTranRule());
                                 if (Objects.isNull(di.getPhase())){
-                                    eleEpdPqdParam.setPhase("M");
+                                    eleEpdPqdParam.setPhase("T");
                                 } else {
                                     eleEpdPqdParam.setPhase(di.getPhase());
                                 }
@@ -767,7 +767,7 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
                                 eleEpdPqdParam.setCurSts(dto.getCurSts());
                                 eleEpdPqdParam.setCtlSts(dto.getCtlSts());
                                 if (Objects.isNull(dto.getPhase())){
-                                    eleEpdPqdParam.setPhase("M");
+                                    eleEpdPqdParam.setPhase("T");
                                 } else {
                                     eleEpdPqdParam.setPhase(dto.getPhase());
                                 }
@@ -797,7 +797,7 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
                                 eleEpdPqdParam.setUnit(inSet.getUnit());
                                 eleEpdPqdParam.setDataType(id);
                                 if (Objects.isNull(inSet.getPhase())){
-                                    eleEpdPqdParam.setPhase("M");
+                                    eleEpdPqdParam.setPhase("T");
                                 } else {
                                     eleEpdPqdParam.setPhase(inSet.getPhase());
                                 }
@@ -820,7 +820,7 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
                                 eleEpdPqdParam.setShowName(wave.getName());
                                 eleEpdPqdParam.setSort(wave.getIdx());
                                 eleEpdPqdParam.setDataType(id);
-                                eleEpdPqdParam.setPhase("M");
+                                eleEpdPqdParam.setPhase("T");
                                 eleEpdPqdParam.setClassId(classId);
                                 EleEpdPqd po = epdFeignClient.add(eleEpdPqdParam).getData();
                                 if (CollectionUtil.isNotEmpty(wave.getParam())){
@@ -1025,91 +1025,91 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
                 List<ApfDto> apfList = templateDto.getApfDto();
                 ApfDto apfDto = apfList.get(idx);
                 name = apfDto.getName();
-                phase = apfDto.getPhase() == null ? "M":apfDto.getPhase();
+                phase = apfDto.getPhase() == null ? "T":apfDto.getPhase();
                 break;
             case DataModel.EVT:
                 log.info("查询evt字典数据");
                 List<EvtDto> evtList = templateDto.getEvtDto();
                 EvtDto evtDto = evtList.get(idx);
                 name = evtDto.getName();
-                phase = evtDto.getPhase() == null ? "M":evtDto.getPhase();
+                phase = evtDto.getPhase() == null ? "T":evtDto.getPhase();
                 break;
             case DataModel.ALM:
                 log.info("查询alm字典数据");
                 List<AlmDto> almList = templateDto.getAlmDto();
                 AlmDto almDto = almList.get(idx);
                 name = almDto.getName();
-                phase = almDto.getPhase() == null ? "M":almDto.getPhase();
+                phase = almDto.getPhase() == null ? "T":almDto.getPhase();
                 break;
             case DataModel.STS:
                 log.info("查询sts字典数据");
                 List<StsDto> stsList = templateDto.getStsDto();
                 StsDto stsDto = stsList.get(idx);
                 name = stsDto.getName();
-                phase = stsDto.getPhase() == null ? "M":stsDto.getPhase();
+                phase = stsDto.getPhase() == null ? "T":stsDto.getPhase();
                 break;
             case DataModel.PARM:
                 log.info("查询parm字典数据");
                 List<ParmDto> parmList = templateDto.getParmDto();
                 ParmDto parmDto = parmList.get(idx);
                 name = parmDto.getName();
-                phase = parmDto.getPhase() == null ? "M":parmDto.getPhase();
+                phase = parmDto.getPhase() == null ? "T":parmDto.getPhase();
                 break;
             case DataModel.SET:
                 log.info("查询set字典数据");
                 List<SetDto> setList = templateDto.getSetDto();
                 SetDto setDto = setList.get(idx);
                 name = setDto.getName();
-                phase = setDto.getPhase() == null ? "M":setDto.getPhase();
+                phase = setDto.getPhase() == null ? "T":setDto.getPhase();
                 break;
             case DataModel.CTRL:
                 log.info("查询ctrl字典数据");
                 List<CtrlDto> ctrlList = templateDto.getCtrlDto();
                 CtrlDto ctrlDto = ctrlList.get(idx);
                 name = ctrlDto.getName();
-                phase = ctrlDto.getPhase() == null ? "M":ctrlDto.getPhase();
+                phase = ctrlDto.getPhase() == null ? "T":ctrlDto.getPhase();
                 break;
             case DataModel.EPD:
                 log.info("查询epd字典数据");
                 List<EpdPqdDto> epdList = templateDto.getEpdDto();
                 EpdPqdDto epdDto = epdList.get(idx);
                 name = epdDto.getName();
-                phase = epdDto.getPhase() == null ? "M":epdDto.getPhase();
+                phase = epdDto.getPhase() == null ? "T":epdDto.getPhase();
                 break;
             case DataModel.PQD:
                 log.info("查询pqd字典数据");
                 List<EpdPqdDto> pqdList = templateDto.getPqdDto();
                 EpdPqdDto pqdDto = pqdList.get(idx);
                 name = pqdDto.getName();
-                phase = pqdDto.getPhase() == null ? "M":pqdDto.getPhase();
+                phase = pqdDto.getPhase() == null ? "T":pqdDto.getPhase();
                 break;
             case DataModel.BMD:
                 log.info("查询bmd字典数据");
                 List<BmdDto> bmdList = templateDto.getBmdDto();
                 BmdDto bmdDto = bmdList.get(idx);
                 name = bmdDto.getName();
-                phase = bmdDto.getPhase() == null ? "M":bmdDto.getPhase();
+                phase = bmdDto.getPhase() == null ? "T":bmdDto.getPhase();
                 break;
             case DataModel.DI:
                 log.info("查询di字典数据");
                 List<DiDto> diList = templateDto.getDiDto();
                 DiDto diDto = diList.get(idx);
                 name = diDto.getName();
-                phase = diDto.getPhase() == null ? "M":diDto.getPhase();
+                phase = diDto.getPhase() == null ? "T":diDto.getPhase();
                 break;
             case DataModel.DO:
                 log.info("查询do字典数据");
                 List<DoDto> doList = templateDto.getDoDto();
                 DoDto doDto = doList.get(idx);
                 name = doDto.getName();
-                phase = doDto.getPhase() == null ? "M":doDto.getPhase();
+                phase = doDto.getPhase() == null ? "T":doDto.getPhase();
                 break;
             case DataModel.INSET:
                 log.info("查询inset字典数据");
                 List<InSetDto> inSetList = templateDto.getInSetDto();
                 InSetDto inSetDto = inSetList.get(idx);
                 name = inSetDto.getName();
-                phase = inSetDto.getPhase() == null ? "M":inSetDto.getPhase();
+                phase = inSetDto.getPhase() == null ? "T":inSetDto.getPhase();
                 break;
             default:
                 break;
@@ -1119,7 +1119,7 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
             log.info("指标名称:"+name+",数据类型:"+id+",相别:"+phase);
             throw new BusinessException(AccessResponseEnum.DICT_MISSING);
         }
-        // M 代表没有数据,因为influxDB要录入数据，此字段是主键，给个默认值
+        // T 代表没有数据,因为influxDB要录入数据，此字段是主键，给个默认值
         if (!Objects.isNull(eleEpdPqd.getHarmStart()) && !Objects.isNull(eleEpdPqd.getHarmEnd())){
             if (Objects.equals(eleEpdPqd.getHarmStart(),1)){
                 for (int i = eleEpdPqd.getHarmStart(); i <= eleEpdPqd.getHarmEnd(); i++) {
@@ -1142,7 +1142,7 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
                         CsDataArray.setDataId(eleEpdPqd.getId());
                         CsDataArray.setName(eleEpdPqd.getName() + "_" + i);
                         CsDataArray.setAnotherName((i-0.5) + "次" +eleEpdPqd.getShowName());
-                        CsDataArray.setStatMethod("M");
+                        CsDataArray.setStatMethod("T");
                         CsDataArray.setDataType(eleEpdPqd.getType());
                         CsDataArray.setPhase(eleEpdPqd.getPhase());
                         list.add(CsDataArray);
@@ -1169,7 +1169,7 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
                         CsDataArray.setDataId(eleEpdPqd.getId());
                         CsDataArray.setName(eleEpdPqd.getName() + "_" + i);
                         CsDataArray.setAnotherName(i + "次" +eleEpdPqd.getShowName());
-                        CsDataArray.setStatMethod("M");
+                        CsDataArray.setStatMethod("T");
                         CsDataArray.setDataType(eleEpdPqd.getType());
                         CsDataArray.setPhase(eleEpdPqd.getPhase());
                         list.add(CsDataArray);
@@ -1196,7 +1196,7 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
                 CsDataArray.setDataId(eleEpdPqd.getId());
                 CsDataArray.setName(eleEpdPqd.getName());
                 CsDataArray.setAnotherName(eleEpdPqd.getShowName());
-                CsDataArray.setStatMethod("M");
+                CsDataArray.setStatMethod("T");
                 CsDataArray.setDataType(eleEpdPqd.getType());
                 CsDataArray.setPhase(eleEpdPqd.getPhase());
                 list.add(CsDataArray);

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

@@ -22,15 +22,16 @@ 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.param.LineInfoParam;
 import com.njcn.csdevice.pojo.param.*;
 import com.njcn.csdevice.pojo.po.*;
 import com.njcn.csdevice.pojo.vo.CsEquipmentDeliveryVO;
+import com.njcn.csharmonic.api.CsHarmonicPlanLineFeignClient;
 import com.njcn.redis.pojo.enums.AppRedisKey;
 import com.njcn.redis.utils.RedisUtil;
 import com.njcn.system.api.DicDataFeignClient;
 import com.njcn.system.api.DictTreeFeignClient;
 import com.njcn.system.enums.DicDataEnum;
-import com.njcn.system.pojo.po.DictData;
 import com.njcn.system.pojo.po.SysDicTreePO;
 import com.njcn.user.api.UserFeignClient;
 import com.njcn.user.enums.AppRoleEnum;
@@ -86,6 +87,8 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
     private final UserFeignClient userFeignClient;
     private final EngineeringFeignClient engineeringFeignClient;
     private final AppProjectFeignClient appProjectFeignClient;
+    private final DeviceMessageFeignClient deviceMessageFeignClient;
+    private final CsHarmonicPlanLineFeignClient csHarmonicPlanLineFeignClient;
 
     @Override
     @Transactional(rollbackFor = {Exception.class})
@@ -276,6 +279,7 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
                 appLineTopologyDiagramPo.setLat(item.getLat());
                 appLineTopologyDiagramPo.setLng(item.getLng());
                 appLineTopologyDiagramPo.setStatus("1");
+                appLineTopologyDiagramPo.setTarget(item.getTarget());
                 appLineTopologyDiagramPoList.add(appLineTopologyDiagramPo);
             }
             List<String> position = csLinePoList.stream().map(CsLinePO::getPosition).collect(Collectors.toList());
@@ -286,25 +290,17 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
                 csLogsFeignClient.addUserLog(logDto);
                 throw new BusinessException(AccessResponseEnum.LINE_POSITION_REPEAT);
             }
+            //删除监测点稳态指标告警的默认指标配置
+            List<String> lineIdList = csLinePoList.stream().map(CsLinePO::getLineId).collect(Collectors.toList());
+            csHarmonicPlanLineFeignClient.deleteByLineIds(lineIdList);
             csLineService.saveBatch(csLinePoList);
+
             redisUtil.saveByKeyWithExpire("accessLineInfo:" + devAccessParam.getNDid(),csLinePoList,30L);
             //缓存监测点信息
-            Map<Integer,String> map = new HashMap<>();
-            for (CsLinePO item : csLinePoList) {
-                if (Objects.isNull(item.getPosition())){
-                    map.put(item.getClDid(),item.getLineId());
-                } else {
-                    DictData dictData = dicDataFeignClient.getDicDataById(item.getPosition()).getData();
-                    if (Objects.equals(dictData.getCode(), DicDataEnum.OUTPUT_SIDE.getCode())){
-                        map.put(0,item.getLineId());
-                    } else if (Objects.equals(dictData.getCode(), DicDataEnum.GRID_SIDE.getCode())){
-                        map.put(1,item.getLineId());
-                    } else if (Objects.equals(dictData.getCode(), DicDataEnum.LOAD_SIDE.getCode())){
-                        map.put(2,item.getLineId());
-                    }
-                }
-            }
-            redisUtil.saveByKey(AppRedisKey.LINE_POSITION+devAccessParam.getNDid(),map);
+            LineInfoParam param = new LineInfoParam();
+            param.setNDid(devAccessParam.getNDid());
+            param.setList(csLinePoList);
+            deviceMessageFeignClient.getLineInfo(param);
             //4.监测点拓扑图表录入关系
             appLineTopologyDiagramService.saveBatch(appLineTopologyDiagramPoList);
             //5.绑定装置和人的关系
@@ -399,6 +395,8 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
             appLineTopologyDiagramPOQueryWrapper.in("line_id",collect);
             appLineTopologyDiagramService.remove(appLineTopologyDiagramPOQueryWrapper);
         }
+        //删除topic表
+        csTopicService.deleteByNDid(nDid);
         //清空缓存
         redisUtil.deleteKeysByString(AppRedisKey.LINE_POSITION+nDid);
     }
@@ -477,22 +475,10 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
             csLineService.saveBatch(csLinePoList);
             redisUtil.saveByKeyWithExpire("accessLineInfo:" + nDid,csLinePoList,30L);
             //缓存监测点信息
-            Map<Integer,String> map = new HashMap<>();
-            for (CsLinePO item : csLinePoList) {
-                if (Objects.isNull(item.getPosition())){
-                    map.put(item.getClDid(),item.getLineId());
-                } else {
-                    DictData dictData = dicDataFeignClient.getDicDataById(item.getPosition()).getData();
-                    if (Objects.equals(dictData.getCode(), DicDataEnum.OUTPUT_SIDE.getCode())){
-                        map.put(0,item.getLineId());
-                    } else if (Objects.equals(dictData.getCode(), DicDataEnum.GRID_SIDE.getCode())){
-                        map.put(1,item.getLineId());
-                    } else if (Objects.equals(dictData.getCode(), DicDataEnum.LOAD_SIDE.getCode())){
-                        map.put(2,item.getLineId());
-                    }
-                }
-            }
-            redisUtil.saveByKey(AppRedisKey.LINE_POSITION+nDid,map);
+            LineInfoParam param = new LineInfoParam();
+            param.setNDid(nDid);
+            param.setList(csLinePoList);
+            deviceMessageFeignClient.getLineInfo(param);
             //4.生成装置和模板的关系表
             CsDevModelRelationAddParm csDevModelRelationAddParm = new CsDevModelRelationAddParm();
             csDevModelRelationAddParm.setDevId(vo.getId());
@@ -635,23 +621,10 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
             csLineService.saveBatch(csLinePoList);
             redisUtil.saveByKeyWithExpire("accessLineInfo:" + nDid,csLinePoList,30L);
             //缓存监测点信息
-            //缓存监测点信息
-            Map<Integer,String> map = new HashMap<>();
-            for (CsLinePO item : csLinePoList) {
-                if (Objects.isNull(item.getPosition())){
-                    map.put(item.getClDid(),item.getLineId());
-                } else {
-                    DictData dictData = dicDataFeignClient.getDicDataById(item.getPosition()).getData();
-                    if (Objects.equals(dictData.getCode(), DicDataEnum.OUTPUT_SIDE.getCode())){
-                        map.put(0,item.getLineId());
-                    } else if (Objects.equals(dictData.getCode(), DicDataEnum.GRID_SIDE.getCode())){
-                        map.put(1,item.getLineId());
-                    } else if (Objects.equals(dictData.getCode(), DicDataEnum.LOAD_SIDE.getCode())){
-                        map.put(2,item.getLineId());
-                    }
-                }
-            }
-            redisUtil.saveByKey(AppRedisKey.LINE_POSITION+nDid,map);
+            LineInfoParam param = new LineInfoParam();
+            param.setNDid(nDid);
+            param.setList(csLinePoList);
+            deviceMessageFeignClient.getLineInfo(param);
             //4.生成装置和模板的关系表
             CsDevModelRelationAddParm csDevModelRelationAddParm = new CsDevModelRelationAddParm();
             csDevModelRelationAddParm.setDevId(vo.getId());
@@ -934,32 +907,11 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
                 csDevModelRelationService.addRelation(po);
                 modelMap.put(item.getType(), item.getModelId());
             }
-            List<CsLinePO> lineList;
             Object object = redisUtil.getObjectByKey("accessLineInfo:" + nDid);
             if (Objects.isNull(object)) {
-                Map<Integer,String> map = new HashMap<>();
-                lineList = csLineFeignClient.findByNdid(nDid).getData();
-                for (CsLinePO item : lineList) {
-                    if (item.getClDid() == 0) {
-                        updateLineIds(modelMap.get(0), item.getClDid(), nDid);
-                    } else {
-                        updateLineIds(modelMap.get(1), item.getClDid(), nDid);
-                    }
-                    //缓存监测点信息
-                    if (Objects.isNull(item.getPosition())){
-                        map.put(item.getClDid(),item.getLineId());
-                    } else {
-                        DictData dictData = dicDataFeignClient.getDicDataById(item.getPosition()).getData();
-                        if (Objects.equals(dictData.getCode(), DicDataEnum.OUTPUT_SIDE.getCode())){
-                            map.put(0,item.getLineId());
-                        } else if (Objects.equals(dictData.getCode(), DicDataEnum.GRID_SIDE.getCode())){
-                            map.put(1,item.getLineId());
-                        } else if (Objects.equals(dictData.getCode(), DicDataEnum.LOAD_SIDE.getCode())){
-                            map.put(2,item.getLineId());
-                        }
-                    }
-                }
-                redisUtil.saveByKey(AppRedisKey.LINE_POSITION+nDid,map);
+                LineInfoParam param = new LineInfoParam();
+                param.setNDid(nDid);
+                deviceMessageFeignClient.getLineInfo(param);
             }
             publisher.send("/Pfm/DevCmd/"+version+"/"+nDid, new Gson().toJson(getJson(mid,TypeEnum.TYPE_5.getCode())), 1, false);
             result = true;

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

@@ -17,10 +17,7 @@ import lombok.RequiredArgsConstructor;
 import org.springframework.beans.BeanUtils;
 import org.springframework.stereotype.Service;
 
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-import java.util.Objects;
+import java.util.*;
 import java.util.stream.Collectors;
 
 /**
@@ -126,6 +123,15 @@ public class CsEquipmentDeliveryServiceImpl extends ServiceImpl<CsEquipmentDeliv
         return result;
     }
 
+    @Override
+    public List<CsEquipmentDeliveryPO> getUseOnlineDevice() {
+        return this.lambdaQuery()
+                .eq(CsEquipmentDeliveryPO::getRunStatus,AccessEnum.ONLINE.getCode())
+                .eq(CsEquipmentDeliveryPO::getDevAccessMethod,"MQTT")
+                .eq(CsEquipmentDeliveryPO::getUsageStatus,1)
+                .list();
+    }
+
     @Override
     public List<CsEquipmentDeliveryPO> getOfflineDev() {
         List<CsEquipmentDeliveryPO> result = new ArrayList<>();

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

@@ -0,0 +1,146 @@
+package com.njcn.access.service.impl;
+
+import cn.hutool.core.collection.CollectionUtil;
+import cn.hutool.core.date.DatePattern;
+import com.njcn.access.enums.AccessEnum;
+import com.njcn.access.pojo.dto.NoticeUserDto;
+import com.njcn.access.service.ICsEquipmentDeliveryService;
+import com.njcn.access.service.ICsHeartService;
+import com.njcn.access.service.IHeartbeatService;
+import com.njcn.access.utils.SendMessageUtil;
+import com.njcn.common.pojo.dto.DeviceLogDTO;
+import com.njcn.csdevice.api.*;
+import com.njcn.csdevice.param.DeviceMessageParam;
+import com.njcn.csdevice.pojo.dto.DevDetailDTO;
+import com.njcn.csdevice.pojo.dto.PqsCommunicateDto;
+import com.njcn.csdevice.pojo.po.CsEquipmentDeliveryPO;
+import com.njcn.mq.message.HeartbeatTimeoutMessage;
+import com.njcn.redis.pojo.enums.AppRedisKey;
+import com.njcn.redis.utils.RedisUtil;
+import com.njcn.user.api.AppUserFeignClient;
+import com.njcn.user.api.UserFeignClient;
+import com.njcn.user.pojo.po.User;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.Resource;
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
+/**
+ * <p>
+ * 数据集表 服务实现类
+ * </p>
+ *
+ * @author xuyang
+ * @since 2023-08-01
+ */
+@Service
+@Slf4j
+public class CsHeartServiceImpl implements ICsHeartService {
+
+    @Resource
+    private ICsEquipmentDeliveryService csEquipmentDeliveryService;
+    @Resource
+    private CsLogsFeignClient csLogsFeignClient;
+    @Resource
+    private EquipmentFeignClient equipmentFeignClient;
+    @Resource
+    private SendMessageUtil sendMessageUtil;
+    @Resource
+    private CsLedgerFeignClient csLedgerFeignclient;
+    @Resource
+    private AppUserFeignClient appUserFeignClient;
+    @Resource
+    private CsDeviceUserFeignClient csDeviceUserFeignClient;
+    @Resource
+    private UserFeignClient userFeignClient;
+    @Resource
+    private IHeartbeatService heartbeatService;
+    @Resource
+    private CsCommunicateFeignClient csCommunicateFeignClient;
+    @Resource
+    private RedisUtil redisUtil;
+    @Resource
+    private DeviceMessageFeignClient deviceMessageFeignClient;
+
+    @Override
+    public void handleHeartbeat(HeartbeatTimeoutMessage message) {
+        String nDid = message.getNDid();
+        Long sendTime = message.getTimestamp();
+        if (heartbeatService.isHeartbeatUpdated(nDid, sendTime)) {
+            return;
+        }
+        log.info("{}->装置离线，执行业务处理", nDid);
+        handleDeviceOffline(nDid);
+    }
+
+    private void handleDeviceOffline(String nDid) {
+        DeviceLogDTO logDto = new DeviceLogDTO();
+        logDto.setUserName("运维管理员");
+        logDto.setLoginName("njcnyw");
+        //装置下线
+        csEquipmentDeliveryService.updateRunStatusBynDid(nDid, AccessEnum.OFFLINE.getCode());
+        //装置调整为注册状态
+        csEquipmentDeliveryService.updateStatusBynDid(nDid,AccessEnum.REGISTERED.getCode(),null,null);
+        logDto.setOperate(nDid +"装置离线");
+        sendMessage(nDid);
+        //记录装置掉线时间
+        PqsCommunicateDto dto = new PqsCommunicateDto();
+        dto.setTime(LocalDateTime.now().format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_PATTERN)));
+        dto.setDevId(nDid);
+        dto.setType(0);
+        dto.setDescription("通讯中断");
+        csCommunicateFeignClient.insertion(dto);
+        csLogsFeignClient.addUserLog(logDto);
+        //清空缓存
+        redisUtil.deleteKeysByString(AppRedisKey.LINE_POSITION+nDid);
+    }
+
+    private void sendMessage(String nDid) {
+        boolean devModel = equipmentFeignClient.judgeDevModel(nDid).getData();
+        if (devModel) {
+            NoticeUserDto dto = sendOffLine(nDid);
+            if (CollectionUtil.isNotEmpty(dto.getPushClientId())) {
+                sendMessageUtil.sendEventToUser(dto);
+                addLogs(dto);
+            }
+        }
+    }
+
+    //掉线通知
+    private NoticeUserDto sendOffLine(String nDid) {
+        NoticeUserDto dto = new NoticeUserDto();
+        dto.setTitle("设备离线");
+        CsEquipmentDeliveryPO po = equipmentFeignClient.findDevByNDid(nDid).getData();
+        DevDetailDTO devDetailDto = csLedgerFeignclient.queryDevDetail(po.getId()).getData();
+        DateTimeFormatter fmt = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
+        LocalDateTime localDateTime = LocalDateTime.now();
+        String dateStr = localDateTime.format(fmt);
+        String content = String.format(devDetailDto.getEngineeringName() + "-" + devDetailDto.getProjectName() + "-" + devDetailDto.getEquipmentName() + "于" + dateStr + "离线");
+        dto.setContent(content);
+        //获取设备关联的用户
+        List<String> eventUser = deviceMessageFeignClient.getEventUserByDeviceId(po.getId(),true).getData();
+        DeviceMessageParam param1 = new DeviceMessageParam();
+        param1.setUserList(eventUser);
+        param1.setEventType(2);
+        //获取打开推送的用户
+        List<User> users = deviceMessageFeignClient.getSendUserByType(param1).getData();
+        if (CollectionUtil.isNotEmpty(users)){
+            dto.setPushClientId(
+                    users.stream().filter(Objects::nonNull).map(User::getDevCode).filter(org.apache.commons.lang3.StringUtils::isNotBlank).distinct().collect(Collectors.toList()));
+        }
+        return dto;
+    }
+
+    private void addLogs(NoticeUserDto noticeUserDto) {
+        DeviceLogDTO dto = new DeviceLogDTO();
+        dto.setUserName("运维管理员");
+        dto.setLoginName("njcnyw");
+        dto.setOperate(noticeUserDto.getContent());
+        csLogsFeignClient.addUserLog(dto);
+    }
+}

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

@@ -7,7 +7,6 @@ import com.njcn.access.mapper.CsTopicMapper;
 import com.njcn.access.pojo.po.CsTopic;
 import com.njcn.access.service.ICsTopicService;
 import org.springframework.stereotype.Service;
-import org.springframework.util.CollectionUtils;
 
 import java.util.List;
 
@@ -42,4 +41,11 @@ public class CsTopicServiceImpl extends ServiceImpl<CsTopicMapper, CsTopic> impl
         }
         return version;
     }
+
+    @Override
+    public void deleteByNDid(String nDid) {
+        LambdaQueryWrapper<CsTopic> lambdaQueryWrapper = new LambdaQueryWrapper<>();
+        lambdaQueryWrapper.eq(CsTopic::getNDid,nDid);
+        this.remove(lambdaQueryWrapper);
+    }
 }

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

@@ -0,0 +1,55 @@
+package com.njcn.access.service.impl;
+
+import com.njcn.access.service.IHeartbeatService;
+import com.njcn.mq.message.HeartbeatTimeoutMessage;
+import com.njcn.mq.template.HeartbeatTimeoutMessageTemplate;
+import com.njcn.redis.utils.RedisUtil;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.Resource;
+
+/**
+ * @author xy
+ */
+@Service
+@Slf4j
+public class HeartbeatServiceImpl implements IHeartbeatService {
+
+    @Resource
+    private HeartbeatTimeoutMessageTemplate heartbeatTimeoutMessageTemplate;
+    @Resource
+    private RedisUtil redisUtil;
+    private static final String HEARTBEAT_REDIS_KEY_PREFIX = "HEARTBEAT:";
+    private static final int DELAY_LEVEL_4MIN = 7;
+    private static final long HEARTBEAT_EXPIRE_SECONDS = 180;
+
+    @Override
+    public void receiveHeartbeat(String nDid) {
+        String redisKey = HEARTBEAT_REDIS_KEY_PREFIX + nDid;
+        long currentTime = System.currentTimeMillis();
+
+        redisUtil.saveByKey(redisKey, currentTime);
+        redisUtil.expire(redisKey, HEARTBEAT_EXPIRE_SECONDS);
+
+        HeartbeatTimeoutMessage message = new HeartbeatTimeoutMessage();
+
+        message.setNDid(nDid);
+        message.setTimestamp(currentTime);
+        message.setDelayLevel(DELAY_LEVEL_4MIN);
+        heartbeatTimeoutMessageTemplate.sendMember(message);
+    }
+
+    @Override
+    public Boolean isHeartbeatUpdated(String nDid, Long sendTime) {
+        String redisKey = HEARTBEAT_REDIS_KEY_PREFIX + nDid;
+        Object lastHeartbeat = redisUtil.getObjectByKey(redisKey);
+
+        if (lastHeartbeat == null) {
+            return false;
+        }
+
+        long lastUpdateTime = Long.parseLong(lastHeartbeat.toString());
+        return lastUpdateTime > sendTime;
+    }
+}

iot-access/access-boot/src/main/resources/bootstrap.yml

@@ -45,6 +45,7 @@ logging:
   config: http://@nacos.url@/nacos/v1/cs/configs?tenant=@nacos.namespace@&group=DEFAULT_GROUP&dataId=logback.xml
   level:
     root: info
+    com.njcn.middle.rocket.template.RocketMQEnhanceTemplate: ERROR
 
 
 #mybatis配置信息

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

@@ -78,20 +78,25 @@ public class RtServiceImpl implements IRtService {
         AppAutoDataMessage.DataArray item = appAutoDataMessage.getMsg().getDataArray().get(0);
         //获取设备类型
         CsEquipmentDeliveryPO po1 =equipmentFeignClient.getDevByLineId(lineId).getData();
+        Float ct = po.getCtRatio().floatValue() / (po.getCt2Ratio() == null ? 1.0f:po.getCt2Ratio().floatValue());
+        Float pt = po.getPtRatio().floatValue() / (po.getPt2Ratio() == null ? 1.0f:po.getPt2Ratio().floatValue());
         //fixme 这边先根据数据集的名称来返回对应实体，这边感觉不太合适，后期有好方案再调整
         //基础数据
         if (dataSet.getName().contains("Ds$Pqd$Rt$Basic$")) {
             //用户Id
-            String userId = redisUtil.getObjectByKey("rtDataUserId:"+lineId).toString();
-            BaseRealDataSet baseRealDataSet = assembleData(dataArrayList,item,po.getConType(),po1.getDevAccessMethod());
-            baseRealDataSet.setUserId(userId);
-            baseRealDataSet.setLineId(lineId);
-            baseRealDataSet.setPt(po.getPtRatio().floatValue());
-            baseRealDataSet.setCt(po.getCtRatio().floatValue());
-            baseRealDataSet.setDataLevel(dataSet.getDataLevel());
-            long timestamp = item.getDataTimeSec() - 8*3600;
-            baseRealDataSet.setDataTime(getTime(timestamp));
-            publisher.send("/Web/RealData/" + lineId, new Gson().toJson(baseRealDataSet), 1, false);
+            Object redisObject = redisUtil.getObjectByKey("rtDataUserId:"+lineId);
+            if (ObjectUtil.isNotNull(redisObject)) {
+                String userId = redisObject.toString();
+                BaseRealDataSet baseRealDataSet = assembleData(dataArrayList,item,po.getConType(),po1.getDevAccessMethod());
+                baseRealDataSet.setUserId(userId);
+                baseRealDataSet.setLineId(lineId);
+                baseRealDataSet.setPt(pt);
+                baseRealDataSet.setCt(ct);
+                baseRealDataSet.setDataLevel(dataSet.getDataLevel());
+                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("实时数据") || dataSet.getName().contains("Ds$Pqd$Rt$01")) {
             //用户Id
             Object redisObject = redisUtil.getObjectByKey("rtDataUserId:"+lineId);
@@ -101,8 +106,8 @@ public class RtServiceImpl implements IRtService {
                     BaseRealDataSet baseRealDataSet = assembleData(dataArrayList,item,po.getConType(),po1.getDevAccessMethod());
                     baseRealDataSet.setUserId(userId);
                     baseRealDataSet.setLineId(lineId);
-                    baseRealDataSet.setPt(po.getPtRatio().floatValue());
-                    baseRealDataSet.setCt(po.getCtRatio().floatValue());
+                    baseRealDataSet.setPt(pt);
+                    baseRealDataSet.setCt(ct);
                     baseRealDataSet.setDataLevel(dataSet.getDataLevel());
                     long timestamp = item.getDataTimeSec();
                     baseRealDataSet.setDataTime(getTime(timestamp));
@@ -114,20 +119,23 @@ public class RtServiceImpl implements IRtService {
         else {
             long timestamp;
             //用户Id
-            String userId = redisUtil.getObjectByKey("rtDataUserId:"+lineId).toString();
-            HarmRealDataSet harmRealDataSet = harmData(dataArrayList,item,dataSet.getDataLevel(),po.getCtRatio());
-            harmRealDataSet.setUserId(userId);
-            harmRealDataSet.setLineId(lineId);
-            harmRealDataSet.setPt(po.getPtRatio().floatValue());
-            harmRealDataSet.setCt(po.getCtRatio().floatValue());
-            harmRealDataSet.setDataLevel(dataSet.getDataLevel());
-            if (ObjectUtil.isNotNull(po.getLineNo())) {
-                timestamp = item.getDataTimeSec();
-            } else {
-                timestamp = item.getDataTimeSec() - 8*3600;
+            Object redisObject = redisUtil.getObjectByKey("rtDataUserId:"+lineId);
+            if (ObjectUtil.isNotNull(redisObject)) {
+                String userId = redisObject.toString();
+                HarmRealDataSet harmRealDataSet = harmData(dataArrayList,item,dataSet.getDataLevel(),po.getCtRatio());
+                harmRealDataSet.setUserId(userId);
+                harmRealDataSet.setLineId(lineId);
+                harmRealDataSet.setPt(pt);
+                harmRealDataSet.setCt(ct);
+                harmRealDataSet.setDataLevel(dataSet.getDataLevel());
+                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);
             }
-            harmRealDataSet.setDataTime(getTime(timestamp));
-            publisher.send("/Web/RealData/" + lineId, new Gson().toJson(harmRealDataSet), 1, false);
         }
     }
 
@@ -217,9 +225,9 @@ public class RtServiceImpl implements IRtService {
     public BaseRealDataSet channelData(Map<String,Float> map,Integer conType) {
         BaseRealDataSet baseRealDataSet = new BaseRealDataSet();
         //频率
-        baseRealDataSet.setFreq(map.get("Pq_FreqM"));
+        baseRealDataSet.setFreq(map.get("Pq_FreqT"));
         //频率偏差
-        baseRealDataSet.setFreqDev(map.get("Pq_FreqDevM"));
+        baseRealDataSet.setFreqDev(map.get("Pq_FreqDevT"));
         //判断监测点的接线方式，不同接线方式电压有效值取值不同
         //星型-相电压  角形、V型-线电压
         //电压有效值
@@ -289,43 +297,43 @@ public class RtServiceImpl implements IRtService {
         baseRealDataSet.setIThdB(map.get("Pq_ThdIB"));
         baseRealDataSet.setIThdC(map.get("Pq_ThdIC"));
         //电压不平衡度
-        baseRealDataSet.setVUnbalance(map.get("Pq_UnbalNegUM"));
+        baseRealDataSet.setVUnbalance(map.get("Pq_UnbalNegUT"));
         //电流不平衡度
-        baseRealDataSet.setIUnbalance(map.get("Pq_UnbalNegIM"));
+        baseRealDataSet.setIUnbalance(map.get("Pq_UnbalNegIT"));
         //有功功率
         baseRealDataSet.setPA(map.get("Pq_PA"));
         baseRealDataSet.setPB(map.get("Pq_PB"));
         baseRealDataSet.setPC(map.get("Pq_PC"));
-        baseRealDataSet.setPTot(map.get("Pq_TotPM"));
+        baseRealDataSet.setPTot(map.get("Pq_TotPT"));
         //无功功率
         baseRealDataSet.setQA(map.get("Pq_QA"));
         baseRealDataSet.setQB(map.get("Pq_QB"));
         baseRealDataSet.setQC(map.get("Pq_QC"));
-        baseRealDataSet.setQTot(map.get("Pq_TotQM"));
+        baseRealDataSet.setQTot(map.get("Pq_TotQT"));
         //视在功率
         baseRealDataSet.setSA(map.get("Pq_SA"));
         baseRealDataSet.setSB(map.get("Pq_SB"));
         baseRealDataSet.setSC(map.get("Pq_SC"));
-        baseRealDataSet.setSTot(map.get("Pq_TotSM"));
+        baseRealDataSet.setSTot(map.get("Pq_TotST"));
         //视在功率因数
         baseRealDataSet.setPfA(map.get("Pq_PFA"));
         baseRealDataSet.setPfB(map.get("Pq_PFB"));
         baseRealDataSet.setPfC(map.get("Pq_PFC"));
-        baseRealDataSet.setPfTot(map.get("Pq_TotPFM"));
+        baseRealDataSet.setPfTot(map.get("Pq_TotPFT"));
         //位移功率因数
         baseRealDataSet.setDpfA(map.get("Pq_DFA"));
         baseRealDataSet.setDpfB(map.get("Pq_DFB"));
         baseRealDataSet.setDpfC(map.get("Pq_DFC"));
-        baseRealDataSet.setDpfTot(map.get("Pq_TotDFM"));
+        baseRealDataSet.setDpfTot(map.get("Pq_TotDFT"));
         return baseRealDataSet;
     }
 
     public BaseRealDataSet channelData2(Map<String,Float> map) {
         BaseRealDataSet baseRealDataSet = new BaseRealDataSet();
         //频率
-        baseRealDataSet.setFreq(map.get("Pq_FreqM"));
+        baseRealDataSet.setFreq(map.get("Pq_FreqT"));
         //频率偏差
-        baseRealDataSet.setFreqDev(map.get("Pq_FreqDevM"));
+        baseRealDataSet.setFreqDev(map.get("Pq_FreqDevT"));
         //判断监测点的接线方式，不同接线方式电压有效值取值不同
         //星型-相电压  角形、V型-线电压
         //电压有效值
@@ -365,34 +373,34 @@ public class RtServiceImpl implements IRtService {
         baseRealDataSet.setIThdB(map.get("Pq_ThdIB"));
         baseRealDataSet.setIThdC(map.get("Pq_ThdIC"));
         //电压不平衡度
-        baseRealDataSet.setVUnbalance(map.get("Pq_UnbalNegUM"));
+        baseRealDataSet.setVUnbalance(map.get("Pq_UnbalNegUT"));
         //电流不平衡度
-        baseRealDataSet.setIUnbalance(map.get("Pq_UnbalNegIM"));
+        baseRealDataSet.setIUnbalance(map.get("Pq_UnbalNegIT"));
         //有功功率
         baseRealDataSet.setPA(map.get("Pq_PA"));
         baseRealDataSet.setPB(map.get("Pq_PB"));
         baseRealDataSet.setPC(map.get("Pq_PC"));
-        baseRealDataSet.setPTot(map.get("Pq_TotPM"));
+        baseRealDataSet.setPTot(map.get("Pq_TotPT"));
         //无功功率
         baseRealDataSet.setQA(map.get("Pq_QA"));
         baseRealDataSet.setQB(map.get("Pq_QB"));
         baseRealDataSet.setQC(map.get("Pq_QC"));
-        baseRealDataSet.setQTot(map.get("Pq_TotQM"));
+        baseRealDataSet.setQTot(map.get("Pq_TotQT"));
         //视在功率
         baseRealDataSet.setSA(map.get("Pq_SA"));
         baseRealDataSet.setSB(map.get("Pq_SB"));
         baseRealDataSet.setSC(map.get("Pq_SC"));
-        baseRealDataSet.setSTot(map.get("Pq_TotSM"));
+        baseRealDataSet.setSTot(map.get("Pq_TotST"));
         //视在功率因数
         baseRealDataSet.setPfA(map.get("Pq_PFA"));
         baseRealDataSet.setPfB(map.get("Pq_PFB"));
         baseRealDataSet.setPfC(map.get("Pq_PFC"));
-        baseRealDataSet.setPfTot(map.get("Pq_TotPFM"));
+        baseRealDataSet.setPfTot(map.get("Pq_TotPFT"));
         //位移功率因数
         baseRealDataSet.setDpfA(map.get("Pq_DFA"));
         baseRealDataSet.setDpfB(map.get("Pq_DFB"));
         baseRealDataSet.setDpfC(map.get("Pq_DFC"));
-        baseRealDataSet.setDpfTot(map.get("Pq_TotDFM"));
+        baseRealDataSet.setDpfTot(map.get("Pq_TotDFT"));
         return baseRealDataSet;
     }
 

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

@@ -11,8 +11,10 @@ import com.njcn.access.utils.ChannelObjectUtil;
 import com.njcn.common.pojo.exception.BusinessException;
 import com.njcn.common.utils.PubUtils;
 import com.njcn.csdevice.api.CsCommunicateFeignClient;
+import com.njcn.csdevice.api.CsLineFeignClient;
 import com.njcn.csdevice.api.DataArrayFeignClient;
 import com.njcn.csdevice.api.DeviceMessageFeignClient;
+import com.njcn.csdevice.param.LineInfoParam;
 import com.njcn.csdevice.pojo.dto.PqsCommunicateDto;
 import com.njcn.csdevice.pojo.param.DataArrayParam;
 import com.njcn.csdevice.pojo.po.CsDataArray;
@@ -61,6 +63,7 @@ public class StatServiceImpl implements IStatService {
     private final CsDeviceFeignClient csDeviceFeignClient;
     private final DeviceMessageFeignClient deviceMessageFeignClient;
     private final CsCommunicateFeignClient csCommunicateFeignClient;
+    private final CsLineFeignClient csLineFeignClient;
 
     @Override
     @Transactional(rollbackFor = Exception.class)
@@ -82,7 +85,9 @@ public class StatServiceImpl implements IStatService {
         String lineId = null;
         Object object1 = redisUtil.getObjectByKey(AppRedisKey.LINE_POSITION+appAutoDataMessage.getId());
         if (Objects.isNull(object1)){
-            deviceMessageFeignClient.getLineInfo(appAutoDataMessage.getId());
+            LineInfoParam param = new LineInfoParam();
+            param.setNDid(appAutoDataMessage.getId());
+            deviceMessageFeignClient.getLineInfo(param);
         }
         //获取当前设备信息判断装置型号，来筛选监测点
         List<CsEquipmentDeliveryPO> poList = channelObjectUtil.objectToList(redisUtil.getObjectByKey(AppRedisKey.DEVICE_LIST),CsEquipmentDeliveryPO.class);
@@ -109,6 +114,7 @@ public class StatServiceImpl implements IStatService {
 
         //获取当前设备信息
         if (CollectionUtil.isNotEmpty(list)) {
+            Map<String,String> map = new Gson().fromJson(String.valueOf(redisUtil.getObjectByKey(AppRedisKey.ELE_EPD_PQD)), Map.class);
             List<String> recordList = new ArrayList<>();
             for (AppAutoDataMessage.DataArray item : list) {
                 switch (item.getDataAttr()) {
@@ -133,7 +139,8 @@ public class StatServiceImpl implements IStatService {
                 }
                 boolean flag = Objects.equals(DicDataEnum.DEV_CLD.getCode(), code) && Objects.equals(po.getDevAccessMethod(), "CLD");
                 int clDid = flag?1:appAutoDataMessage.getMsg().getClDid();
-                String key = AppRedisKey.DEV_MODEL.concat(dataArrayParam.getId() + dataArrayParam.getDid() + clDid + dataArrayParam.getStatMethod() + dataArrayParam.getIdx());
+//                String key = AppRedisKey.DEV_MODEL.concat(dataArrayParam.getId() + dataArrayParam.getDid() + clDid + dataArrayParam.getStatMethod() + dataArrayParam.getIdx());
+                String key = AppRedisKey.DEV_MODEL.concat(dataArrayParam.getId() + dataArrayParam.getDid() + clDid + dataArrayParam.getIdx());
                 Object object = redisUtil.getObjectByKey(key);
                 List<CsDataArray> dataArrayList;
                 if (Objects.isNull(object)){
@@ -141,7 +148,7 @@ public class StatServiceImpl implements IStatService {
                 } else {
                     dataArrayList = objectToList(object);
                 }
-                List<String> result = assembleData(lineId,dataArrayList,item,appAutoDataMessage.getMsg().getClDid(),dataArrayParam.getStatMethod(),po.getProcess(),code,po.getDevAccessMethod());
+                List<String> result = assembleData(lineId,dataArrayList,item,appAutoDataMessage.getMsg().getClDid(),dataArrayParam.getStatMethod(),po.getProcess(),code,po.getDevAccessMethod(),map);
                 recordList.addAll(result);
                 //获取时间
                 boolean timeFlag = Objects.equals(DicDataEnum.DEV_CLD.getCode(), code) && Objects.equals(po.getDevAccessMethod(), "CLD");
@@ -152,7 +159,7 @@ public class StatServiceImpl implements IStatService {
             }
             if (CollectionUtil.isNotEmpty(recordList)){
                 //influx数据批量入库
-                influxDbUtils.batchInsert(influxDbUtils.getDbName(), "", InfluxDB.ConsistencyLevel.ALL, TimeUnit.MILLISECONDS, recordList);
+                influxDbUtils.batchInsert(influxDbUtils.getDbName(), "", InfluxDB.ConsistencyLevel.ALL, TimeUnit.SECONDS, recordList);
                 //记录监测点最新数据时间
                 CsLineLatestData csLineLatestData = new CsLineLatestData();
                 csLineLatestData.setLineId(lineId);
@@ -190,36 +197,55 @@ public class StatServiceImpl implements IStatService {
     /**
      * influxDB数据组装
      */
-    public List<String> assembleData(String lineId,List<CsDataArray> dataArrayList,AppAutoDataMessage.DataArray item,Integer clDid,String statMethod,Integer process,String devType,String accessMethod) {
+    public List<String> assembleData(String lineId,List<CsDataArray> dataArrayList,AppAutoDataMessage.DataArray item,Integer clDid,String statMethod,Integer process,String devType,String accessMethod, Map<String,String> map) {
         List<String> records = new ArrayList<String>();
-        //解码
         List<Float> floats = PubUtils.byteArrayToFloatList(Base64.getDecoder().decode(item.getData()));
         if (CollectionUtil.isEmpty(floats)){
             throw new BusinessException(StatResponseEnum.AUTO_DATA_NULL);
         }
-        //校验模板和解码数据数量能否对应上
         if (!Objects.equals(dataArrayList.size(),floats.size())){
             throw new BusinessException(StatResponseEnum.ARRAY_DATA_NOT_MATCH);
         }
-        Map<String,String> map = new Gson().fromJson(String.valueOf(redisUtil.getObjectByKey(AppRedisKey.ELE_EPD_PQD)), Map.class);
+
+        boolean flag = Objects.equals(DicDataEnum.DEV_CLD.getCode(), devType) && Objects.equals(accessMethod, "CLD");
+        //fixme 捂脸设备上送的是北京时间，时序数据库录入时 需要utc时间，减去8小时
+        long originalTimeSec = flag ? item.getDataTimeSec() : item.getDataTimeSec() - 8 * 3600;
+
         for (int i = 0; i < dataArrayList.size(); i++) {
             String tableName = map.get(dataArrayList.get(i).getName());
+            long adjustedTimeSec;
+
+            //短时闪变 || 电压波动 10分钟
+            if (Objects.equals(tableName,"data_flicker") || Objects.equals(tableName,"data_fluc")) {
+                adjustedTimeSec = (originalTimeSec / 600) * 600;
+            }
+            //长时闪变 2小时
+            else if (Objects.equals(tableName,"data_plt")) {
+                adjustedTimeSec = (originalTimeSec / 7200) * 7200;
+            }
+            else {
+                adjustedTimeSec = originalTimeSec;
+            }
             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");
+            tags.put(InfluxDBTableConstant.VALUE_TYPE,statMethod.toUpperCase());
+            if (Objects.isNull(item.getDataTag())) {
+                tags.put(InfluxDBTableConstant.QUALITY_FLAG,"0");
+            } else {
+                tags.put(InfluxDBTableConstant.QUALITY_FLAG,String.valueOf(item.getDataTag()));
+            }
             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));
-            fields.put(InfluxDBTableConstant.IS_ABNORMAL,item.getDataTag());
-            //fixme 设备上送的是北京时间，时序数据库录入时 需要utc时间，减去8小时
-            boolean flag = Objects.equals(DicDataEnum.DEV_CLD.getCode(), devType) && Objects.equals(accessMethod, "CLD");
-            Point point = influxDbUtils.pointBuilder(tableName, flag?item.getDataTimeSec():item.getDataTimeSec()-8*3600, TimeUnit.SECONDS, tags, fields);
+            if (Objects.isNull(dataArrayList.get(i).getInfluxDbName())) {
+                fields.put(dataArrayList.get(i).getName(),Objects.equals(floats.get(i),3.14159f) ? null:floats.get(i));
+            } else {
+                fields.put(dataArrayList.get(i).getInfluxDbName(),Objects.equals(floats.get(i),3.14159f) ? null:floats.get(i));
+            }
+            fields.put(InfluxDBTableConstant.CL_DID,clDid.toString());
+            fields.put(InfluxDBTableConstant.PROCESS,process.toString());
+
+            Point point = influxDbUtils.pointBuilder(tableName, adjustedTimeSec, TimeUnit.SECONDS, tags, fields);
             BatchPoints batchPoints = BatchPoints.database(influxDbUtils.getDbName()).retentionPolicy("").consistency(InfluxDB.ConsistencyLevel.ALL).build();
             batchPoints.point(point);
             records.add(batchPoints.lineProtocol());
@@ -238,4 +264,5 @@ public class StatServiceImpl implements IStatService {
         }
         return urlList;
     }
+
 }

iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/mapper/CsEventUserMapper.java

@@ -1,16 +1,16 @@
-package com.njcn.zlevent.mapper;
-
-import com.baomidou.mybatisplus.core.mapper.BaseMapper;
-import com.njcn.csharmonic.pojo.po.CsEventUserPO;
-
-/**
- * <p>
- * 暂态事件表 Mapper 接口
- * </p>
- *
- * @author xuyang
- * @since 2023-08-23
- */
-public interface CsEventUserMapper extends BaseMapper<CsEventUserPO> {
-
-}
+//package com.njcn.zlevent.mapper;
+//
+//import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+//import com.njcn.csharmonic.pojo.po.CsEventUserPO;
+//
+///**
+// * <p>
+// * 暂态事件表 Mapper 接口
+// * </p>
+// *
+// * @author xuyang
+// * @since 2023-08-23
+// */
+//public interface CsEventUserMapper extends BaseMapper<CsEventUserPO> {
+//
+//}

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

@@ -1,161 +1,72 @@
-package com.njcn.zlevent.service;
-
-import cn.hutool.core.collection.CollectionUtil;
-import cn.hutool.core.date.DatePattern;
-import com.njcn.access.pojo.dto.NoticeUserDto;
-import com.njcn.access.utils.SendMessageUtil;
-import com.njcn.csdevice.api.CsLedgerFeignClient;
-import com.njcn.csdevice.api.DeviceMessageFeignClient;
-import com.njcn.csdevice.api.EquipmentFeignClient;
-import com.njcn.csdevice.api.EventLogsFeignClient;
-import com.njcn.csdevice.param.DeviceMessageParam;
-import com.njcn.csdevice.pojo.dto.DevDetailDTO;
-import com.njcn.csdevice.pojo.po.CsEventSendMsg;
-import com.njcn.csharmonic.pojo.po.CsEventUserPO;
-import com.njcn.system.api.EpdFeignClient;
-import com.njcn.user.pojo.po.User;
-import lombok.RequiredArgsConstructor;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.scheduling.annotation.Async;
-import org.springframework.stereotype.Service;
-
-import java.time.LocalDateTime;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Objects;
-import java.util.stream.Collectors;
-
-/**
- * @author xy
- */
-@Service
-@Slf4j
-@RequiredArgsConstructor
-public class AppNotificationService {
-
-    private final DeviceMessageFeignClient deviceMessageFeignClient;
-    private final EquipmentFeignClient equipmentFeignClient;
-    private final EpdFeignClient epdFeignClient;
-    private final ICsEventUserService csEventUserService;
-    private final SendMessageUtil sendMessageUtil;
-    private final EventLogsFeignClient eventLogsFeignClient;
-    private final CsLedgerFeignClient csLedgerFeignClient;
-
-    @Async("eventNotificationExecutor")
-    public void sendAppNotification(Integer eventType, String type, String devId,
-                                    String eventName, LocalDateTime eventTime,
-                                    String id, String nDid, Double amplitude, Double persistTime) {
-        int code;
-        List<User> users = new ArrayList<>();
-        List<String> eventUser;
-        List<String> devCodeList;
-        List<String> userList = new ArrayList<>();
-        List<CsEventSendMsg> csEventSendMsgList = new ArrayList<>();
-        NoticeUserDto noticeUserDto = new NoticeUserDto();
-        NoticeUserDto.Payload payload = new NoticeUserDto.Payload();
-        String content = null;
-        List<CsEventUserPO> result = new ArrayList<>();
-        //获取设备类型 true:治理设备 false:其他类型的设备
-        boolean devModel = equipmentFeignClient.judgeDevModel(nDid).getData();
-        if (devModel) {
-            DevDetailDTO devDetailDto = csLedgerFeignClient.queryDevDetail(devId).getData();
-            //事件处理
-            if (eventType == 1){
-                eventName = epdFeignClient.findByName(eventName).getData().getShowName();
-                switch (type) {
-                    case "1":
-                        code = 3;
-                        //设备自身事件 不推送给用户，推送给业务管理
-                        eventUser = deviceMessageFeignClient.getEventUserByDeviceId(devId,false).getData();
-                        if (CollectionUtil.isNotEmpty(eventUser)) {
-                            eventUser.forEach(item->{
-                                CsEventUserPO csEventUser = new CsEventUserPO();
-                                csEventUser.setUserId(item);
-                                csEventUser.setStatus(0);
-                                csEventUser.setEventId(id);
-                                result.add(csEventUser);
-                            });
-
-                            DeviceMessageParam param1 = new DeviceMessageParam();
-                            param1.setUserList(eventUser);
-                            param1.setEventType(2);
-                            users = deviceMessageFeignClient.getSendUserByType(param1).getData();
-                            if (CollectionUtil.isNotEmpty(users)){
-                                for (User user : users){
-                                    userList.add(user.getDevCode());
-                                }
-                                noticeUserDto.setPushClientId(userList);
-                                noticeUserDto.setTitle("运行事件");
-                            }
-                        }
-                        break;
-                    case "2":
-                        code = 0;
-                        //暂态事件
-                        eventUser = deviceMessageFeignClient.getEventUserByDeviceId(devId,true).getData();
-                        if (CollectionUtil.isNotEmpty(eventUser)) {
-                            eventUser.forEach(item->{
-                                CsEventUserPO csEventUser = new CsEventUserPO();
-                                csEventUser.setUserId(item);
-                                csEventUser.setStatus(0);
-                                csEventUser.setEventId(id);
-                                result.add(csEventUser);
-                            });
-                            DeviceMessageParam param1 = new DeviceMessageParam();
-                            param1.setUserList(eventUser);
-                            param1.setEventType(0);
-                            users = deviceMessageFeignClient.getSendUserByType(param1).getData();
-                            if (CollectionUtil.isNotEmpty(users)){
-                                devCodeList = users.stream().map(User::getDevCode).distinct().collect(Collectors.toList());
-                                noticeUserDto.setPushClientId(devCodeList);
-                                noticeUserDto.setTitle("暂态事件");
-                                content = devDetailDto.getEngineeringName() + "-" + devDetailDto.getProjectName() + "-" + devDetailDto.getEquipmentName()
-                                        + "于" + eventTime.format(DatePattern.NORM_DATETIME_MS_FORMATTER) + "发生暂态事件，事件类型："
-                                        + eventName
-                                        + "，特征幅值：" + amplitude + "%"
-                                        + "，持续时间：" + persistTime + "s";
-                            }
-                        }
-                        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;
-                }
-                //获取台账信息
-                if (Objects.isNull(content)) {
-                    content = devDetailDto.getEngineeringName() + "-" + devDetailDto.getProjectName() + "-" + devDetailDto.getEquipmentName() + "于" + eventTime.format(DatePattern.NORM_DATETIME_MS_FORMATTER) + "发生" + eventName;
-                }
-                noticeUserDto.setContent(content);
-                payload.setType(code);
-                payload.setPath("/pages/index/message1?type="+payload.getType());
-                noticeUserDto.setPayload(payload);
-            }
-//            //告警处理
-//            else if (eventType == 2){
+//package com.njcn.zlevent.service;
+//
+//import cn.hutool.core.collection.CollectionUtil;
+//import cn.hutool.core.date.DatePattern;
+//import com.njcn.access.pojo.dto.NoticeUserDto;
+//import com.njcn.access.utils.SendMessageUtil;
+//import com.njcn.csdevice.api.CsLedgerFeignClient;
+//import com.njcn.csdevice.api.DeviceMessageFeignClient;
+//import com.njcn.csdevice.api.EquipmentFeignClient;
+//import com.njcn.csdevice.api.EventLogsFeignClient;
+//import com.njcn.csdevice.param.DeviceMessageParam;
+//import com.njcn.csdevice.pojo.dto.DevDetailDTO;
+//import com.njcn.csdevice.pojo.po.CsEventSendMsg;
+//import com.njcn.csharmonic.pojo.po.CsEventUserPO;
+//import com.njcn.system.api.EpdFeignClient;
+//import com.njcn.user.pojo.po.User;
+//import lombok.RequiredArgsConstructor;
+//import lombok.extern.slf4j.Slf4j;
+//import org.springframework.scheduling.annotation.Async;
+//import org.springframework.stereotype.Service;
+//
+//import java.time.LocalDateTime;
+//import java.util.ArrayList;
+//import java.util.List;
+//import java.util.Objects;
+//import java.util.stream.Collectors;
+//
+///**
+// * @author xy
+// */
+//@Service
+//@Slf4j
+//@RequiredArgsConstructor
+//public class AppNotificationService {
+//
+//    private final DeviceMessageFeignClient deviceMessageFeignClient;
+//    private final EquipmentFeignClient equipmentFeignClient;
+//    private final EpdFeignClient epdFeignClient;
+//    private final ICsEventUserService csEventUserService;
+//    private final SendMessageUtil sendMessageUtil;
+//    private final EventLogsFeignClient eventLogsFeignClient;
+//    private final CsLedgerFeignClient csLedgerFeignClient;
+//
+//    @Async("eventNotificationExecutor")
+//    public void sendAppNotification(Integer eventType, String type, String devId,
+//                                        String eventName, LocalDateTime eventTime,
+//                                    String id, String nDid, Double amplitude, Double persistTime,String dropZone) {
+//        int code;
+//        List<User> users = new ArrayList<>();
+//        List<String> eventUser;
+//        List<String> devCodeList;
+//        List<String> userList = new ArrayList<>();
+//        List<CsEventSendMsg> csEventSendMsgList = new ArrayList<>();
+//        NoticeUserDto noticeUserDto = new NoticeUserDto();
+//        NoticeUserDto.Payload payload = new NoticeUserDto.Payload();
+//        String content = null;
+//        List<CsEventUserPO> result = new ArrayList<>();
+//        //获取设备类型 true:治理设备 false:其他类型的设备
+//        boolean devModel = equipmentFeignClient.judgeDevModel(nDid).getData();
+//        if (devModel) {
+//            DevDetailDTO devDetailDto = csLedgerFeignClient.queryDevDetail(devId).getData();
+//            //事件处理
+//            if (eventType == 1){
+//                eventName = epdFeignClient.findByName(eventName).getData().getShowName();
 //                switch (type) {
 //                    case "1":
-//                        //Ⅰ级告警 不推送给用户，推送给业务管理
-//                        eventUser = getEventUser(devId,false);
+//                        code = 3;
+//                        //设备自身事件 不推送给用户，推送给业务管理
+//                        eventUser = deviceMessageFeignClient.getEventUserByDeviceId(devId,false).getData();
 //                        if (CollectionUtil.isNotEmpty(eventUser)) {
 //                            eventUser.forEach(item->{
 //                                CsEventUserPO csEventUser = new CsEventUserPO();
@@ -164,19 +75,24 @@ public class AppNotificationService {
 //                                csEventUser.setEventId(id);
 //                                result.add(csEventUser);
 //                            });
-//                            users = getSendUser(eventUser,3);
+//
+//                            DeviceMessageParam param1 = new DeviceMessageParam();
+//                            param1.setUserList(eventUser);
+//                            param1.setEventType(2);
+//                            users = deviceMessageFeignClient.getSendUserByType(param1).getData();
 //                            if (CollectionUtil.isNotEmpty(users)){
-//                                eventName = epdFeignClient.findByName(eventName).getData().getShowName();
-//                                devCodeList = users.stream().map(User::getDevCode).distinct().collect(Collectors.toList());
-//                                noticeUserDto.setPushClientId(devCodeList);
+//                                for (User user : users){
+//                                    userList.add(user.getDevCode());
+//                                }
+//                                noticeUserDto.setPushClientId(userList);
+//                                noticeUserDto.setTitle("运行事件");
 //                            }
 //                        }
 //                        break;
 //                    case "2":
-//                        eventName = epdFeignClient.findByName(eventName).getData().getShowName();
-//                    case "3":
-//                        //Ⅱ、Ⅲ级告警推送相关用户及业务管理员
-//                        eventUser = getEventUser(devId,true);
+//                        code = 0;
+//                        //暂态事件
+//                        eventUser = deviceMessageFeignClient.getEventUserByDeviceId(devId,true).getData();
 //                        if (CollectionUtil.isNotEmpty(eventUser)) {
 //                            eventUser.forEach(item->{
 //                                CsEventUserPO csEventUser = new CsEventUserPO();
@@ -185,54 +101,139 @@ public class AppNotificationService {
 //                                csEventUser.setEventId(id);
 //                                result.add(csEventUser);
 //                            });
-//                            users = getSendUser(eventUser,3);
+//                            DeviceMessageParam param1 = new DeviceMessageParam();
+//                            param1.setUserList(eventUser);
+//                            param1.setEventType(0);
+//                            users = deviceMessageFeignClient.getSendUserByType(param1).getData();
 //                            if (CollectionUtil.isNotEmpty(users)){
 //                                devCodeList = users.stream().map(User::getDevCode).distinct().collect(Collectors.toList());
 //                                noticeUserDto.setPushClientId(devCodeList);
+//                                noticeUserDto.setTitle("暂态事件");
+//                                content = devDetailDto.getEngineeringName() + "-" + devDetailDto.getProjectName() + "-" + devDetailDto.getEquipmentName()
+//                                        + "于" + eventTime.format(DatePattern.NORM_DATETIME_MS_FORMATTER) + "发生暂态事件，事件类型："
+//                                        + eventName
+//                                        + "，特征幅值：" + amplitude + "%"
+//                                        + "，持续时间：" + persistTime + "s"
+//                                        + "，落点区域：" + (Objects.isNull(dropZone)?"未知":dropZone);
 //                            }
 //                        }
 //                        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;
 //                }
-//                noticeUserDto.setTitle("告警事件");
-//                DevDetailDTO devDetailDto = csLedgerFeignclient.queryDevDetail(devId).getData();
-//                content = devDetailDto.getEngineeringName() + "-" + devDetailDto.getProjectName() + "-" + devDetailDto.getEquipmentName() + "于" + eventTime.format(DatePattern.NORM_DATETIME_MS_FORMATTER) + "发生告警,告警信息:" + eventName;
+//                //获取台账信息
+//                if (Objects.isNull(content)) {
+//                    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.setType(code);
+//                payload.setPath("/pages/index/message1?type="+payload.getType());
 //                noticeUserDto.setPayload(payload);
 //            }
-            if (CollectionUtil.isNotEmpty(noticeUserDto.getPushClientId())) {
-                List<String> filteredList = noticeUserDto.getPushClientId().stream()
-                        .filter(Objects::nonNull)
-                        .distinct()
-                        .collect(Collectors.toList());
-                if (CollectionUtil.isNotEmpty(filteredList)) {
-                    noticeUserDto.setPushClientId(filteredList);
-                    sendMessageUtil.sendEventToUser(noticeUserDto);
-                }
-            }
-            //记录推送日志
-            for (User item : users) {
-                CsEventSendMsg csEventSendMsg = new CsEventSendMsg();
-                csEventSendMsg.setUserId(item.getId());
-                csEventSendMsg.setEventId(id);
-                csEventSendMsg.setSendTime(LocalDateTime.now());
-                if (Objects.isNull(item.getDevCode())){
-                    csEventSendMsg.setStatus(0);
-                    csEventSendMsg.setRemark("用户设备识别码为空");
-                } else {
-                    csEventSendMsg.setDevCode(item.getDevCode());
-                    csEventSendMsg.setStatus(1);
-                }
-                csEventSendMsgList.add(csEventSendMsg);
-            }
-            eventLogsFeignClient.addLogs(csEventSendMsgList);
-            //事件用户关系入库
-            if (CollectionUtil.isNotEmpty(result)){
-                csEventUserService.saveBatch(result);
-            }
-        }
-    }
-}
+////            //告警处理
+////            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)
+//                        .distinct()
+//                        .collect(Collectors.toList());
+//                if (CollectionUtil.isNotEmpty(filteredList)) {
+//                    noticeUserDto.setPushClientId(filteredList);
+//                    sendMessageUtil.sendEventToUser(noticeUserDto);
+//                }
+//            }
+//            //记录推送日志
+//            for (User item : users) {
+//                CsEventSendMsg csEventSendMsg = new CsEventSendMsg();
+//                csEventSendMsg.setUserId(item.getId());
+//                csEventSendMsg.setEventId(id);
+//                csEventSendMsg.setSendTime(LocalDateTime.now());
+//                if (Objects.isNull(item.getDevCode())){
+//                    csEventSendMsg.setStatus(0);
+//                    csEventSendMsg.setRemark("用户设备识别码为空");
+//                } else {
+//                    csEventSendMsg.setDevCode(item.getDevCode());
+//                    csEventSendMsg.setStatus(1);
+//                }
+//                csEventSendMsgList.add(csEventSendMsg);
+//            }
+//            eventLogsFeignClient.addLogs(csEventSendMsgList);
+//            //事件用户关系入库
+//            if (CollectionUtil.isNotEmpty(result)){
+//                csEventUserService.saveBatch(result);
+//            }
+//        }
+//    }
+//}

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

@@ -19,6 +19,14 @@ public interface ICsEventService extends IService<CsEventPO> {
     /**
      * 事件添加波形文件地址
      */
-    List<String> updateCsEvent(CsEventParam csEventParam);
+    List<CsEventPO> updateCsEvent(CsEventParam csEventParam);
+
+    /**
+     * 暂降事件添加原因和类型
+     * 暂降原因(0:未知 1:短路故障 2:电压调节器 3:感动电机 4:电压跌落)
+     * 暂降类型(0:BC相间故障 1:C相接地故障 2:AC相间故障 3:A相接地故障 4:AB相间故障
+     *        5:B相接地故障 6:BC相间接地 7:AC相间接地 8:AB相间接地 9:三相故障 10:未知)
+     */
+    void updateEventCauseAndType(String id, Integer cause, Integer type);
 
 }

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

@@ -1,16 +1,16 @@
-package com.njcn.zlevent.service;
-
-import com.baomidou.mybatisplus.extension.service.IService;
-import com.njcn.csharmonic.pojo.po.CsEventUserPO;
-
-/**
- * <p>
- * 暂态事件表 服务类
- * </p>
- *
- * @author xuyang
- * @since 2023-08-23
- */
-public interface ICsEventUserService extends IService<CsEventUserPO> {
-
-}
+//package com.njcn.zlevent.service;
+//
+//import com.baomidou.mybatisplus.extension.service.IService;
+//import com.njcn.csharmonic.pojo.po.CsEventUserPO;
+//
+///**
+// * <p>
+// * 暂态事件表 服务类
+// * </p>
+// *
+// * @author xuyang
+// * @since 2023-08-23
+// */
+//public interface ICsEventUserService extends IService<CsEventUserPO> {
+//
+//}

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

@@ -1,70 +1,71 @@
-package com.njcn.zlevent.service;
-
-import cn.hutool.core.collection.CollectionUtil;
-import cn.hutool.core.date.DatePattern;
-import cn.hutool.core.util.StrUtil;
-import com.njcn.csdevice.api.CsLedgerFeignClient;
-import com.njcn.csdevice.api.SmsSendFeignClient;
-import com.njcn.csdevice.pojo.dto.DevDetailDTO;
-import com.njcn.cssystem.api.AppMsgSetFeignClient;
-import com.njcn.user.api.UserFeignClient;
-import com.njcn.user.pojo.po.User;
-import lombok.RequiredArgsConstructor;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.annotation.Value;
-import org.springframework.scheduling.annotation.Async;
-import org.springframework.stereotype.Service;
-
-import java.time.LocalDateTime;
-import java.util.List;
-import java.util.Objects;
-import java.util.stream.Collectors;
-
-/**
- * @author xy
- */
-@Service
-@Slf4j
-@RequiredArgsConstructor
-public class SmsNotificationService {
-
-    private final AppMsgSetFeignClient appMsgSetFeignClient;
-    private final UserFeignClient userFeignClient;
-    private final SmsSendFeignClient smsSendFeignClient;
-    private final CsLedgerFeignClient csLedgerFeignclient;
-
-    @Value("${msg.msg_sign:南京灿能电力}")
-    private String msgSign;
-
-    @Async("smsNotificationExecutor")
-    public void sendSmsForDipEvent(String deviceId, LocalDateTime eventTime,double amplitude,double persistTime) {
-        try {
-            List<String> userIdList = appMsgSetFeignClient.queryUserIdsByDeviceId(deviceId).getData();
-            if (CollectionUtil.isNotEmpty(userIdList)) {
-                List<User> userList = userFeignClient.getUserListByIds(userIdList).getData();
-                if (CollectionUtil.isNotEmpty(userList)) {
-                    List<User> userList1 = userList.stream()
-                            .filter(item -> StrUtil.isNotBlank(item.getPhone()) && Objects.equals(item.getSmsNotice(), 1))
-                            .collect(Collectors.toList());
-                    if (CollectionUtil.isNotEmpty(userList1)) {
-                        DevDetailDTO devDetailDto = csLedgerFeignclient.queryDevDetail(deviceId).getData();
-                        String msgContent = "【" + msgSign + "】" + devDetailDto.getEngineeringName()
-                                + "-" + devDetailDto.getProjectName() + "-" + devDetailDto.getEquipmentName()
-                                + "于" + eventTime.format(DatePattern.NORM_DATETIME_MS_FORMATTER) + "发生暂降事件"
-                                + "，特征幅值：" + amplitude + "%"
-                                + "，持续时间：" + persistTime + "s";
-                        userList1.forEach(item -> {
-                            try {
-                                smsSendFeignClient.sendSmsSimple(item.getPhone(), msgContent, "verify_code");
-                            } catch (Exception e) {
-                                log.error("发送短信失败，手机号: {}", item.getPhone(), e);
-                            }
-                        });
-                    }
-                }
-            }
-        } catch (Exception e) {
-            log.error("异步发送暂降事件短信失败，设备ID: {}", deviceId, e);
-        }
-    }
-}
+//package com.njcn.zlevent.service;
+//
+//import cn.hutool.core.collection.CollectionUtil;
+//import cn.hutool.core.date.DatePattern;
+//import cn.hutool.core.util.StrUtil;
+//import com.njcn.csdevice.api.CsLedgerFeignClient;
+//import com.njcn.csdevice.api.SmsSendFeignClient;
+//import com.njcn.csdevice.pojo.dto.DevDetailDTO;
+//import com.njcn.cssystem.api.AppMsgSetFeignClient;
+//import com.njcn.user.api.UserFeignClient;
+//import com.njcn.user.pojo.po.User;
+//import lombok.RequiredArgsConstructor;
+//import lombok.extern.slf4j.Slf4j;
+//import org.springframework.beans.factory.annotation.Value;
+//import org.springframework.scheduling.annotation.Async;
+//import org.springframework.stereotype.Service;
+//
+//import java.time.LocalDateTime;
+//import java.util.List;
+//import java.util.Objects;
+//import java.util.stream.Collectors;
+//
+///**
+// * @author xy
+// */
+//@Service
+//@Slf4j
+//@RequiredArgsConstructor
+//public class SmsNotificationService {
+//
+//    private final AppMsgSetFeignClient appMsgSetFeignClient;
+//    private final UserFeignClient userFeignClient;
+//    private final SmsSendFeignClient smsSendFeignClient;
+//    private final CsLedgerFeignClient csLedgerFeignclient;
+//
+//    @Value("${msg.msg_sign:南京灿能电力}")
+//    private String msgSign;
+//
+//    @Async("smsNotificationExecutor")
+//    public void sendSmsForDipEvent(String deviceId, LocalDateTime eventTime,double amplitude,double persistTime,String dropZone) {
+//        try {
+//            List<String> userIdList = appMsgSetFeignClient.queryUserIdsByDeviceId(deviceId).getData();
+//            if (CollectionUtil.isNotEmpty(userIdList)) {
+//                List<User> userList = userFeignClient.getUserListByIds(userIdList).getData();
+//                if (CollectionUtil.isNotEmpty(userList)) {
+//                    List<User> userList1 = userList.stream()
+//                            .filter(item -> StrUtil.isNotBlank(item.getPhone()) && Objects.equals(item.getSmsNotice(), 1))
+//                            .collect(Collectors.toList());
+//                    if (CollectionUtil.isNotEmpty(userList1)) {
+//                        DevDetailDTO devDetailDto = csLedgerFeignclient.queryDevDetail(deviceId).getData();
+//                        String msgContent = "【" + msgSign + "】" + devDetailDto.getEngineeringName()
+//                                + "-" + devDetailDto.getProjectName() + "-" + devDetailDto.getEquipmentName()
+//                                + "于" + eventTime.format(DatePattern.NORM_DATETIME_MS_FORMATTER) + "发生暂降事件"
+//                                + "，特征幅值：" + amplitude + "%"
+//                                + "，持续时间：" + persistTime + "s"
+//                                + "，落点区域：" + (Objects.isNull(dropZone)?"未知":dropZone);
+//                        userList1.forEach(item -> {
+//                            try {
+//                                smsSendFeignClient.sendSmsSimple(item.getPhone(), msgContent, "verify_code");
+//                            } catch (Exception e) {
+//                                log.error("发送短信失败，手机号: {}", item.getPhone(), e);
+//                            }
+//                        });
+//                    }
+//                }
+//            }
+//        } catch (Exception e) {
+//            log.error("异步发送暂降事件短信失败，设备ID: {}", deviceId, e);
+//        }
+//    }
+//}

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

@@ -9,6 +9,8 @@ import com.njcn.common.pojo.exception.BusinessException;
 import com.njcn.csdevice.api.EquipmentFeignClient;
 import com.njcn.csdevice.pojo.po.CsEquipmentDeliveryPO;
 import com.njcn.csharmonic.pojo.po.CsEventPO;
+import com.njcn.cssystem.api.MsgSendFeignClient;
+import com.njcn.cssystem.pojo.param.MsgSendParam;
 import com.njcn.mq.message.AppEventMessage;
 import com.njcn.redis.pojo.enums.AppRedisKey;
 import com.njcn.redis.utils.RedisUtil;
@@ -19,7 +21,7 @@ import com.njcn.system.pojo.po.EleEpdPqd;
 import com.njcn.system.pojo.po.SysDicTreePO;
 import com.njcn.zlevent.mapper.CsEventMapper;
 import com.njcn.zlevent.pojo.po.CsEventLogs;
-import com.njcn.zlevent.service.AppNotificationService;
+//import com.njcn.zlevent.service.AppNotificationService;
 import com.njcn.zlevent.service.ICsAlarmService;
 import com.njcn.zlevent.service.ICsEventLogsService;
 import com.njcn.zlevent.service.ICsEventService;
@@ -54,7 +56,8 @@ public class CsAlarmServiceImpl extends ServiceImpl<CsEventMapper, CsEventPO> im
     private final EpdFeignClient epdFeignClient;
     private final RedisUtil redisUtil;
     private final ChannelObjectUtil channelObjectUtil;
-    private final AppNotificationService appNotificationService;
+//    private final AppNotificationService appNotificationService;
+    private final MsgSendFeignClient msgSendFeignClient;
 
     @Override
     @Transactional(rollbackFor = Exception.class)
@@ -125,10 +128,21 @@ public class CsAlarmServiceImpl extends ServiceImpl<CsEventMapper, CsEventPO> im
                 csEventService.saveBatch(list1);
                 //推送事件逻辑处理 && cs_event_user入库 && 修改字典中告警事件的编码
                 for (AppEventMessage.DataArray item : dataArray) {
+                    MsgSendParam msgSendParam = new MsgSendParam();
+                    msgSendParam.setEventType(2);
+                    msgSendParam.setType(item.getType());
+                    msgSendParam.setDevId(po.getId());
+                    msgSendParam.setEventTime(eventTime);
+                    msgSendParam.setId(id);
+                    msgSendParam.setNDid(po.getNdid());
                     if (Objects.isNull(item.getCode())){
-                        appNotificationService.sendAppNotification(2,item.getType(),po.getId(),item.getName(),eventTime,id,po.getNdid(),null,null);
+                        msgSendParam.setEventName(item.getName());
+                        msgSendFeignClient.appMsgSend(msgSendParam);
+//                        appNotificationService.sendAppNotification(2,item.getType(),po.getId(),item.getName(),eventTime,id,po.getNdid(),null,null,null);
                     } else {
-                        appNotificationService.sendAppNotification(2,item.getType(),po.getId(),item.getCode(),eventTime,id,po.getNdid(),null, null);
+                        msgSendParam.setEventName(item.getCode());
+                        msgSendFeignClient.appMsgSend(msgSendParam);
+//                        appNotificationService.sendAppNotification(2,item.getType(),po.getId(),item.getCode(),eventTime,id,po.getNdid(),null, null,null);
                         //更新字典信息
                         EleEpdPqd eleEpdPqd = epdFeignClient.findByName(item.getName()).getData();
                         EleEpdPqdParam.EleEpdPqdUpdateParam updateParam = new EleEpdPqdParam.EleEpdPqdUpdateParam();

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

@@ -1,9 +1,11 @@
 package com.njcn.zlevent.service.impl;
 
-import cn.hutool.core.collection.CollectionUtil;
 import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
 import com.njcn.csharmonic.pojo.po.CsEventPO;
+import com.njcn.system.api.DicDataFeignClient;
+import com.njcn.system.enums.DicDataTypeEnum;
+import com.njcn.system.pojo.po.DictData;
 import com.njcn.zlevent.mapper.CsEventMapper;
 import com.njcn.zlevent.param.CsEventParam;
 import com.njcn.zlevent.service.ICsEventService;
@@ -11,11 +13,11 @@ import lombok.AllArgsConstructor;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
-import java.text.SimpleDateFormat;
 import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
-import java.util.*;
-import java.util.stream.Collectors;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
 
 /**
  * <p>
@@ -29,11 +31,11 @@ import java.util.stream.Collectors;
 @AllArgsConstructor
 public class CsEventServiceImpl extends ServiceImpl<CsEventMapper, CsEventPO> implements ICsEventService {
 
+    private final DicDataFeignClient dicDataFeignClient;
+
     @Override
     @Transactional(rollbackFor = Exception.class)
-    public List<String> updateCsEvent(CsEventParam csEventParam) {
-        List<String> eventList = new ArrayList<>();
-
+    public List<CsEventPO> updateCsEvent(CsEventParam csEventParam) {
         DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSSSSS");
         LocalDateTime dateTime = LocalDateTime.parse(csEventParam.getStartTime(), formatter);
         // 减去1毫秒
@@ -50,10 +52,30 @@ public class CsEventServiceImpl extends ServiceImpl<CsEventMapper, CsEventPO> im
             lambdaUpdateWrapper.eq(CsEventPO::getLocation, csEventParam.getLocation());
         }
         this.update(lambdaUpdateWrapper);
-        List<CsEventPO> list = this.baseMapper.selectList(lambdaUpdateWrapper);
-        if (CollectionUtil.isNotEmpty(list)){
-            eventList = list.stream().map(CsEventPO::getId).collect(Collectors.toList());
-        }
-        return eventList;
+        return this.baseMapper.selectList(lambdaUpdateWrapper);
     }
+
+    @Override
+    public void updateEventCauseAndType(String id, Integer cause, Integer type) {
+        List<DictData> list1 = dicDataFeignClient.getDicDataByTypeCode(DicDataTypeEnum.EVENT_REASON.getCode()).getData();
+        String id1 = list1.stream()
+                .filter(item -> Objects.equals(item.getAlgoDescribe(), cause))
+                .map(DictData::getId)
+                .findFirst()
+                .orElse(null);
+
+        List<DictData> list2 = dicDataFeignClient.getDicDataByTypeCode(DicDataTypeEnum.EVENT_TYPE.getCode()).getData();
+        String id2 = list2.stream()
+                .filter(item -> Objects.equals(item.getAlgoDescribe(), type))
+                .map(DictData::getId)
+                .findFirst()
+                .orElse(null);
+
+        LambdaUpdateWrapper<CsEventPO> lambdaUpdateWrapper = new LambdaUpdateWrapper<>();
+        lambdaUpdateWrapper.set(CsEventPO::getAdvanceReason,id1)
+                .set(CsEventPO::getAdvanceType,id2)
+                .eq(CsEventPO::getId,id);
+        this.update(lambdaUpdateWrapper);
+    }
+
 }

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

@@ -1,21 +1,20 @@
-package com.njcn.zlevent.service.impl;
-
-import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
-import com.njcn.csharmonic.pojo.po.CsEventPO;
-import com.njcn.csharmonic.pojo.po.CsEventUserPO;
-import com.njcn.zlevent.mapper.CsEventUserMapper;
-import com.njcn.zlevent.service.ICsEventUserService;
-import org.springframework.stereotype.Service;
-
-/**
- * <p>
- * 暂态事件表 服务实现类
- * </p>
- *
- * @author xuyang
- * @since 2023-08-23
- */
-@Service
-public class CsEventUserServiceImpl extends ServiceImpl<CsEventUserMapper, CsEventUserPO> implements ICsEventUserService {
-
-}
+//package com.njcn.zlevent.service.impl;
+//
+//import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+//import com.njcn.csharmonic.pojo.po.CsEventUserPO;
+//import com.njcn.zlevent.mapper.CsEventUserMapper;
+//import com.njcn.zlevent.service.ICsEventUserService;
+//import org.springframework.stereotype.Service;
+//
+///**
+// * <p>
+// * 暂态事件表 服务实现类
+// * </p>
+// *
+// * @author xuyang
+// * @since 2023-08-23
+// */
+//@Service
+//public class CsEventUserServiceImpl extends ServiceImpl<CsEventUserMapper, CsEventUserPO> implements ICsEventUserService {
+//
+//}

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

@@ -8,6 +8,7 @@ import com.njcn.common.pojo.exception.BusinessException;
 import com.njcn.csdevice.api.DeviceMessageFeignClient;
 import com.njcn.csdevice.api.EquipmentFeignClient;
 import com.njcn.csdevice.enums.AlgorithmResponseEnum;
+import com.njcn.csdevice.param.LineInfoParam;
 import com.njcn.mq.message.AppEventMessage;
 import com.njcn.redis.pojo.enums.AppRedisKey;
 import com.njcn.redis.utils.RedisUtil;
@@ -51,7 +52,9 @@ public class CsWaveAnalysisServiceImpl implements ICsWaveAnalysisService {
             List<WaveTimeDto> list = new ArrayList<>();
             Object object1 = redisUtil.getObjectByKey(AppRedisKey.LINE_POSITION+appEventMessage.getId());
             if (Objects.isNull(object1)){
-                deviceMessageFeignClient.getLineInfo(appEventMessage.getId());
+                LineInfoParam param = new LineInfoParam();
+                param.setNDid(appEventMessage.getId());
+                deviceMessageFeignClient.getLineInfo(param);
             }
             //获取装置id
             String deviceId = equipmentFeignClient.findDevByNDid(appEventMessage.getId()).getData().getId();

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

@@ -11,13 +11,17 @@ import com.njcn.csdevice.api.CsLineFeignClient;
 import com.njcn.csdevice.api.DeviceMessageFeignClient;
 import com.njcn.csdevice.api.EquipmentFeignClient;
 import com.njcn.csdevice.api.WlRecordFeignClient;
+import com.njcn.csdevice.param.LineInfoParam;
 import com.njcn.csdevice.pojo.param.WlRecordParam;
 import com.njcn.csdevice.pojo.po.CsEquipmentDeliveryPO;
 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.cssystem.api.MsgSendFeignClient;
+import com.njcn.cssystem.pojo.param.MsgSendParam;
 import com.njcn.event.common.mapper.WlRmpEventDetailMapper;
+import com.njcn.event.common.service.EventAnalysisService;
 import com.njcn.event.pojo.po.RmpEventDetailPO;
 import com.njcn.influx.pojo.constant.InfluxDBTableConstant;
 import com.njcn.influx.utils.InfluxDbUtils;
@@ -33,10 +37,10 @@ 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.service.AppNotificationService;
+//import com.njcn.zlevent.service.AppNotificationService;
 import com.njcn.zlevent.service.ICsEventService;
 import com.njcn.zlevent.service.IEventService;
-import com.njcn.zlevent.service.SmsNotificationService;
+//import com.njcn.zlevent.service.SmsNotificationService;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.influxdb.InfluxDB;
@@ -79,8 +83,10 @@ public class EventServiceImpl implements IEventService {
     private final WlRmpEventDetailMapper wlRmpEventDetailMapper;
     private final DictTreeFeignClient dictTreeFeignClient;
     private final DeviceMessageFeignClient deviceMessageFeignClient;
-    private final AppNotificationService appNotificationService;
-    private final SmsNotificationService smsNotificationService;
+//    private final AppNotificationService appNotificationService;
+//    private final SmsNotificationService smsNotificationService;
+    private final EventAnalysisService eventAnalysisService;
+    private final MsgSendFeignClient msgSendFeignClient;
 
     @Override
     @DSTransactional
@@ -96,7 +102,9 @@ public class EventServiceImpl implements IEventService {
         }
         //判断监测点是否存在
         if (Objects.isNull(object1)){
-            deviceMessageFeignClient.getLineInfo(appEventMessage.getId());
+            LineInfoParam param = new LineInfoParam();
+            param.setNDid(appEventMessage.getId());
+            deviceMessageFeignClient.getLineInfo(param);
         }
         //获取装置id
         CsEquipmentDeliveryPO po = equipmentFeignClient.findDevByNDid(appEventMessage.getId()).getData();
@@ -186,12 +194,19 @@ public class EventServiceImpl implements IEventService {
                                 csEvent.setLocation("load");
                             }
                         }
+                        String dropZone = eventAnalysisService.determineDropZone(String.valueOf(csEvent.getAmplitude()),String.valueOf(csEvent.getPersistTime()));
+                        csEvent.setLandPoint(dropZone);
+                        AppEventMessage.Param param = new AppEventMessage.Param();
+                        param.setName("Evt_Param_DropZone");
+                        param.setData(dropZone);
+                        params.add(param);
                         //fixme 设备上送的是北京时间，时序数据库录入时 需要utc时间，减去8小时
                         Point point = influxDbUtils.pointBuilder(tableName, item.getDataTimeSec()-8*3600, TimeUnit.SECONDS, tags, fields);
                         BatchPoints batchPoints = BatchPoints.database(influxDbUtils.getDbName()).retentionPolicy("").consistency(InfluxDB.ConsistencyLevel.ALL).build();
                         batchPoints.point(point);
                         records.add(batchPoints.lineProtocol());
                     }
+
                     list1.add(csEvent);
                 }
             }
@@ -211,6 +226,7 @@ public class EventServiceImpl implements IEventService {
                 for (AppEventMessage.DataArray item : dataArray) {
                     double amplitude = 0.0;
                     double persistTime = 0.0;
+                    String dropZone = null;
                     List<AppEventMessage.Param> params = item.getParam();
                     if (CollectionUtil.isNotEmpty(params)) {
                         for (AppEventMessage.Param param : params) {
@@ -220,12 +236,34 @@ public class EventServiceImpl implements IEventService {
                             if (Objects.equals(param.getName(),ZlConstant.EVT_PARAM_TM)) {
                                 persistTime = Double.parseDouble(String.format("%.2f", Double.parseDouble(param.getData().toString())));
                             }
+                            if (Objects.equals(param.getName(),"Evt_Param_DropZone")) {
+                                dropZone = param.getData().toString();
+                            }
                         }
                     }
-                    appNotificationService.sendAppNotification(1, item.getType(), po.getId(), item.getName(), eventTime, id, po.getNdid(),amplitude,persistTime);
+                    MsgSendParam msgSendParam = new MsgSendParam();
+                    msgSendParam.setEventType(1);
+                    msgSendParam.setType("2");
+                    msgSendParam.setDevId(po.getId());
+                    msgSendParam.setEventName(item.getName());
+                    msgSendParam.setEventTime(eventTime);
+                    msgSendParam.setId(id);
+                    msgSendParam.setNDid(po.getNdid());
+                    msgSendParam.setAmplitude(amplitude);
+                    msgSendParam.setPersistTime(persistTime);
+                    msgSendParam.setDropZone(dropZone);
+//                    appNotificationService.sendAppNotification(1, item.getType(), po.getId(), item.getName(), eventTime, id, po.getNdid(),amplitude,persistTime,dropZone);
+                    msgSendFeignClient.appMsgSend(msgSendParam);
                     //如果是暂降事件，则异步发送短信
                     if (Objects.equals(item.getName(), "Evt_Sys_DipStr")) {
-                        smsNotificationService.sendSmsForDipEvent(po.getId(), eventTime,amplitude,persistTime);
+                        MsgSendParam msgSendParam2 = new MsgSendParam();
+                        msgSendParam2.setDevId(po.getId());
+                        msgSendParam2.setEventTime(eventTime);
+                        msgSendParam2.setAmplitude(amplitude);
+                        msgSendParam2.setPersistTime(persistTime);
+                        msgSendParam2.setDropZone(dropZone);
+                        msgSendFeignClient.smsMsgSend(msgSendParam2);
+//                        smsNotificationService.sendSmsForDipEvent(po.getId(), eventTime,amplitude,persistTime,dropZone);
                     }
                 }
             }

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

@@ -14,6 +14,8 @@ import com.njcn.access.enums.TypeEnum;
 import com.njcn.access.pojo.dto.ReqAndResDto;
 import com.njcn.access.pojo.dto.file.FileDto;
 import com.njcn.access.utils.*;
+import com.njcn.advance.api.EventCauseFeignClient;
+import com.njcn.advance.pojo.dto.EventAnalysisDTO;
 import com.njcn.common.config.GeneralInfo;
 import com.njcn.common.pojo.exception.BusinessException;
 import com.njcn.csdevice.api.DeviceFtpFeignClient;
@@ -24,6 +26,7 @@ import com.njcn.csdevice.pojo.dto.CsEquipmentDeliveryDTO;
 import com.njcn.csharmonic.api.WavePicFeignClient;
 import com.njcn.csharmonic.enums.CsHarmonicResponseEnum;
 import com.njcn.csharmonic.pojo.dto.DownloadMakeUpDto;
+import com.njcn.csharmonic.pojo.po.CsEventPO;
 import com.njcn.middle.rocket.domain.BaseMessage;
 import com.njcn.mq.message.AppFileMessage;
 import com.njcn.oss.constant.GeneralConstant;
@@ -85,6 +88,7 @@ public class FileServiceImpl implements IFileService {
     private final FileCommonUtils fileCommonUtils;
     private final DeviceFtpFeignClient deviceFtpFeignClient;
     private final PortableOffLogFeignClient portableOffLogFeignClient;
+    private final EventCauseFeignClient eventCauseFeignClient;
 
     private final CommonProducer commonProducer;
     private final SendMessageUtil sendMessageUtil;
@@ -302,13 +306,21 @@ public class FileServiceImpl implements IFileService {
                         csWaveService.updateCsWave(fileName);
                         //波形文件关联事件
                         filePath = filePath.replaceAll(GeneralConstant.CFG,"").replaceAll(GeneralConstant.DAT,"");
-                        List<String> eventList = correlateEvents(fileInfoDto,filePath,fileName);
+                        List<CsEventPO> eventList = correlateEvents(fileInfoDto,filePath,fileName);
                         if (CollectionUtil.isNotEmpty(eventList) && devModel){
+                            String finalFilePath = filePath;
                             eventList.forEach(item -> {
                                 //波形文件解析成图片
-                                wavePicFeignClient.getWavePics(item);
+                                wavePicFeignClient.getWavePics(item.getId());
+                                //如果是暂降则计算暂降类型和暂降原因
+                                if (Objects.equals(item.getTag(),"Evt_Sys_DipStr")) {
+                                    EventAnalysisDTO var1 = new EventAnalysisDTO();
+                                    var1.setWlFilePath(finalFilePath);
+                                    EventAnalysisDTO dto = eventCauseFeignClient.analysisCauseAndType(var1).getData();
+                                    csEventService.updateEventCauseAndType(item.getId(),dto.getCause(),dto.getType());
+                                }
                                 //同步更新r_mp_event_detail，将波形路径录入
-                                wavePicFeignClient.updateEventById(item);
+                                wavePicFeignClient.updateEventById(item.getId());
                             });
                         }
                     }
@@ -354,13 +366,21 @@ public class FileServiceImpl implements IFileService {
                                     csWaveService.updateCsWave(fileName);
                                     //波形文件关联事件
                                     filePath = filePath.replaceAll(GeneralConstant.CFG, "").replaceAll(GeneralConstant.DAT, "");
-                                    List<String> eventList = correlateEvents(fileInfoDto, filePath, fileName);
+                                    List<CsEventPO> eventList = correlateEvents(fileInfoDto, filePath, fileName);
                                     if (CollectionUtil.isNotEmpty(eventList) && devModel){
+                                        String finalFilePath = filePath;
                                         eventList.forEach(item -> {
                                             //波形文件解析成图片
-                                            wavePicFeignClient.getWavePics(item);
+                                            wavePicFeignClient.getWavePics(item.getId());
+                                            //如果是暂降则计算暂降类型和暂降原因
+                                            if (Objects.equals(item.getTag(),"Evt_Sys_DipStr")) {
+                                                EventAnalysisDTO var1 = new EventAnalysisDTO();
+                                                var1.setWlFilePath(finalFilePath);
+                                                EventAnalysisDTO dto2 = eventCauseFeignClient.analysisCauseAndType(var1).getData();
+                                                csEventService.updateEventCauseAndType(item.getId(),dto2.getCause(),dto2.getType());
+                                            }
                                             //同步更新r_mp_event_detail，将波形路径录入
-                                            wavePicFeignClient.updateEventById(item);
+                                            wavePicFeignClient.updateEventById(item.getId());
                                         });
                                     }
                                 }
@@ -941,8 +961,8 @@ public class FileServiceImpl implements IFileService {
     /**
      * 波形文件关联事件
      */
-    public List<String> correlateEvents(FileInfoDto fileInfoDto, String path, String fileName) {
-        List<String> list = new ArrayList<>();
+    public List<CsEventPO> correlateEvents(FileInfoDto fileInfoDto, String path, String fileName) {
+        List<CsEventPO> list = new ArrayList<>();
         String[] parts = fileName.split(StrUtil.SLASH);
         fileName = parts[parts.length - 1].split("\\.")[0];
         boolean result = csWaveService.findCountByName(fileName);

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

@@ -0,0 +1,134 @@
+package com.njcn.message.consumer;
+
+import com.njcn.access.api.CsHeartbeatFeignClient;
+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.HeartbeatTimeoutMessage;
+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.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.HEARTBEAT_TIMEOUT_TOPIC,
+        consumerGroup = BusinessTopic.HEARTBEAT_TIMEOUT_TOPIC,
+        selectorExpression = BusinessTopic.HeartTag.APF_TAG,
+        consumeThreadNumber = 1,
+        enableMsgTrace = true
+)
+@Slf4j
+public class HeartbeatTimeoutConsumer extends EnhanceConsumerMessageHandler<HeartbeatTimeoutMessage> implements RocketMQListener<HeartbeatTimeoutMessage> {
+
+    @Resource
+    private RedisUtil redisUtil;
+    @Resource
+    private RocketMqLogFeignClient rocketMqLogFeignClient;
+    @Resource
+    private CsHeartbeatFeignClient csHeartbeatFeignClient;
+
+    @Override
+    protected void handleMessage(HeartbeatTimeoutMessage appFileMessage) {
+        csHeartbeatFeignClient.handleHeartbeat(appFileMessage);
+    }
+
+
+    /***
+     * 通过redis分布式锁判断当前消息所处状态
+     * 1、null 查不到该key的数据，属于第一次消费，放行
+     * 2、fail 上次消息消费时发生异常，放行
+     * 3、being processed 正在处理，打回去
+     * 4、success 最近72小时消费成功，避免重复消费，打回去
+     */
+    @Override
+    public boolean filter(HeartbeatTimeoutMessage message) {
+        String keyStatus = redisUtil.getStringByKey(AppRedisKey.RMQ_FILE_CONSUME_KEY.concat(message.getKey()));
+        if (Objects.isNull(keyStatus) || keyStatus.equalsIgnoreCase(MessageStatus.FAIL)) {
+            redisUtil.saveByKeyWithExpire(AppRedisKey.RMQ_FILE_CONSUME_KEY.concat(message.getKey()), MessageStatus.BEING_PROCESSED, 60L);
+            return false;
+        }
+        return true;
+    }
+
+    /**
+     * 消费成功，缓存到redis72小时，避免重复消费
+     */
+    @Override
+    protected void consumeSuccess(HeartbeatTimeoutMessage message) {
+        redisUtil.saveByKeyWithExpire(AppRedisKey.RMQ_FILE_CONSUME_KEY.concat(message.getKey()), MessageStatus.SUCCESS, 300L);
+    }
+
+    /**
+     * 发生异常时，进行错误信息入库保存
+     * 默认没有实现类，子类可以实现该方法，调用feign接口进行入库保存
+     */
+    @Override
+    protected void saveExceptionMsgLog(HeartbeatTimeoutMessage message, String identity, Exception exception) {
+        redisUtil.saveByKeyWithExpire(AppRedisKey.RMQ_FILE_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(HeartbeatTimeoutMessage appFileMessage) {
+        super.dispatchMessage(appFileMessage);
+    }
+}