fix(alarm): 修复报警服务中的消息发送逻辑

- 移除注释掉的appNotificationService调用代码
- 修复事件服务中消息发送参数的条件判断逻辑
- 更新设备模型服务中监测点模板生成逻辑，支持不同设备类型
- 修复心跳服务中设备离线通知逻辑，暂时关闭消息发送功能
- 添加对直接连接设备和便携式设备的特殊处理逻辑
- 优化监测点名称生成规则，增加输出侧监测点支持

iot-access/access-boot/src/main/java/com/njcn/access/config/MqttAccessSchedulerConfig.java

@@ -49,4 +49,21 @@ public class MqttAccessSchedulerConfig {
         log.info("初始化MQTT接入信号量, 并发上限={}", concurrencyLimit);
         return new Semaphore(concurrencyLimit);
     }
+
+
+    /**
+     * MQTT消息异步处理线程池
+     * 用于将 @MqttSubscribe 回调中的重活(Feign调用、Kafka发送)异步化
+     * 避免阻塞Paho单线程回调，导致心跳消息排队延迟
+     */
+    @Bean(destroyMethod = "shutdownNow")
+    public ExecutorService mqttMessageExecutor() {
+        return new ThreadPoolExecutor(
+                10, 20,
+                60L, TimeUnit.SECONDS,
+                new LinkedBlockingQueue<>(500),
+                r -> new Thread(r, "mqtt-msg-handler"),
+                new ThreadPoolExecutor.CallerRunsPolicy()
+        );
+    }
 }

iot-access/access-boot/src/main/java/com/njcn/access/config/MqttOptionsFixer.java

@@ -0,0 +1,21 @@
+package com.njcn.access.config;
+
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.config.BeanPostProcessor;
+import org.springframework.stereotype.Component;
+
+@Component
+@Slf4j
+public class MqttOptionsFixer implements BeanPostProcessor {
+
+    @Override
+    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
+        if (bean instanceof MqttConnectOptions) {
+            MqttConnectOptions options = (MqttConnectOptions) bean;
+            options.setMaxInflight(200);
+        }
+        return bean;
+    }
+}

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

@@ -18,7 +18,7 @@ public class UpdateDevStatusTimer {
 
     private final ICsHeartService csHeartService;
 
-    @Scheduled(cron = "0 0/30 * * * ?")
+    @Scheduled(cron = "0 0/10 * * * ?")
     public void timer() {
         csHeartService.updateDevStatus();
     }

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

@@ -43,6 +43,7 @@ import com.njcn.system.pojo.param.CsWaveParam;
 import com.njcn.system.pojo.param.EleEpdPqdParam;
 import com.njcn.system.pojo.param.EleEvtParam;
 import com.njcn.system.pojo.po.EleEpdPqd;
+import com.njcn.system.pojo.po.SysDicTreePO;
 import com.njcn.system.pojo.vo.DictTreeVO;
 import com.njcn.web.utils.RequestUtil;
 import lombok.AllArgsConstructor;
@@ -125,7 +126,7 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
             //2.录入数据集、详细数据(主设备、模块、监测设备、便携式设备...)
             analysisDataSet(templateDto,csDevModelPo.getId());
             //3.录入监测点模板表(记录当前模板有几个监测点，治理类型的模板目前规定1个监测点，电能质量模板根据逻辑子设备来)
-            addCsLineModel(templateDto,csDevModelPo.getId());
+            addCsLineModel(templateDto,csDevModelPo.getId(),devType);
             //4.如果是云前置的模板录入,重新录入完需要将模板关系信息重新录入
             if (Objects.equals(devType,DicDataEnum.DEV_CLD.getCode())) {
                 List<CsEquipmentDeliveryPO> list = eequipmentFeignClient.getAll().getData();
@@ -1205,7 +1206,7 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
     /**
      * 根据模板文件来录入监测点模板
      */
-    private void addCsLineModel(TemplateDto templateDto,String pId) {
+    private void addCsLineModel(TemplateDto templateDto,String pId,String devType) {
         List<CsLineModel> result = new ArrayList<>();
         //fixme 先用数据类型来区分模板的类型
         if (templateDto.getDataList().contains("Apf") || templateDto.getDataList().contains("Dvr")){
@@ -1217,19 +1218,36 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
             result.add(csLineModel);
         } else {
             List<ClDevDto> list = templateDto.getClDevDtoList();
-            list.forEach(item->{
+            //根据设备型号获取设备类型，根据设备类型生成监测点模板表数据
-                CsLineModel csLineModel = new CsLineModel();
+            DictTreeVO vo1 = dictTreeFeignClient.queryByCode(devType).getData();
-                csLineModel.setLineId(IdUtil.fastSimpleUUID());
+            SysDicTreePO po1 = dictTreeFeignClient.queryById(vo1.getPid()).getData();
-                csLineModel.setPid(pId);
+            if (Objects.equals(po1.getCode(),"Direct_Connected_Device") || Objects.equals(po1.getCode(),"Portable")){
-                if (Objects.equals(item.getClDid(),1) || Objects.equals(item.getLocation(), TypeEnum.GUID.getCode())){
+                list.forEach(item->{
-                    csLineModel.setName("电网侧监测点");
+                    CsLineModel csLineModel = new CsLineModel();
-                    csLineModel.setPosition(dicDataFeignClient.getDicDataByCode(DicDataEnum.GRID_SIDE.getCode()).getData().getId());
+                    csLineModel.setLineId(IdUtil.fastSimpleUUID());
-                } else if (Objects.equals(item.getClDid(),2) || Objects.equals(item.getLocation(), TypeEnum.LOAD.getCode())){
+                    csLineModel.setPid(pId);
-                    csLineModel.setName("负载侧监测点");
+                    if (Objects.equals(item.getClDid(),1) || Objects.equals(item.getLocation(), TypeEnum.GUID.getCode())){
-                    csLineModel.setPosition(dicDataFeignClient.getDicDataByCode(DicDataEnum.LOAD_SIDE.getCode()).getData().getId());
+                        csLineModel.setName("电网侧监测点");
-                }
+                        csLineModel.setPosition(dicDataFeignClient.getDicDataByCode(DicDataEnum.GRID_SIDE.getCode()).getData().getId());
-                result.add(csLineModel);
+                    } else if (Objects.equals(item.getClDid(),2) || Objects.equals(item.getLocation(), TypeEnum.LOAD.getCode())){
-            });
+                        csLineModel.setName("负载侧监测点");
+                        csLineModel.setPosition(dicDataFeignClient.getDicDataByCode(DicDataEnum.LOAD_SIDE.getCode()).getData().getId());
+                    } else {
+                        csLineModel.setName("输出侧监测点");
+                        csLineModel.setPosition(dicDataFeignClient.getDicDataByCode(DicDataEnum.OUTPUT_SIDE.getCode()).getData().getId());
+                    }
+                    result.add(csLineModel);
+                });
+            } else {
+                list.forEach(item->{
+                    CsLineModel csLineModel = new CsLineModel();
+                    csLineModel.setLineId(IdUtil.fastSimpleUUID());
+                    csLineModel.setPid(pId);
+                    csLineModel.setName(item.getClDid() + "#监测点");
+                    csLineModel.setPosition("");
+                    result.add(csLineModel);
+                });
+            }
         }
         if (CollectionUtil.isNotEmpty(result)){
             csLineModelService.addList(result);

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

@@ -122,7 +122,8 @@ public class CsHeartServiceImpl implements ICsHeartService {
         csEquipmentDeliveryService.updateRunStatusBynDid(nDid, AccessEnum.OFFLINE.getCode());
         //装置调整为注册状态
         csEquipmentDeliveryService.updateStatusBynDid(nDid,AccessEnum.REGISTERED.getCode(),null,null);
-        sendMessage(nDid);
+        //设备离线通知关闭
+        //sendMessage(nDid);
         //记录装置掉线时间
         PqsCommunicateDto dto = new PqsCommunicateDto();
         dto.setTime(LocalDateTime.now().format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_PATTERN)));

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

@@ -21,7 +21,7 @@ public class HeartbeatServiceImpl implements IHeartbeatService {
     @Resource
     private RedisUtil redisUtil;
     private static final String HEARTBEAT_REDIS_KEY_PREFIX = "HEARTBEAT:";
-    private static final int DELAY_LEVEL_4MIN = 7;
+    private static final int DELAY_LEVEL_3MIN = 7;
     private static final long HEARTBEAT_EXPIRE_SECONDS = 180;
 
     @Override
@@ -29,14 +29,12 @@ public class HeartbeatServiceImpl implements IHeartbeatService {
         String redisKey = HEARTBEAT_REDIS_KEY_PREFIX + nDid;
         long currentTime = System.currentTimeMillis();
 
-        redisUtil.saveByKey(redisKey, currentTime);
+        redisUtil.saveByKeyWithExpire(redisKey, currentTime, HEARTBEAT_EXPIRE_SECONDS);
-        redisUtil.expire(redisKey, HEARTBEAT_EXPIRE_SECONDS);
 
         HeartbeatTimeoutMessage message = new HeartbeatTimeoutMessage();
-
         message.setNDid(nDid);
         message.setTimestamp(currentTime);
-        message.setDelayLevel(DELAY_LEVEL_4MIN);
+        message.setDelayLevel(DELAY_LEVEL_3MIN);
         heartbeatTimeoutMessageTemplate.sendMember(message);
     }
 

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

@@ -308,6 +308,9 @@ public class StatServiceImpl implements IStatService {
             tags.put(InfluxDBTableConstant.LINE_ID,lineId);
             tags.put(InfluxDBTableConstant.PHASIC_TYPE,dataArrayList.get(i).getPhase());
             tags.put(InfluxDBTableConstant.VALUE_TYPE,statMethod.toUpperCase());
+            tags.put(InfluxDBTableConstant.CL_DID,clDid.toString());
+            tags.put(InfluxDBTableConstant.PROCESS,process.toString());
+
             if (Objects.isNull(item.getDataTag())) {
                 tags.put(InfluxDBTableConstant.QUALITY_FLAG,"0");
             } else {
@@ -320,8 +323,6 @@ public class StatServiceImpl implements IStatService {
             } 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();
@@ -387,8 +388,8 @@ public class StatServiceImpl implements IStatService {
                 data.forEach(item->{
                     DataArrayLiteDto dataArrayLiteDto = new DataArrayLiteDto();
                     dataArrayLiteDto.setName(item.getName());
-                    dataArrayLiteDto.setInfluxDbName(item.getInfluxDbName());
                     dataArrayLiteDto.setPhase(item.getPhase());
+                    dataArrayLiteDto.setInfluxDbName(item.getInfluxDbName());
                     result.add(dataArrayLiteDto);
                 });
             }

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

@@ -21,7 +21,6 @@ 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.ICsAlarmService;
 import com.njcn.zlevent.service.ICsEventLogsService;
 import com.njcn.zlevent.service.ICsEventService;
@@ -138,11 +137,9 @@ public class CsAlarmServiceImpl extends ServiceImpl<CsEventMapper, CsEventPO> im
                     if (Objects.isNull(item.getCode())){
                         msgSendParam.setEventName(item.getName());
                         msgSendFeignClient.appMsgSend(msgSendParam);
-//                        appNotificationService.sendAppNotification(2,item.getType(),po.getId(),item.getName(),eventTime,id,po.getNdid(),null,null,null);
                     } else {
                         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/EventServiceImpl.java

@@ -247,18 +247,20 @@ public class EventServiceImpl implements IEventService {
                             }
                         }
                     }
-                    MsgSendParam msgSendParam = new MsgSendParam();
+                    if (Objects.equals(item.getType(),"2")){
-                    msgSendParam.setEventType(1);
+                        MsgSendParam msgSendParam = new MsgSendParam();
-                    msgSendParam.setType("2");
+                        msgSendParam.setEventType(1);
-                    msgSendParam.setDevId(po.getId());
+                        msgSendParam.setType("2");
-                    msgSendParam.setEventName(item.getName());
+                        msgSendParam.setDevId(po.getId());
-                    msgSendParam.setEventTime(eventTime);
+                        msgSendParam.setEventName(item.getName());
-                    msgSendParam.setId(id);
+                        msgSendParam.setEventTime(eventTime);
-                    msgSendParam.setNDid(po.getNdid());
+                        msgSendParam.setId(id);
-                    msgSendParam.setAmplitude(amplitude);
+                        msgSendParam.setNDid(po.getNdid());
-                    msgSendParam.setPersistTime(persistTime);
+                        msgSendParam.setAmplitude(amplitude);
-                    msgSendParam.setDropZone(dropZone);
+                        msgSendParam.setPersistTime(persistTime);
-                    msgSendFeignClient.appMsgSend(msgSendParam);
+                        msgSendParam.setDropZone(dropZone);
+                        msgSendFeignClient.appMsgSend(msgSendParam);
+                    }
                     //如果是暂降事件，则异步发送短信
                     if (Objects.equals(item.getName(), "Evt_Sys_DipStr")) {
                         MsgSendParam msgSendParam2 = new MsgSendParam();
@@ -312,7 +314,9 @@ public class EventServiceImpl implements IEventService {
         rmpEventDetailPo.setDealFlag(0);
         rmpEventDetailPo.setFileFlag(0);
         rmpEventDetailPo.setPhase(item.getPhase());
-        rmpEventDetailPo.setSagsource(getSagSource(item.getSagSource()));
+        if (!Objects.isNull(item.getSagSource())) {
+            rmpEventDetailPo.setSagsource(getSagSource(item.getSagSource()));
+        }
         rmpEventDetailPo.setSeverity(item.getSeverity());
         wlRmpEventDetailMapper.insert(rmpEventDetailPo);
     }
@@ -427,6 +431,7 @@ public class EventServiceImpl implements IEventService {
 
     @Override
     public void getCldEventData(CldLogMessage cldLogMessage) {
+        log.info("数据报文=====：" + cldLogMessage);
         CsEventPO po = new CsEventPO();
         po.setStartTime(LocalDateTime.parse(cldLogMessage.getTime(), DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
         po.setTag(cldLogMessage.getLog());
@@ -437,7 +442,7 @@ public class EventServiceImpl implements IEventService {
         //前置告警
         if (Objects.equals(cldLogMessage.getLevel(),"process")) {
             //这边将前置服务器id当作设备id
-                po.setDeviceId(cldLogMessage.getNodeId());
+            po.setDeviceId(cldLogMessage.getNodeId());
             po.setClDid(Integer.valueOf(cldLogMessage.getProcessNo()));
             po.setType(4);
         }