refactor(mqtt): 优化MQTT消息处理机制并修复空指针异常
- 修复EventServiceImpl中SagSource为空时的空指针异常 - 在HeartbeatServiceImpl中统一心跳超时级别命名规范 - 将Redis操作合并为原子方法减少网络调用次数 - 新增MQTT消息异步处理线程池避免Paho回调阻塞 - 将MQTT消息处理逻辑重构为异步提交模式提升性能 - 为Cloud事件添加数据报文日志便于调试追踪 - 优化心跳消息处理确保关键操作同步执行
This commit is contained in:
@@ -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()
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
@@ -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();
|
||||
}
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
|
||||
@@ -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);
|
||||
});
|
||||
}
|
||||
|
||||
@@ -312,7 +312,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 +429,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 +440,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);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user