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

- 移除注释掉的appNotificationService调用代码 - 修复事件服务中消息发送参数的条件判断逻辑 - 更新设备模型服务中监测点模板生成逻辑，支持不同设备类型 - 修复心跳服务中设备离线通知逻辑，暂时关闭消息发送功能 - 添加对直接连接设备和便携式设备的特殊处理逻辑 - 优化监测点名称生成规则，增加输出侧监测点支持
refactor(mqtt): 优化MQTT消息处理机制并修复空指针异常
2026-07-14 14:42:53 +08:00 · 2026-07-08 10:08:52 +08:00
10 changed files with 857 additions and 233 deletions
--- a/iot-access/access-boot/src/main/java/com/njcn/access/config/MqttAccessSchedulerConfig.java
+++ b/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()
+        );
+    }
 }
--- a/iot-access/access-boot/src/main/java/com/njcn/access/config/MqttOptionsFixer.java
+++ b/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;
+    }
+}
--- a/iot-access/access-boot/src/main/java/com/njcn/access/handler/MqttMessageHandler.java
+++ b/iot-access/access-boot/src/main/java/com/njcn/access/handler/MqttMessageHandler.java
--- a/iot-access/access-boot/src/main/java/com/njcn/access/runner/UpdateDevStatusTimer.java
+++ b/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();
    }
--- a/iot-access/access-boot/src/main/java/com/njcn/access/service/impl/CsDevModelServiceImpl.java
+++ b/iot-access/access-boot/src/main/java/com/njcn/access/service/impl/CsDevModelServiceImpl.java
@@ -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();
-                csLineModel.setLineId(IdUtil.fastSimpleUUID());
-                csLineModel.setPid(pId);
-                if (Objects.equals(item.getClDid(),1) || Objects.equals(item.getLocation(), TypeEnum.GUID.getCode())){
-                    csLineModel.setName("电网侧监测点");
-                    csLineModel.setPosition(dicDataFeignClient.getDicDataByCode(DicDataEnum.GRID_SIDE.getCode()).getData().getId());
-                } 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());
-                }
-                result.add(csLineModel);
-            });
+            //根据设备型号获取设备类型，根据设备类型生成监测点模板表数据
+            DictTreeVO vo1 = dictTreeFeignClient.queryByCode(devType).getData();
+            SysDicTreePO po1 = dictTreeFeignClient.queryById(vo1.getPid()).getData();
+            if (Objects.equals(po1.getCode(),"Direct_Connected_Device") || Objects.equals(po1.getCode(),"Portable")){
+                list.forEach(item->{
+                    CsLineModel csLineModel = new CsLineModel();
+                    csLineModel.setLineId(IdUtil.fastSimpleUUID());
+                    csLineModel.setPid(pId);
+                    if (Objects.equals(item.getClDid(),1) || Objects.equals(item.getLocation(), TypeEnum.GUID.getCode())){
+                        csLineModel.setName("电网侧监测点");
+                        csLineModel.setPosition(dicDataFeignClient.getDicDataByCode(DicDataEnum.GRID_SIDE.getCode()).getData().getId());
+                    } 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);
--- a/iot-access/access-boot/src/main/java/com/njcn/access/service/impl/CsHeartServiceImpl.java
+++ b/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)));
--- a/iot-access/access-boot/src/main/java/com/njcn/access/service/impl/HeartbeatServiceImpl.java
+++ b/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.expire(redisKey, HEARTBEAT_EXPIRE_SECONDS);
+        redisUtil.saveByKeyWithExpire(redisKey, currentTime, 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);
    }

--- a/iot-analysis/analysis-stat/stat-boot/src/main/java/com/njcn/stat/service/impl/StatServiceImpl.java
+++ b/iot-analysis/analysis-stat/stat-boot/src/main/java/com/njcn/stat/service/impl/StatServiceImpl.java
@@ -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);
                });
            }
--- a/iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/impl/CsAlarmServiceImpl.java
+++ b/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();
--- a/iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/impl/EventServiceImpl.java
+++ b/iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/impl/EventServiceImpl.java
@@ -247,18 +247,20 @@ public class EventServiceImpl implements IEventService {
                            }
                        }
                    }
-                    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);
-                    msgSendFeignClient.appMsgSend(msgSendParam);
+                    if (Objects.equals(item.getType(),"2")){
+                        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);
+                        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);
        }