From c6f46cc319559211b934c197e972372fdc41973c Mon Sep 17 00:00:00 2001 From: xy <748613696@qq.com> Date: Wed, 8 Jul 2026 10:08:52 +0800 Subject: [PATCH] =?UTF-8?q?refactor(mqtt):=20=E4=BC=98=E5=8C=96MQTT?= =?UTF-8?q?=E6=B6=88=E6=81=AF=E5=A4=84=E7=90=86=E6=9C=BA=E5=88=B6=E5=B9=B6?= =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E7=A9=BA=E6=8C=87=E9=92=88=E5=BC=82=E5=B8=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 修复EventServiceImpl中SagSource为空时的空指针异常 - 在HeartbeatServiceImpl中统一心跳超时级别命名规范 - 将Redis操作合并为原子方法减少网络调用次数 - 新增MQTT消息异步处理线程池避免Paho回调阻塞 - 将MQTT消息处理逻辑重构为异步提交模式提升性能 - 为Cloud事件添加数据报文日志便于调试追踪 - 优化心跳消息处理确保关键操作同步执行 --- .../config/MqttAccessSchedulerConfig.java | 17 + .../njcn/access/config/MqttOptionsFixer.java | 21 + .../access/handler/MqttMessageHandler.java | 948 ++++++++++++++---- .../access/runner/UpdateDevStatusTimer.java | 2 +- .../service/impl/HeartbeatServiceImpl.java | 8 +- .../stat/service/impl/StatServiceImpl.java | 7 +- .../service/impl/EventServiceImpl.java | 7 +- 7 files changed, 808 insertions(+), 202 deletions(-) create mode 100644 iot-access/access-boot/src/main/java/com/njcn/access/config/MqttOptionsFixer.java diff --git 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 index 48328a7..f562650 100644 --- 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() + ); + } } \ No newline at end of file diff --git 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 new file mode 100644 index 0000000..96778e4 --- /dev/null +++ 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; + } +} diff --git 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 index 5358dc1..7e4d064 100644 --- 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 @@ -61,6 +61,7 @@ import java.time.format.DateTimeFormatter; import java.time.format.DateTimeFormatterBuilder; import java.time.temporal.ChronoField; import java.util.*; +import java.util.concurrent.ExecutorService; import java.util.stream.Collectors; import static org.assertj.core.api.AssertionsForClassTypes.assertThat; @@ -100,27 +101,31 @@ public class MqttMessageHandler { private final IHeartbeatService heartbeatService; private final LogMessageTemplate logMessageTemplate; private final CsDeviceRegistryFeignClient csDeviceRegistryFeignClient; + private final ExecutorService mqttMessageExecutor; @Autowired Validator validator; @MqttSubscribe(value = "/Dev/DevTopic/{edgeId}",qos = 1) - @Transactional(rollbackFor = Exception.class) public void devTopic(String topic, MqttMessage message, @NamedValue("edgeId") String nDid, @Payload String payload){ - //业务流程开始 Gson gson = new Gson(); ReqAndResParam.Res res = gson.fromJson(new String(message.getPayload(), StandardCharsets.UTF_8), ReqAndResParam.Res.class); - //日志记录 + mqttMessageExecutor.submit(() -> { + try { + doDevTopic(nDid, res); + } catch (Exception e) { + log.error("devTopic业务处理异常, nDid={}", nDid, e); + } + }); + } + + private void doDevTopic(String nDid, ReqAndResParam.Res res) { LogMessage logDto = new LogMessage(); logDto.setUserIndex("系统"); logDto.setLoginName("系统"); logDto.setOperate("系统端收到装置端"+nDid+"发送的主题信息，code = " + res.getCode()); logDto.setResult(1); logMessageTemplate.sendMember(logDto); - //检验传递的参数是否准确 Set> validate = validator.validate(res); -// validate.forEach(constraintViolation -> { -// System.out.println(constraintViolation.getMessage()); -// }); if (Objects.equals(res.getCode(),AccessEnum.SUCCESS.getCode())){ if (Objects.equals(res.getType(), Integer.parseInt(TypeEnum.TYPE_16.getCode()))){ List list = new ArrayList<>(); @@ -144,32 +149,79 @@ public class MqttMessageHandler { logDto.setResult(0); logDto.setFailReason(AccessResponseEnum.MESSAGE_TYPE_ERROR.getMessage()); logMessageTemplate.sendMember(logDto); - //log.info(AccessResponseEnum.MESSAGE_TYPE_ERROR.getMessage()); } } else { logDto.setResult(0); logDto.setFailReason(AccessResponseEnum.RESPONSE_ERROR.getMessage()); logMessageTemplate.sendMember(logDto); - //log.info(AccessResponseEnum.RESPONSE_ERROR.getMessage()); } } - /** - * 装置注册应答 - * 1.收到注册信息，修改装置出厂表，装置的状态，调整为注册；然后开始接入流程 - * 2.询问当前装置类型的模板。有则完成接入；没有则告警出来，需要人工手动上传模板信息 - * @param topic - * @param message - * @param payload - */ +// @MqttSubscribe(value = "/Dev/DevTopic/{edgeId}",qos = 1) +// public void devTopic(String topic, MqttMessage message, @NamedValue("edgeId") String nDid, @Payload String payload){ +// //业务流程开始 +// Gson gson = new Gson(); +// ReqAndResParam.Res res = gson.fromJson(new String(message.getPayload(), StandardCharsets.UTF_8), ReqAndResParam.Res.class); +// //日志记录 +// LogMessage logDto = new LogMessage(); +// logDto.setUserIndex("系统"); +// logDto.setLoginName("系统"); +// logDto.setOperate("系统端收到装置端"+nDid+"发送的主题信息，code = " + res.getCode()); +// logDto.setResult(1); +// logMessageTemplate.sendMember(logDto); +// //检验传递的参数是否准确 +// Set> validate = validator.validate(res); +//// validate.forEach(constraintViolation -> { +//// System.out.println(constraintViolation.getMessage()); +//// }); +// if (Objects.equals(res.getCode(),AccessEnum.SUCCESS.getCode())){ +// if (Objects.equals(res.getType(), Integer.parseInt(TypeEnum.TYPE_16.getCode()))){ +// List list = new ArrayList<>(); +// Map> map = (Map>)res.getMsg(); +// List topicList = map.get("Topic"); +// topicList.forEach(item->{ +// CsTopic csTopic = new CsTopic(); +// csTopic.setNDid(nDid); +// csTopic.setTopic(item); +// String version = item.split("/")[3]; +// if (version.startsWith("V")){ +// csTopic.setVersion(version); +// } +// list.add(csTopic); +// }); +// csTopicService.addTopic(nDid,list); +// String version = list.stream().map(CsTopic::getVersion).filter(Objects::nonNull).findFirst().orElse("V1"); +// redisUtil.saveByKeyWithExpire(nDid +":version",version,30L); +// logMessageTemplate.sendMember(logDto); +// } else { +// logDto.setResult(0); +// logDto.setFailReason(AccessResponseEnum.MESSAGE_TYPE_ERROR.getMessage()); +// logMessageTemplate.sendMember(logDto); +// //log.info(AccessResponseEnum.MESSAGE_TYPE_ERROR.getMessage()); +// } +// } else { +// logDto.setResult(0); +// logDto.setFailReason(AccessResponseEnum.RESPONSE_ERROR.getMessage()); +// logMessageTemplate.sendMember(logDto); +// //log.info(AccessResponseEnum.RESPONSE_ERROR.getMessage()); +// } +// } + @MqttSubscribe(value = "/Dev/DevReg/{edgeId}",qos = 1) - @Transactional(rollbackFor = Exception.class) public void devOperation(String topic, MqttMessage message, @NamedValue("edgeId") String nDid, @Payload String payload){ log.info("收到注册应答响应--->{}", nDid); - //业务处理 Gson gson = new Gson(); ReqAndResDto.Res res = gson.fromJson(new String(message.getPayload(), StandardCharsets.UTF_8), ReqAndResDto.Res.class); - //日志记录 + mqttMessageExecutor.submit(() -> { + try { + doDevOperation(nDid, res); + } catch (Exception e) { + log.error("devOperation业务处理异常, nDid={}", nDid, e); + } + }); + } + + private void doDevOperation(String nDid, ReqAndResDto.Res res) { LogMessage logDto = new LogMessage(); logDto.setUserIndex("系统"); logDto.setLoginName("系统"); @@ -178,7 +230,6 @@ public class MqttMessageHandler { logMessageTemplate.sendMember(logDto); if (Objects.equals(res.getCode(),AccessEnum.SUCCESS.getCode())){ if (Objects.equals(res.getType(),Integer.parseInt(TypeEnum.TYPE_17.getCode()))){ - //询问模板数据 ReqAndResDto.Req reqAndResParam = new ReqAndResDto.Req(); reqAndResParam.setMid(1); reqAndResParam.setDid(0); @@ -187,7 +238,6 @@ public class MqttMessageHandler { reqAndResParam.setExpire(-1); String version = csTopicService.getVersion(nDid); publisher.send("/Pfm/DevCmd/"+version+"/"+nDid,new Gson().toJson(reqAndResParam),1,false); - //记录日志 logDto.setUserIndex("系统"); logDto.setLoginName("系统"); logDto.setOperate("注册阶段：系统端向装置端"+nDid+"发送询问模板请求"); @@ -197,63 +247,101 @@ public class MqttMessageHandler { logDto.setResult(0); logDto.setFailReason(AccessResponseEnum.MESSAGE_TYPE_ERROR.getMessage()); logMessageTemplate.sendMember(logDto); - //log.info(AccessResponseEnum.MESSAGE_TYPE_ERROR.getMessage()); } } else { logDto.setResult(0); logDto.setFailReason(AccessResponseEnum.REGISTER_RESPONSE_ERROR.getMessage()); logMessageTemplate.sendMember(logDto); - //log.info(AccessResponseEnum.REGISTER_RESPONSE_ERROR.getMessage()); } } - /** - * 设备响应 - * @param topic - * @param message - * @param version - * @param nDid - * @param payload - */ +// /** +// * 装置注册应答 +// * 1.收到注册信息，修改装置出厂表，装置的状态，调整为注册；然后开始接入流程 +// * 2.询问当前装置类型的模板。有则完成接入；没有则告警出来，需要人工手动上传模板信息 +// * @param topic +// * @param message +// * @param payload +// */ +// @MqttSubscribe(value = "/Dev/DevReg/{edgeId}",qos = 1) +// public void devOperation(String topic, MqttMessage message, @NamedValue("edgeId") String nDid, @Payload String payload){ +// log.info("收到注册应答响应--->{}", nDid); +// //业务处理 +// Gson gson = new Gson(); +// ReqAndResDto.Res res = gson.fromJson(new String(message.getPayload(), StandardCharsets.UTF_8), ReqAndResDto.Res.class); +// //日志记录 +// LogMessage logDto = new LogMessage(); +// logDto.setUserIndex("系统"); +// logDto.setLoginName("系统"); +// logDto.setOperate("系统端收到装置端"+nDid+"注册应答响应，code = " + res.getCode()); +// logDto.setResult(1); +// logMessageTemplate.sendMember(logDto); +// if (Objects.equals(res.getCode(),AccessEnum.SUCCESS.getCode())){ +// if (Objects.equals(res.getType(),Integer.parseInt(TypeEnum.TYPE_17.getCode()))){ +// //询问模板数据 +// ReqAndResDto.Req reqAndResParam = new ReqAndResDto.Req(); +// reqAndResParam.setMid(1); +// reqAndResParam.setDid(0); +// reqAndResParam.setPri(AccessEnum.FIRST_CHANNEL.getCode()); +// reqAndResParam.setType(Integer.parseInt(TypeEnum.TYPE_3.getCode())); +// reqAndResParam.setExpire(-1); +// String version = csTopicService.getVersion(nDid); +// publisher.send("/Pfm/DevCmd/"+version+"/"+nDid,new Gson().toJson(reqAndResParam),1,false); +// //记录日志 +// logDto.setUserIndex("系统"); +// logDto.setLoginName("系统"); +// logDto.setOperate("注册阶段：系统端向装置端"+nDid+"发送询问模板请求"); +// logDto.setResult(1); +// logMessageTemplate.sendMember(logDto); +// } else { +// logDto.setResult(0); +// logDto.setFailReason(AccessResponseEnum.MESSAGE_TYPE_ERROR.getMessage()); +// logMessageTemplate.sendMember(logDto); +// //log.info(AccessResponseEnum.MESSAGE_TYPE_ERROR.getMessage()); +// } +// } else { +// logDto.setResult(0); +// logDto.setFailReason(AccessResponseEnum.REGISTER_RESPONSE_ERROR.getMessage()); +// logMessageTemplate.sendMember(logDto); +// //log.info(AccessResponseEnum.REGISTER_RESPONSE_ERROR.getMessage()); +// } +// } + @MqttSubscribe(value = "/Pfm/DevRsp/{version}/{edgeId}",qos = 1) - @Transactional(rollbackFor = Exception.class) - public void devAccessOperation(String topic, MqttMessage message, @NamedValue("version") String version, @NamedValue("edgeId") String nDid, @Payload String payload) throws InterruptedException { - //日志实体 + public void devAccessOperation(String topic, MqttMessage message, @NamedValue("version") String version, @NamedValue("edgeId") String nDid, @Payload String payload) { LogMessage logDto = new LogMessage(); logDto.setUserIndex("系统"); logDto.setLoginName("系统"); logDto.setResult(1); - //业务处理 Gson gson = new Gson(); ReqAndResDto.Res res = gson.fromJson(new String(message.getPayload(), StandardCharsets.UTF_8), ReqAndResDto.Res.class); - //redisUtil.saveByKeyWithExpire("devResponse",res.getCode(),5L); + mqttMessageExecutor.submit(() -> { + try { + doDevAccessOperation(nDid, version, res, logDto, gson); + } catch (Exception e) { + log.error("devAccessOperation业务处理异常, nDid={}", nDid, e); + } + }); + } + + private void doDevAccessOperation(String nDid, String version, ReqAndResDto.Res res, LogMessage logDto, Gson gson) throws InterruptedException { if (Objects.equals(res.getCode(),AccessEnum.SUCCESS.getCode())) { switch (res.getType()){ - /** - * 装置类型模板应答 - * 1.判断网关的类型 - * 2.直联设备的DevCfg和DevMod是以直联设备为准，上送平台端，平台端保存。通过校验DevMod模板信息来从平台端模板池中选取对应的模板，如果找不到匹配模板需告警提示人工干预处理。 - * 3.平台端需读取装置的DevMod来判断网关支持的设备模板（包含设备型号和模板版本），根据app提交的接入子设备DID匹配数据模板（型号及版本），生成DevCfg下发给网关，网关根据下发信息生成就地设备点表。 - */ case 4611: logDto.setOperate("系统端收到装置端"+nDid+"模板应答，code = " + res.getCode()); logMessageTemplate.sendMember(logDto); - //log.info("{},装置模板应答,应答code {}",nDid,res.getCode()); - ModelDto modelDto = gson.fromJson(new String(message.getPayload(), StandardCharsets.UTF_8), ModelDto.class); + ModelDto modelDto = gson.fromJson(new Gson().toJson(res), ModelDto.class); List list = modelDto.getMsg().getDevMod(); List list2 = modelDto.getMsg().getDevCfg(); if (CollectionUtils.isEmpty(list)) { - //log.error(AccessResponseEnum.MODEL_VERSION_ERROR.getMessage()); logDto.setOperate("查看装置端模板报文数据"); logDto.setResult(0); logDto.setFailReason(AccessResponseEnum.MODEL_VERSION_ERROR.getMessage()); logMessageTemplate.sendMember(logDto); - //有异常删除缓存的模板信息 redisUtil.delete(AppRedisKey.MODEL + nDid); log.error("{}{}", nDid, AccessResponseEnum.MODEL_VERSION_ERROR.getMessage()); throw new BusinessException(AccessResponseEnum.MODEL_VERSION_ERROR); } - //校验前置传递的装置模板库中是否存在 List modelList = new ArrayList<>(); list.forEach(item->{ Integer did = null; @@ -265,12 +353,10 @@ public class MqttMessageHandler { CsModelDto csModelDto = new CsModelDto(); CsDevModelPO po = devModelFeignClient.findModel(item.getDevType(),item.getVersionNo(),item.getVersionDate()).getData(); if (Objects.isNull(po)){ - //log.error(AccessResponseEnum.MODEL_NO_FIND.getMessage()); logDto.setOperate(nDid + "查询系统中是否存在模板"); logDto.setResult(0); logDto.setFailReason(AccessResponseEnum.MODEL_NO_FIND.getMessage()); logMessageTemplate.sendMember(logDto); - //有异常删除缓存的模板信息 redisUtil.delete(AppRedisKey.MODEL + nDid); throw new BusinessException(AccessResponseEnum.MODEL_NO_FIND); } @@ -281,7 +367,6 @@ public class MqttMessageHandler { logDto.setResult(0); logDto.setFailReason(AccessResponseEnum.MODULE_NUMBER_IS_NULL.getMessage()); logMessageTemplate.sendMember(logDto); - //有异常删除缓存的模板信息 redisUtil.delete(AppRedisKey.MODEL + nDid); throw new BusinessException(AccessResponseEnum.MODULE_NUMBER_IS_NULL); } @@ -293,10 +378,8 @@ public class MqttMessageHandler { csModelDto.setType(po.getType()); modelList.add(csModelDto); }); - //存储模板id String key2 = AppRedisKey.MODEL + nDid; redisUtil.saveByKey(key2,modelList); - //存储监测点模板信息，用于界面回显 List modelId = modelList.stream().map(CsModelDto::getModelId).collect(Collectors.toList()); List lineList = csLineModelService.getMonitorNumByModelId(modelId); String key = AppRedisKey.LINE + nDid; @@ -307,35 +390,24 @@ public class MqttMessageHandler { logDto.setOperate("系统端收到装置端"+nDid+"接入应答，code = " + res.getCode()); logDto.setResult(1); logMessageTemplate.sendMember(logDto); - //log.info("{},收到接入应答响应,应答code {}",nDid,res.getCode()); if (Objects.equals(res.getCode(),AccessEnum.SUCCESS.getCode())){ int mid = 1; - //修改装置状态 csEquipmentDeliveryService.updateStatusBynDid(nDid,AccessEnum.ACCESS.getCode(),null,null); csEquipmentDeliveryService.updateRunStatusBynDid(nDid,AccessEnum.ONLINE.getCode()); - //记录设备上线 PqsCommunicateDto dto = new PqsCommunicateDto(); dto.setTime(LocalDateTime.now().format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_PATTERN))); dto.setDevId(nDid); dto.setType(1); dto.setDescription("通讯正常"); csCommunicateFeignClient.insertion(dto); - //询问设备软件信息 askDevData(nDid,version,1,mid); - //更新治理监测点信息和设备容量 askDevData(nDid,version,2,(res.getMid()+1)); - //更新电网侧、负载侧监测点信息 askDevData(nDid,version,3,(res.getMid()+1)); - //接入后系统重置装置心跳 heartbeatService.receiveHeartbeat(nDid); - //修改redis的mid redisUtil.saveByKey(AppRedisKey.DEVICE_MID + nDid,1); - //接入成功标识 redisUtil.saveByKeyWithExpire("online" + nDid,"online",10L); - //录波任务倒计时 redisUtil.saveByKeyWithExpire("startFile:" + nDid,null,60L); } else { - //log.info(AccessResponseEnum.ACCESS_RESPONSE_ERROR.getMessage()); logDto.setResult(0); logDto.setFailReason(AccessResponseEnum.ACCESS_RESPONSE_ERROR.getMessage()); logMessageTemplate.sendMember(logDto); @@ -351,17 +423,13 @@ public class MqttMessageHandler { logDto.setOperate("系统端收到装置端"+nDid+"更新设备软件信息报文，code = " + res.getCode()); logDto.setResult(1); RspDataDto.SoftInfo softInfo = JSON.parseObject(JSON.toJSONString(rspDataDto.getDataArray()), RspDataDto.SoftInfo.class); - //记录设备软件信息 CsSoftInfoPO csSoftInfoPo = new CsSoftInfoPO(); BeanUtils.copyProperties(softInfo,csSoftInfoPo); String id = IdUtil.fastSimpleUUID(); csSoftInfoPo.setId(id); DateTimeFormatter formatter = new DateTimeFormatterBuilder() - // 优先尝试紧凑格式 .optionalStart().appendPattern("yyyyMMdd").optionalEnd() - // 再尝试带横线格式 .optionalStart().appendPattern("yyyy-MM-dd").optionalEnd() - // 默认时间部分 .parseDefaulting(ChronoField.HOUR_OF_DAY, 0) .parseDefaulting(ChronoField.MINUTE_OF_HOUR, 0) .parseDefaulting(ChronoField.SECOND_OF_MINUTE, 0) @@ -370,7 +438,6 @@ public class MqttMessageHandler { assertThat(localDateTime).isNotNull(); csSoftInfoPo.setAppDate(localDateTime); csSoftInfoFeignClient.saveSoftInfo(csSoftInfoPo); - //更新设备软件id 先看是否存在软件信息,删除 然后在录入 CsEquipmentDeliveryPO po = equipmentFeignClient.findDevByNDid(nDid).getData(); String soft = po.getSoftinfoId(); if (StringUtil.isNotBlank(soft)){ @@ -386,7 +453,6 @@ public class MqttMessageHandler { redisUtil.saveByKeyWithExpire("lineInfo:"+nDid,devInfo,30L); List list3 = new ArrayList<>(); boolean hasZeroClDid = devInfo.stream().anyMatch(item -> item.getClDid() == 0); - //治理设备 if (hasZeroClDid) { logDto.setOperate("系统端收到装置端"+nDid+"更新APF容量报文，code = " + res.getCode()); logDto.setResult(1); @@ -394,7 +460,6 @@ public class MqttMessageHandler { if (Objects.equals(item.getClDid(),0)){ updateLineInfo(nDid,item); } - //2.录入各个模块设备容量 CsDeviceRegistry csDeviceRegistry = csDeviceRegistryFeignClient.queryByCurrentNdidAndClDid(nDid, 0).getData(); String lineId; if (Objects.isNull(csDeviceRegistry)) { @@ -416,9 +481,7 @@ public class MqttMessageHandler { csDevCapacity.setCapacity(Objects.isNull(item.getCapacityA())?0.0:item.getCapacityA()); list3.add(csDevCapacity); }); - } - //其余设备 - else { + } else { logDto.setOperate("系统端收到装置端"+nDid+"更新监测点台账报文，code = " + res.getCode()); logDto.setResult(1); devInfo.forEach(item->{ @@ -427,13 +490,11 @@ public class MqttMessageHandler { } if (CollectionUtil.isNotEmpty(list3)) { devCapacityFeignClient.addList(list3); - //3.更新设备模块个数 equipmentFeignClient.updateModuleNumber(nDid,(devInfo.size()-1)); } } else if (Objects.equals(res.getDid(),2)) { logDto.setOperate("系统端收到装置端"+nDid+"更新电网侧、负载侧监测点信息报文，code = " + res.getCode()); logDto.setResult(1); - //1.更新电网侧、负载侧监测点相关信息 devInfo.forEach(item->{ updateLineInfo(nDid,item); }); @@ -494,7 +555,6 @@ public class MqttMessageHandler { Thread.sleep(5000); } else { String result = getEnum(res.getCode()); - //log.info(result); logDto.setOperate("装置响应"); logDto.setResult(0); logDto.setFailReason(result); @@ -503,6 +563,302 @@ public class MqttMessageHandler { } } + +// /** +// * 设备响应 +// * @param topic +// * @param message +// * @param version +// * @param nDid +// * @param payload +// */ +// @MqttSubscribe(value = "/Pfm/DevRsp/{version}/{edgeId}",qos = 1) +// public void devAccessOperation(String topic, MqttMessage message, @NamedValue("version") String version, @NamedValue("edgeId") String nDid, @Payload String payload) throws InterruptedException { +// //日志实体 +// LogMessage logDto = new LogMessage(); +// logDto.setUserIndex("系统"); +// logDto.setLoginName("系统"); +// logDto.setResult(1); +// //业务处理 +// Gson gson = new Gson(); +// ReqAndResDto.Res res = gson.fromJson(new String(message.getPayload(), StandardCharsets.UTF_8), ReqAndResDto.Res.class); +// //redisUtil.saveByKeyWithExpire("devResponse",res.getCode(),5L); +// if (Objects.equals(res.getCode(),AccessEnum.SUCCESS.getCode())) { +// switch (res.getType()){ +// /** +// * 装置类型模板应答 +// * 1.判断网关的类型 +// * 2.直联设备的DevCfg和DevMod是以直联设备为准，上送平台端，平台端保存。通过校验DevMod模板信息来从平台端模板池中选取对应的模板，如果找不到匹配模板需告警提示人工干预处理。 +// * 3.平台端需读取装置的DevMod来判断网关支持的设备模板（包含设备型号和模板版本），根据app提交的接入子设备DID匹配数据模板（型号及版本），生成DevCfg下发给网关，网关根据下发信息生成就地设备点表。 +// */ +// case 4611: +// logDto.setOperate("系统端收到装置端"+nDid+"模板应答，code = " + res.getCode()); +// logMessageTemplate.sendMember(logDto); +// //log.info("{},装置模板应答,应答code {}",nDid,res.getCode()); +// ModelDto modelDto = gson.fromJson(new String(message.getPayload(), StandardCharsets.UTF_8), ModelDto.class); +// List list = modelDto.getMsg().getDevMod(); +// List list2 = modelDto.getMsg().getDevCfg(); +// if (CollectionUtils.isEmpty(list)) { +// //log.error(AccessResponseEnum.MODEL_VERSION_ERROR.getMessage()); +// logDto.setOperate("查看装置端模板报文数据"); +// logDto.setResult(0); +// logDto.setFailReason(AccessResponseEnum.MODEL_VERSION_ERROR.getMessage()); +// logMessageTemplate.sendMember(logDto); +// //有异常删除缓存的模板信息 +// redisUtil.delete(AppRedisKey.MODEL + nDid); +// log.error("{}{}", nDid, AccessResponseEnum.MODEL_VERSION_ERROR.getMessage()); +// throw new BusinessException(AccessResponseEnum.MODEL_VERSION_ERROR); +// } +// //校验前置传递的装置模板库中是否存在 +// List modelList = new ArrayList<>(); +// list.forEach(item->{ +// Integer did = null; +// for (DevCfgDto item2 : list2) { +// if (Objects.equals(item.getDevType(),item2.getDevType())){ +// did = item2.getDid(); +// } +// } +// CsModelDto csModelDto = new CsModelDto(); +// CsDevModelPO po = devModelFeignClient.findModel(item.getDevType(),item.getVersionNo(),item.getVersionDate()).getData(); +// if (Objects.isNull(po)){ +// //log.error(AccessResponseEnum.MODEL_NO_FIND.getMessage()); +// logDto.setOperate(nDid + "查询系统中是否存在模板"); +// logDto.setResult(0); +// logDto.setFailReason(AccessResponseEnum.MODEL_NO_FIND.getMessage()); +// logMessageTemplate.sendMember(logDto); +// //有异常删除缓存的模板信息 +// redisUtil.delete(AppRedisKey.MODEL + nDid); +// throw new BusinessException(AccessResponseEnum.MODEL_NO_FIND); +// } +// if (Objects.equals(po.getType(),0)){ +// List dataSetList = dataSetFeignClient.getModuleDataSet(po.getId()).getData(); +// if (CollectionUtils.isEmpty(dataSetList)){ +// logDto.setOperate("查看APF模块个数"); +// logDto.setResult(0); +// logDto.setFailReason(AccessResponseEnum.MODULE_NUMBER_IS_NULL.getMessage()); +// logMessageTemplate.sendMember(logDto); +// //有异常删除缓存的模板信息 +// redisUtil.delete(AppRedisKey.MODEL + nDid); +// throw new BusinessException(AccessResponseEnum.MODULE_NUMBER_IS_NULL); +// } +// csModelDto.setModuleNumber(dataSetList.size()); +// } +// csModelDto.setDevType(po.getDevTypeName()); +// csModelDto.setModelId(po.getId()); +// csModelDto.setDid(did); +// csModelDto.setType(po.getType()); +// modelList.add(csModelDto); +// }); +// //存储模板id +// String key2 = AppRedisKey.MODEL + nDid; +// redisUtil.saveByKey(key2,modelList); +// //存储监测点模板信息，用于界面回显 +// List modelId = modelList.stream().map(CsModelDto::getModelId).collect(Collectors.toList()); +// List lineList = csLineModelService.getMonitorNumByModelId(modelId); +// String key = AppRedisKey.LINE + nDid; +// redisUtil.saveByKeyWithExpire(key,lineList,600L); +// logMessageTemplate.sendMember(logDto); +// break; +// case 4613: +// logDto.setOperate("系统端收到装置端"+nDid+"接入应答，code = " + res.getCode()); +// logDto.setResult(1); +// logMessageTemplate.sendMember(logDto); +// //log.info("{},收到接入应答响应,应答code {}",nDid,res.getCode()); +// if (Objects.equals(res.getCode(),AccessEnum.SUCCESS.getCode())){ +// int mid = 1; +// //修改装置状态 +// csEquipmentDeliveryService.updateStatusBynDid(nDid,AccessEnum.ACCESS.getCode(),null,null); +// csEquipmentDeliveryService.updateRunStatusBynDid(nDid,AccessEnum.ONLINE.getCode()); +// //记录设备上线 +// PqsCommunicateDto dto = new PqsCommunicateDto(); +// dto.setTime(LocalDateTime.now().format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_PATTERN))); +// dto.setDevId(nDid); +// dto.setType(1); +// dto.setDescription("通讯正常"); +// csCommunicateFeignClient.insertion(dto); +// //询问设备软件信息 +// askDevData(nDid,version,1,mid); +// //更新治理监测点信息和设备容量 +// askDevData(nDid,version,2,(res.getMid()+1)); +// //更新电网侧、负载侧监测点信息 +// askDevData(nDid,version,3,(res.getMid()+1)); +// //接入后系统重置装置心跳 +// heartbeatService.receiveHeartbeat(nDid); +// //修改redis的mid +// redisUtil.saveByKey(AppRedisKey.DEVICE_MID + nDid,1); +// //接入成功标识 +// redisUtil.saveByKeyWithExpire("online" + nDid,"online",10L); +// //录波任务倒计时 +// redisUtil.saveByKeyWithExpire("startFile:" + nDid,null,60L); +// } else { +// //log.info(AccessResponseEnum.ACCESS_RESPONSE_ERROR.getMessage()); +// logDto.setResult(0); +// logDto.setFailReason(AccessResponseEnum.ACCESS_RESPONSE_ERROR.getMessage()); +// logMessageTemplate.sendMember(logDto); +// throw new BusinessException(AccessResponseEnum.ACCESS_RESPONSE_ERROR); +// } +// logMessageTemplate.sendMember(logDto); +// break; +// case 4614: +// RspDataDto rspDataDto = JSON.parseObject(JSON.toJSONString(res.getMsg()), RspDataDto.class); +// if (!Objects.isNull(rspDataDto.getDataType())) { +// switch (rspDataDto.getDataType()){ +// case 1: +// logDto.setOperate("系统端收到装置端"+nDid+"更新设备软件信息报文，code = " + res.getCode()); +// logDto.setResult(1); +// RspDataDto.SoftInfo softInfo = JSON.parseObject(JSON.toJSONString(rspDataDto.getDataArray()), RspDataDto.SoftInfo.class); +// //记录设备软件信息 +// CsSoftInfoPO csSoftInfoPo = new CsSoftInfoPO(); +// BeanUtils.copyProperties(softInfo,csSoftInfoPo); +// String id = IdUtil.fastSimpleUUID(); +// csSoftInfoPo.setId(id); +// DateTimeFormatter formatter = new DateTimeFormatterBuilder() +// // 优先尝试紧凑格式 +// .optionalStart().appendPattern("yyyyMMdd").optionalEnd() +// // 再尝试带横线格式 +// .optionalStart().appendPattern("yyyy-MM-dd").optionalEnd() +// // 默认时间部分 +// .parseDefaulting(ChronoField.HOUR_OF_DAY, 0) +// .parseDefaulting(ChronoField.MINUTE_OF_HOUR, 0) +// .parseDefaulting(ChronoField.SECOND_OF_MINUTE, 0) +// .toFormatter(); +// LocalDateTime localDateTime = LocalDateTime.parse(softInfo.getAppDate(), formatter); +// assertThat(localDateTime).isNotNull(); +// csSoftInfoPo.setAppDate(localDateTime); +// csSoftInfoFeignClient.saveSoftInfo(csSoftInfoPo); +// //更新设备软件id 先看是否存在软件信息,删除 然后在录入 +// CsEquipmentDeliveryPO po = equipmentFeignClient.findDevByNDid(nDid).getData(); +// String soft = po.getSoftinfoId(); +// if (StringUtil.isNotBlank(soft)){ +// csSoftInfoFeignClient.removeSoftInfo(soft); +// } +// equipmentFeignClient.updateSoftInfo(nDid,csSoftInfoPo.getId()); +// logMessageTemplate.sendMember(logDto); +// break; +// case 2: +// List devInfo = JSON.parseArray(JSON.toJSONString(rspDataDto.getDataArray()), RspDataDto.LdevInfo.class); +// if (CollectionUtil.isNotEmpty(devInfo)){ +// if (Objects.equals(res.getDid(),1)){ +// redisUtil.saveByKeyWithExpire("lineInfo:"+nDid,devInfo,30L); +// List list3 = new ArrayList<>(); +// boolean hasZeroClDid = devInfo.stream().anyMatch(item -> item.getClDid() == 0); +// //治理设备 +// if (hasZeroClDid) { +// logDto.setOperate("系统端收到装置端"+nDid+"更新APF容量报文，code = " + res.getCode()); +// logDto.setResult(1); +// devInfo.forEach(item->{ +// if (Objects.equals(item.getClDid(),0)){ +// updateLineInfo(nDid,item); +// } +// //2.录入各个模块设备容量 +// CsDeviceRegistry csDeviceRegistry = csDeviceRegistryFeignClient.queryByCurrentNdidAndClDid(nDid, 0).getData(); +// String lineId; +// if (Objects.isNull(csDeviceRegistry)) { +// List lines = csLineFeignClient.findByNdid(nDid).getData(); +// if (CollectionUtil.isEmpty(lines)) { +// throw new BusinessException("通过装置NDID获取监测点信息失败"); +// } +// CsLinePO line = lines.stream().filter(item2->Objects.equals(item2.getClDid(),0)).findFirst().orElse(null); +// if (Objects.isNull(line)) { +// throw new BusinessException("通过逻辑子设备ID获取监测点信息失败"); +// } +// lineId = line.getLineId(); +// } else { +// lineId = csDeviceRegistry.getId(); +// } +// CsDevCapacityPO csDevCapacity = new CsDevCapacityPO(); +// csDevCapacity.setLineId(lineId); +// csDevCapacity.setCldid(item.getClDid()); +// csDevCapacity.setCapacity(Objects.isNull(item.getCapacityA())?0.0:item.getCapacityA()); +// list3.add(csDevCapacity); +// }); +// } +// //其余设备 +// else { +// logDto.setOperate("系统端收到装置端"+nDid+"更新监测点台账报文，code = " + res.getCode()); +// logDto.setResult(1); +// devInfo.forEach(item->{ +// updateLineInfo(nDid,item); +// }); +// } +// if (CollectionUtil.isNotEmpty(list3)) { +// devCapacityFeignClient.addList(list3); +// //3.更新设备模块个数 +// equipmentFeignClient.updateModuleNumber(nDid,(devInfo.size()-1)); +// } +// } else if (Objects.equals(res.getDid(),2)) { +// logDto.setOperate("系统端收到装置端"+nDid+"更新电网侧、负载侧监测点信息报文，code = " + res.getCode()); +// logDto.setResult(1); +// //1.更新电网侧、负载侧监测点相关信息 +// devInfo.forEach(item->{ +// updateLineInfo(nDid,item); +// }); +// } +// } +// logMessageTemplate.sendMember(logDto); +// break; +// case 15: +// logDto.setOperate("系统端收到装置端"+nDid+"更新设备软件信息报文，code = " + res.getCode()); +// logDto.setResult(1); +// JSONObject jsonObject = JSONObject.parseObject(JSON.toJSONString(res)); +// AppAutoDataMessage appAutoDataMessage = JSONObject.toJavaObject(jsonObject, AppAutoDataMessage.class); +// appAutoDataMessage.setId(nDid); +// rtFeignClient.apfRtAnalysis(appAutoDataMessage); +// logMessageTemplate.sendMember(logDto); +// break; +// case 48: +// logDto.setOperate("系统端收到装置端"+nDid+"询问项目列表报文，code = " + res.getCode()); +// logDto.setResult(1); +// List projectInfoList = JSON.parseArray(JSON.toJSONString(rspDataDto.getDataArray()), RspDataDto.ProjectInfo.class); +// CsDeviceRegistry csDeviceRegistry = csDeviceRegistryFeignClient.queryByCurrentNdidAndClDid(nDid, rspDataDto.getClDid()).getData(); +// String lineId; +// if (Objects.isNull(csDeviceRegistry)) { +// List lines = csLineFeignClient.findByNdid(nDid).getData(); +// if (CollectionUtil.isEmpty(lines)) { +// throw new BusinessException("通过装置NDID获取监测点信息失败"); +// } +// CsLinePO line = lines.stream().filter(item->Objects.equals(item.getClDid(),rspDataDto.getClDid())).findFirst().orElse(null); +// if (Objects.isNull(line)) { +// throw new BusinessException("通过逻辑子设备ID获取监测点信息失败"); +// } +// lineId = line.getLineId(); +// } else { +// lineId = csDeviceRegistry.getId(); +// } +// String key3 = AppRedisKey.PROJECT_INFO + lineId; +// redisUtil.saveByKeyWithExpire(key3,projectInfoList,60L); +// logMessageTemplate.sendMember(logDto); +// break; +// default: +// break; +// } +// } +// break; +// case 4663: +// if (Objects.equals(res.getCode(),AccessEnum.SUCCESS.getCode())){ +// String key4 = AppRedisKey.CONTROL + nDid; +// redisUtil.saveByKeyWithExpire(key4,"success",10L); +// } +// break; +// default: +// break; +// } +// } else if (Objects.equals(res.getCode(),AccessEnum.START_CHANNEL.getCode())) { +// logDto.setOperate(AccessEnum.START_CHANNEL.getMessage() + "，系统等待5s"); +// logDto.setResult(1); +// logMessageTemplate.sendMember(logDto); +// Thread.sleep(5000); +// } else { +// String result = getEnum(res.getCode()); +// //log.info(result); +// logDto.setOperate("装置响应"); +// logDto.setResult(0); +// logDto.setFailReason(result); +// logMessageTemplate.sendMember(logDto); +// throw new BusinessException(result); +// } +// } + public void updateLineInfo(String nDid,RspDataDto.LdevInfo item) { CsDeviceRegistry csDeviceRegistry = csDeviceRegistryFeignClient.queryByCurrentNdidAndClDid(nDid, item.getClDid()).getData(); String lineId; @@ -534,6 +890,117 @@ public class MqttMessageHandler { overLimitWlMapper.insert(overlimit); } + @MqttSubscribe(value = "/Dev/Data/{version}/{edgeId}",qos = 1) + public void devHeartBeat(String topic, MqttMessage message, @NamedValue("version") String version, @NamedValue("edgeId") String nDid, @Payload String payload) { + Gson gson = new Gson(); + String payloadStr = new String(message.getPayload(), StandardCharsets.UTF_8); + ReqAndResDto.Req res = gson.fromJson(payloadStr, ReqAndResDto.Req.class); + switch (res.getType()){ + case 4865: + // ① 心跳记录 — 必须同步执行，绝不能被后续重活阻塞 + heartbeatService.receiveHeartbeat(nDid); + // ② 心跳响应+业务逻辑 — 异步处理，立即释放Paho回调线程 + mqttMessageExecutor.submit(() -> { + try { + handleHeartbeatBusiness(nDid, version, res, gson); + } catch (Exception e) { + log.error("心跳业务处理异常, nDid={}", nDid, e); + } + }); + break; + case 4866: + AutoDataDto dataDto = gson.fromJson(payloadStr, AutoDataDto.class); + // ① 设备应答 — 同步发送，设备在等待响应 + if (dataDto.getMid() > 0){ + ReqAndResDto.Res response = new ReqAndResDto.Res(); + response.setMid(dataDto.getMid()); + response.setDid(dataDto.getDid()); + response.setPri(AccessEnum.FIRST_CHANNEL.getCode()); + response.setType(Integer.parseInt(TypeEnum.TYPE_15.getCode())); + response.setCode(200); + publisher.send("/Dev/DataRsp/"+version+"/"+nDid,gson.toJson(response),1,false); + } + // ② 数据处理(Kafka发送/Feign调用) — 异步处理，不阻塞Paho回调线程 + mqttMessageExecutor.submit(() -> { + try { + handleDataBusiness(nDid, dataDto, payloadStr); + } catch (Exception e) { + log.error("数据处理异常, nDid={}", nDid, e); + } + }); + break; + default: + break; + } + } + + /** + * 心跳业务处理（异步）：Feign查询设备状态 + MQTT心跳响应 + 异常设备处理 + */ + private void handleHeartbeatBusiness(String nDid, String version, ReqAndResDto.Req res, Gson gson) { + 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 abnormalList = new ArrayList<>(); + if (object instanceof ArrayList){ + abnormalList.addAll((List) object); + } + //todo APF设备不存在逻辑设备掉线的情况，网关下的设备会存在 + abnormalList.forEach(item->{ + log.warn("异常设备ID：{}", item); + }); + } + } + } + } + + /** + * 主动数据处理（异步）：暂态事件、实时数据、统计数据/电度数据 + */ + private void handleDataBusiness(String nDid, AutoDataDto dataDto, String rawPayload) { + Gson gson = new Gson(); + switch (dataDto.getMsg().getDataAttr()) { + //暂态事件、录波处理、工程信息 + case 0: + EventDto eventDto = gson.fromJson(rawPayload, EventDto.class); + JSONObject jsonObject0 = JSONObject.parseObject(JSON.toJSONString(eventDto)); + AppEventMessage appEventMessage = JSONObject.toJavaObject(jsonObject0, AppEventMessage.class); + appEventMessage.setId(nDid); + appEventMessageTemplate.sendMember(appEventMessage); + break; + //实时数据 + case 1: + JSONObject jsonObject2 = JSONObject.parseObject(JSON.toJSONString(dataDto)); + AppAutoDataMessage appAutoDataMessage = JSONObject.toJavaObject(jsonObject2, AppAutoDataMessage.class); + appAutoDataMessage.setId(nDid); + rtFeignClient.analysis(appAutoDataMessage); + break; + //处理主动上送的统计数据、电度数据 + case 2: + case 3: + JSONObject jsonObject3 = JSONObject.parseObject(JSON.toJSONString(dataDto)); + AppAutoDataMessage appAutoDataMessage2 = JSONObject.toJavaObject(jsonObject3, AppAutoDataMessage.class); + appAutoDataMessage2.setId(nDid); + appAutoDataMessageTemplate.sendMember(appAutoDataMessage2); + break; + } + } + + /** * 装置心跳 && 主动数据上送 * fixme 这边由于接收文件数据时间跨度会很长，途中有其他请求进来会中断之前的程序，目前是记录中断的位置，等处理完成再继续请求接收文件 @@ -543,110 +1010,109 @@ public class MqttMessageHandler { * @param nDid * @param payload */ - @MqttSubscribe(value = "/Dev/Data/{version}/{edgeId}",qos = 1) - @Transactional(rollbackFor = Exception.class) - public void devHeartBeat(String topic, MqttMessage message, @NamedValue("version") String version, @NamedValue("edgeId") String nDid, @Payload String payload) { - //解析数据 +// @MqttSubscribe(value = "/Dev/Data/{version}/{edgeId}",qos = 1) +// public void devHeartBeat(String topic, MqttMessage message, @NamedValue("version") String version, @NamedValue("edgeId") String nDid, @Payload String payload) { +// //解析数据 +// Gson gson = new Gson(); +// ReqAndResDto.Req res = gson.fromJson(new String(message.getPayload(), StandardCharsets.UTF_8), ReqAndResDto.Req.class); +// //响应请求 +// switch (res.getType()){ +// case 4865: +// heartbeatService.receiveHeartbeat(nDid); +// //有心跳，则将装置改成在线 +// //csEquipmentDeliveryService.updateRunStatusBynDid(nDid,AccessEnum.ONLINE.getCode()); +// //处理心跳 判断设备是否接入，如果设备已经接入则响应，不然忽略 +// 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 abnormalList = new ArrayList<>(); +// if (object instanceof ArrayList){ +// abnormalList.addAll((List) object); +// } +// //todo APF设备不存在逻辑设备掉线的情况，网关下的设备会存在 +// abnormalList.forEach(item->{ +// System.out.println("异常设备ID："+item); +// }); +// } +// } +// } +// break; +// case 4866: +// AutoDataDto dataDto = gson.fromJson(new String(message.getPayload(), StandardCharsets.UTF_8), AutoDataDto.class); +// //mid大于0，则需要应答设备侧 +// if (dataDto.getMid() > 0){ +// ReqAndResDto.Res response = new ReqAndResDto.Res(); +// response.setMid(dataDto.getMid()); +// response.setDid(dataDto.getDid()); +// response.setPri(AccessEnum.FIRST_CHANNEL.getCode()); +// response.setType(Integer.parseInt(TypeEnum.TYPE_15.getCode())); +// response.setCode(200); +// publisher.send("/Dev/DataRsp/"+version+"/"+nDid,new Gson().toJson(response),1,false); +// } +// //判断事件类型 +// switch (dataDto.getMsg().getDataAttr()) { +// //暂态事件、录波处理、工程信息 +// case 0: +// 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); +// appEventMessage.setId(nDid); +// appEventMessageTemplate.sendMember(appEventMessage); +// break; +// //实时数据 +// case 1: +// JSONObject jsonObject2 = JSONObject.parseObject(JSON.toJSONString(dataDto)); +// AppAutoDataMessage appAutoDataMessage = JSONObject.toJavaObject(jsonObject2, AppAutoDataMessage.class); +// appAutoDataMessage.setId(nDid); +// rtFeignClient.analysis(appAutoDataMessage); +// break; +// //处理主动上送的统计数据、电度数据 +// case 2: +// case 3: +// JSONObject jsonObject3 = JSONObject.parseObject(JSON.toJSONString(dataDto)); +// AppAutoDataMessage appAutoDataMessage2 = JSONObject.toJavaObject(jsonObject3, AppAutoDataMessage.class); +// appAutoDataMessage2.setId(nDid); +// appAutoDataMessageTemplate.sendMember(appAutoDataMessage2); +// break; +// } +// break; +// default: +// break; +// } +// } + + + @MqttSubscribe(value = "/Pfm/DevFileRsp/{version}/{edgeId}",qos = 1) + public void file(String topic, MqttMessage message, @NamedValue("version") String version, @NamedValue("edgeId") String nDid, @Payload String payload) { Gson gson = new Gson(); - ReqAndResDto.Req res = gson.fromJson(new String(message.getPayload(), StandardCharsets.UTF_8), ReqAndResDto.Req.class); - //响应请求 - switch (res.getType()){ - case 4865: - heartbeatService.receiveHeartbeat(nDid); - //有心跳，则将装置改成在线 - //csEquipmentDeliveryService.updateRunStatusBynDid(nDid,AccessEnum.ONLINE.getCode()); - //处理心跳 判断设备是否接入，如果设备已经接入则响应，不然忽略 - 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 abnormalList = new ArrayList<>(); - if (object instanceof ArrayList){ - abnormalList.addAll((List) object); - } - //todo APF设备不存在逻辑设备掉线的情况，网关下的设备会存在 - abnormalList.forEach(item->{ - System.out.println("异常设备ID："+item); - }); - } - } - } - break; - case 4866: - AutoDataDto dataDto = gson.fromJson(new String(message.getPayload(), StandardCharsets.UTF_8), AutoDataDto.class); - //mid大于0，则需要应答设备侧 - if (dataDto.getMid() > 0){ - ReqAndResDto.Res response = new ReqAndResDto.Res(); - response.setMid(dataDto.getMid()); - response.setDid(dataDto.getDid()); - response.setPri(AccessEnum.FIRST_CHANNEL.getCode()); - response.setType(Integer.parseInt(TypeEnum.TYPE_15.getCode())); - response.setCode(200); - publisher.send("/Dev/DataRsp/"+version+"/"+nDid,new Gson().toJson(response),1,false); - } - //判断事件类型 - switch (dataDto.getMsg().getDataAttr()) { - //暂态事件、录波处理、工程信息 - case 0: - 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); - appEventMessage.setId(nDid); - appEventMessageTemplate.sendMember(appEventMessage); - break; - //实时数据 - case 1: - JSONObject jsonObject2 = JSONObject.parseObject(JSON.toJSONString(dataDto)); - AppAutoDataMessage appAutoDataMessage = JSONObject.toJavaObject(jsonObject2, AppAutoDataMessage.class); - appAutoDataMessage.setId(nDid); - rtFeignClient.analysis(appAutoDataMessage); - break; - //处理主动上送的统计数据、电度数据 - case 2: - case 3: - JSONObject jsonObject3 = JSONObject.parseObject(JSON.toJSONString(dataDto)); - AppAutoDataMessage appAutoDataMessage2 = JSONObject.toJavaObject(jsonObject3, AppAutoDataMessage.class); - appAutoDataMessage2.setId(nDid); - appAutoDataMessageTemplate.sendMember(appAutoDataMessage2); - break; - } - break; - default: - break; - } + String payloadStr = new String(message.getPayload(), StandardCharsets.UTF_8); + mqttMessageExecutor.submit(() -> { + try { + doFile(gson.fromJson(payloadStr, FileDto.class), version, nDid); + } catch (Exception e) { + log.error("file业务处理异常, nDid={}", nDid, e); + } + }); } - /** - * 文件传输 - * @param topic - * @param message - * @param version - * @param nDid - * @param payload - */ - @MqttSubscribe(value = "/Pfm/DevFileRsp/{version}/{edgeId}",qos = 1) - @Transactional(rollbackFor = Exception.class) - public void file(String topic, MqttMessage message, @NamedValue("version") String version, @NamedValue("edgeId") String nDid, @Payload String payload) { - //解析数据 - Gson gson = new Gson(); - FileDto fileDto = gson.fromJson(new String(message.getPayload(), StandardCharsets.UTF_8), FileDto.class); + private void doFile(FileDto fileDto, String version, String nDid) { JSONObject jsonObject = JSONObject.parseObject(JSON.toJSONString(fileDto)); AppFileMessage appFileMessage = JSONObject.toJavaObject(jsonObject, AppFileMessage.class); appFileMessage.setId(nDid); - //响应请求 switch (fileDto.getType()){ case 4657: if (Objects.equals(fileDto.getCode(),AccessEnum.SUCCESS.getCode())) { @@ -687,7 +1153,6 @@ public class MqttMessageHandler { if (Objects.equals(fileDto.getCode(),AccessEnum.SUCCESS.getCode())){ appFileStreamMessageTemplate.sendMember(appFileMessage); } - //todo 处理文件信息，先缓存起来，后期在询问 else if (Objects.equals(fileDto.getCode(),AccessEnum.REFUSE_WAIT.getCode())) { log.info("需要缓存请求的文件信息"); } @@ -715,25 +1180,126 @@ public class MqttMessageHandler { } } - /** - * 装置异常事件记录 - * @param topic - * @param message - * @param version - * @param nDid - * @param payload - */ +// /** +// * 文件传输 +// * @param topic +// * @param message +// * @param version +// * @param nDid +// * @param payload +// */ +// @MqttSubscribe(value = "/Pfm/DevFileRsp/{version}/{edgeId}",qos = 1) +// @Transactional(rollbackFor = Exception.class) +// public void file(String topic, MqttMessage message, @NamedValue("version") String version, @NamedValue("edgeId") String nDid, @Payload String payload) { +// //解析数据 +// Gson gson = new Gson(); +// FileDto fileDto = gson.fromJson(new String(message.getPayload(), StandardCharsets.UTF_8), FileDto.class); +// JSONObject jsonObject = JSONObject.parseObject(JSON.toJSONString(fileDto)); +// AppFileMessage appFileMessage = JSONObject.toJavaObject(jsonObject, AppFileMessage.class); +// appFileMessage.setId(nDid); +// //响应请求 +// switch (fileDto.getType()){ +// case 4657: +// if (Objects.equals(fileDto.getCode(),AccessEnum.SUCCESS.getCode())) { +// String key = AppRedisKey.PROJECT_INFO + nDid; +// if (Objects.isNull(fileDto.getMsg().getType())) { +// handleDefaultCase(fileDto, nDid); +// } else { +// if (Objects.equals("dir", fileDto.getMsg().getType())) { +// saveDirectoryInfo(fileDto.getMsg().getDirInfo(), key); +// } else if (Objects.equals("file", fileDto.getMsg().getType())){ +// saveFileInfo(fileDto.getMsg().getFileInfo(), key); +// appFileMessageTemplate.sendMember(appFileMessage); +// } +// } +// } else if (Objects.equals(fileDto.getCode(),AccessEnum.NOT_FIND.getCode())) { +// Object object = redisUtil.getObjectByKey("fileMid:" + nDid); +// if (Objects.nonNull(object)) { +// String data = redisUtil.getObjectByKey("fileMid:" + nDid).toString(); +// String [] arr = data.split("concat"); +// Integer mid = Integer.parseInt(arr[0]); +// String fileName = arr[1]; +// if (Objects.equals(mid,fileDto.getMid())) { +// List list = channelObjectUtil.objectToList( redisUtil.getObjectByKey("eventFile:" + nDid),WaveTimeDto.class); +// list.removeIf(item -> item.getFileName().equals(fileName)); +// redisUtil.saveByKey("eventFile:" + nDid, list); +// if (CollectionUtil.isNotEmpty(list)) { +// redisUtil.delete("handleEvent:" + nDid); +// waveFeignClient.channelWave(nDid); +// } +// } +// } +// } +// break; +// case 4658: +// FileRedisDto dto = new FileRedisDto(); +// dto.setCode(fileDto.getCode()); +// redisUtil.saveByKeyWithExpire(AppRedisKey.DOWNLOAD + fileDto.getMsg().getName() + fileDto.getMid(),dto,60L); +// if (Objects.equals(fileDto.getCode(),AccessEnum.SUCCESS.getCode())){ +// appFileStreamMessageTemplate.sendMember(appFileMessage); +// } +// //todo 处理文件信息，先缓存起来，后期在询问 +// else if (Objects.equals(fileDto.getCode(),AccessEnum.REFUSE_WAIT.getCode())) { +// log.info("需要缓存请求的文件信息"); +// } +// break; +// case 4659: +// log.info("装置收到系统上传的文件"); +// FileRedisDto fileRedisDto = new FileRedisDto(); +// fileRedisDto.setCode(fileDto.getCode()); +// redisUtil.saveByKeyWithExpire(AppRedisKey.UPLOAD.concat(nDid).concat(String.valueOf(fileDto.getMid())),fileRedisDto,10L); +// redisUtil.saveByKeyWithExpire("uploading","uploading",20L); +// break; +// case 4660: +// log.info("设备目录/文件删除应答"); +// redisUtil.saveByKeyWithExpire( "deleteDir"+ nDid,fileDto.getCode(),10L); +// break; +// case 4661: +// log.info("设备目录创建应答"); +// redisUtil.saveByKeyWithExpire( "createDir"+ nDid,fileDto.getCode(),10L); +// break; +// case 4662: +// log.info("装置根目录应答"); +// redisUtil.saveByKeyWithExpire(AppRedisKey.DEVICE_ROOT_PATH + nDid,fileDto.getMsg().getName(),10L); +// default: +// break; +// } +// } + @MqttSubscribe(value = "/Dev/Error/{edgeId}",qos = 1) - @Transactional(rollbackFor = Exception.class) public void devErrorInfo(String topic, MqttMessage message, @NamedValue("version") String version, @NamedValue("edgeId") String nDid, @Payload String payload) { - //解析数据 Gson gson = new Gson(); - 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); - appEventMessage.setId(nDid); - appEventMessageTemplate.sendMember(appEventMessage); + String payloadStr = new String(message.getPayload(), StandardCharsets.UTF_8); + mqttMessageExecutor.submit(() -> { + try { + EventDto eventDto = gson.fromJson(payloadStr, EventDto.class); + JSONObject jsonObject0 = JSONObject.parseObject(JSON.toJSONString(eventDto)); + AppEventMessage appEventMessage = JSONObject.toJavaObject(jsonObject0, AppEventMessage.class); + appEventMessage.setId(nDid); + appEventMessageTemplate.sendMember(appEventMessage); + } catch (Exception e) { + log.error("devErrorInfo业务处理异常, nDid={}", nDid, e); + } + }); } +// /** +// * 装置异常事件记录 +// * @param topic +// * @param message +// * @param version +// * @param nDid +// * @param payload +// */ +// @MqttSubscribe(value = "/Dev/Error/{edgeId}",qos = 1) +// public void devErrorInfo(String topic, MqttMessage message, @NamedValue("version") String version, @NamedValue("edgeId") String nDid, @Payload String payload) { +// //解析数据 +// Gson gson = new Gson(); +// 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); +// appEventMessage.setId(nDid); +// appEventMessageTemplate.sendMember(appEventMessage); +// } private void saveDirectoryInfo(List dirInfo, String key) { if (!CollectionUtil.isEmpty(dirInfo)) { diff --git 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 index 5f21baa..0b36222 100644 --- 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(); } diff --git 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 index 82b550a..e25735d 100644 --- 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); } diff --git 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 index 172c61f..dfc44fb 100644 --- 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); }); } diff --git 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 index 2d118c9..6134b50 100644 --- 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 @@ -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); }