From b32219c1c102c844fa0fb2bc88c81423fe064c4c Mon Sep 17 00:00:00 2001 From: xy <748613696@qq.com> Date: Thu, 2 Jul 2026 19:07:36 +0800 Subject: [PATCH] =?UTF-8?q?refactor(access):=20=E9=87=8D=E6=9E=84=E8=AE=BE?= =?UTF-8?q?=E5=A4=87=E6=8E=A5=E5=85=A5=E4=BB=BB=E5=8A=A1=E5=92=8C=E4=BC=98?= =?UTF-8?q?=E5=8C=96=E7=BA=BF=E7=A8=8B=E7=AE=A1=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 移除废弃的日志消息相关依赖和代码 - 优化 AccessApplicationRunner 和 AutoAccessTimer 的线程池使用 - 采用共享线程池和信号量控制并发访问设备 - 批量处理设备接入任务并添加批次间隔控制 - 重构 autoAccess2 方法移除事务注解并改用异步处理 - 修复更新线路ID时参数传递错误的问题 - 统一错误处理和异常捕获机制 - 添加更详细的日志记录和错误追踪功能 --- .../config/MqttAccessSchedulerConfig.java | 52 +++ .../access/controller/CsHeartController.java | 22 +- .../runner/AccessApplicationRunner.java | 293 ++++++++++----- .../njcn/access/runner/AutoAccessTimer.java | 345 ++++++++++++------ .../access/runner/UpdateDevStatusTimer.java | 26 ++ .../service/ICsEquipmentDeliveryService.java | 5 + .../njcn/access/service/ICsHeartService.java | 6 + .../service/impl/CsDeviceServiceImpl.java | 246 ++++++++++--- .../impl/CsEquipmentDeliveryServiceImpl.java | 10 + .../service/impl/CsHeartServiceImpl.java | 47 +++ .../stat/service/impl/StatServiceImpl.java | 20 +- .../service/impl/EventServiceImpl.java | 10 +- 12 files changed, 816 insertions(+), 266 deletions(-) create mode 100644 iot-access/access-boot/src/main/java/com/njcn/access/config/MqttAccessSchedulerConfig.java create mode 100644 iot-access/access-boot/src/main/java/com/njcn/access/runner/UpdateDevStatusTimer.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 new file mode 100644 index 0000000..48328a7 --- /dev/null +++ b/iot-access/access-boot/src/main/java/com/njcn/access/config/MqttAccessSchedulerConfig.java @@ -0,0 +1,52 @@ +package com.njcn.access.config; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +import java.util.concurrent.*; + +/** + * MQTT接入任务共享资源池配置 + */ +@Configuration +@Slf4j +public class MqttAccessSchedulerConfig { + + /** + * 全局MQTT发送并发上限 + * 建议设为 MqttConnectOptions.maxInflight 的 50%-70% + */ + @Value("${mqtt.access.concurrency-limit:200}") + private int concurrencyLimit; + + @Value("${mqtt.access.worker-threads:5}") + private int workerThreads; + + @Bean(destroyMethod = "shutdownNow") + public ScheduledExecutorService sharedScheduler() { + return Executors.newSingleThreadScheduledExecutor(r -> { + Thread t = new Thread(r, "mqtt-access-scheduler"); + t.setDaemon(true); + return t; + }); + } + + @Bean(destroyMethod = "shutdownNow") + public ExecutorService sharedWorkerPool() { + return new ThreadPoolExecutor( + workerThreads, workerThreads, + 0L, TimeUnit.MILLISECONDS, + new LinkedBlockingQueue<>(200), + r -> new Thread(r, "mqtt-access-worker"), + new ThreadPoolExecutor.CallerRunsPolicy() // 队列满时调用者线程执行，天然背压 + ); + } + + @Bean + public Semaphore mqttSemaphore() { + log.info("初始化MQTT接入信号量, 并发上限={}", concurrencyLimit); + return new Semaphore(concurrencyLimit); + } +} \ No newline at end of file diff --git a/iot-access/access-boot/src/main/java/com/njcn/access/controller/CsHeartController.java b/iot-access/access-boot/src/main/java/com/njcn/access/controller/CsHeartController.java index e3ab18c..be250f8 100644 --- a/iot-access/access-boot/src/main/java/com/njcn/access/controller/CsHeartController.java +++ b/iot-access/access-boot/src/main/java/com/njcn/access/controller/CsHeartController.java @@ -17,7 +17,8 @@ import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; -import springfox.documentation.annotations.ApiIgnore; + +import java.util.List; /** * 类的介绍： @@ -31,7 +32,6 @@ import springfox.documentation.annotations.ApiIgnore; @RequestMapping("/heartbeat") @Api(tags = "心跳") @AllArgsConstructor -@ApiIgnore public class CsHeartController extends BaseController { private final ICsHeartService csHeartService; @@ -45,4 +45,22 @@ public class CsHeartController extends BaseController { csHeartService.handleHeartbeat(message); return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, null, methodDescribe); } + + @OperateInfo(info = LogEnum.BUSINESS_COMMON) + @PostMapping("/getMissHeartbeat") + @ApiOperation("获取物联心跳丢失状态在线的设备") + public HttpResult> getMissHeartbeat(){ + String methodDescribe = getMethodDescribe("getMissHeartbeat"); + List result = csHeartService.getMissHeartbeat(); + return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, result, methodDescribe); + } + + @OperateInfo(info = LogEnum.BUSINESS_COMMON) + @PostMapping("/updateDevStatus") + @ApiOperation("手动调整状态") + public HttpResult updateDevStatus(){ + String methodDescribe = getMethodDescribe("updateDevStatus"); + csHeartService.updateDevStatus(); + return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, null, methodDescribe); + } } diff --git a/iot-access/access-boot/src/main/java/com/njcn/access/runner/AccessApplicationRunner.java b/iot-access/access-boot/src/main/java/com/njcn/access/runner/AccessApplicationRunner.java index 938b0c8..95ceff7 100644 --- a/iot-access/access-boot/src/main/java/com/njcn/access/runner/AccessApplicationRunner.java +++ b/iot-access/access-boot/src/main/java/com/njcn/access/runner/AccessApplicationRunner.java @@ -6,8 +6,6 @@ import com.njcn.access.service.ICsTopicService; import com.njcn.access.service.impl.CsDeviceServiceImpl; import com.njcn.access.utils.ChannelObjectUtil; import com.njcn.csdevice.pojo.po.CsEquipmentDeliveryPO; -import com.njcn.mq.message.LogMessage; -import com.njcn.mq.template.LogMessageTemplate; import com.njcn.redis.pojo.enums.AppRedisKey; import com.njcn.redis.utils.RedisUtil; import com.njcn.system.enums.DicDataEnum; @@ -18,110 +16,241 @@ import org.springframework.boot.ApplicationArguments; import org.springframework.boot.ApplicationRunner; import org.springframework.stereotype.Component; -import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.Objects; -import java.util.concurrent.*; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; -/** - * 类的介绍：用来重新发起设备的接入，存在程序意外停止了，缓存失效导致无法更新装置的状态，所以需要在程序启动时发起设备的接入 - * - * @author xuyang - * @version 1.0.0 - * @createTime 2023/8/28 13:57 - */ @Component @Slf4j @RequiredArgsConstructor public class AccessApplicationRunner implements ApplicationRunner { + private static final long STARTUP_DELAY = 60L; + private static final int BATCH_SIZE = 20; + private static final long BATCH_INTERVAL_MS = 500L; + private final RedisUtil redisUtil; private final ICsEquipmentDeliveryService csEquipmentDeliveryService; private final ICsTopicService csTopicService; private final CsDeviceServiceImpl csDeviceService; private final ChannelObjectUtil channelObjectUtil; - ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1); - private static final long ACCESS_TIME = 60L; + // ✅ 注入公共资源 + private final ScheduledExecutorService sharedScheduler; + private final ExecutorService sharedWorkerPool; + private final Semaphore mqttSemaphore; @Override public void run(ApplicationArguments args) { - //项目启动60s后发起自动接入 - Runnable task = () -> { - log.info("系统重启,所有符合条件的装置发起接入!"); - List list = csEquipmentDeliveryService.getOnlineDev(); - if (CollUtil.isNotEmpty(list)) { - //获取字典数据 - List dictTreeKey = channelObjectUtil.objectToList(redisUtil.getObjectByKey(AppRedisKey.DICT_TREE), SysDicTreePO.class); - Map dictTreeMap = dictTreeKey.stream().collect(Collectors.toMap(SysDicTreePO::getId, item -> item)); - //获取主题版本信息 - List nDidIds = list.stream().map(CsEquipmentDeliveryPO::getNdid).collect(Collectors.toList()); - Map topicVersions = csTopicService.getVersion(nDidIds); - - ExecutorService executor = Executors.newFixedThreadPool(10); - // 将任务平均分配给10个子列表 - List> subLists = new ArrayList<>(); - int partitionSize = list.size() / 10; - for (int i = 0; i < 10; i++) { - int start = i * partitionSize; - int end = (i == 9) ? list.size() : start + partitionSize; - subLists.add(list.subList(start, end)); - } - // 创建一个ExecutorService来处理这些任务 - List> futures = new ArrayList<>(); - // 提交任务给线程池执行 - for (int i = 0; i < 10; i++) { - int index = i; - futures.add(executor.submit(new Callable() { - @Override - public Void call() { - accessDev(subLists.get(index), dictTreeMap, topicVersions); - return null; - } - })); - } - // 等待所有任务完成 - for (Future future : futures) { - try { - future.get(); - } catch (InterruptedException | ExecutionException e) { - throw new RuntimeException(e); - } - } - // 关闭ExecutorService - executor.shutdown(); + sharedScheduler.schedule(() -> { + try { + executeStartupRecovery(); + } catch (Throwable t) { + log.error("启动恢复任务异常", t); } - }; - scheduler.schedule(task,ACCESS_TIME,TimeUnit.SECONDS); - scheduler.shutdown(); + }, STARTUP_DELAY, TimeUnit.SECONDS); + + log.info("启动恢复任务已调度, 延迟={}s", STARTUP_DELAY); } - public void accessDev(List list, Map dictTreeMap, Map topicVersions) { - if (CollUtil.isNotEmpty(list)) { - try { - list.forEach(item->{ - //System.out.println(Thread.currentThread().getName() + ": reboot : nDid : " + item.getNdid()); - //判断设备类型 便携式设备需要特殊处理 未注册的要先注册、再接入;已注册的直接重新接入 - if (Objects.equals(dictTreeMap.get(item.getDevType()).getCode(), DicDataEnum.PORTABLE.getCode()) && Objects.equals(item.getStatus(),1)) { - //csDeviceService.wlDevRegister(item.getNdid()); - log.info("请先手动注册、接入"); - } else { - String version = topicVersions.get(item.getNdid()); - if (Objects.isNull(version)) { - version = "V1"; - } - boolean result = csDeviceService.autoAccess(item.getNdid(),version,1); - if (result) { - redisUtil.saveByKey(AppRedisKey.DEVICE_MID + item.getNdid(), 1); - } - } - }); - } catch (Exception e) { - log.error(e.getMessage()); + private void executeStartupRecovery() { + List list = csEquipmentDeliveryService.getOnlineDev(); + if (CollUtil.isEmpty(list)) { + log.info("启动恢复: 无需恢复的设备"); + return; + } + log.info("启动恢复: 待恢复设备数={}", list.size()); + + List dictTreeKey = channelObjectUtil.objectToList( + redisUtil.getObjectByKey(AppRedisKey.DICT_TREE), SysDicTreePO.class); + Map dictTreeMap = dictTreeKey.stream() + .collect(Collectors.toMap(SysDicTreePO::getId, item -> item, (a, b) -> a)); + + List ndidIds = list.stream().map(CsEquipmentDeliveryPO::getNdid).collect(Collectors.toList()); + Map topicVersions = csTopicService.getVersion(ndidIds); + + List> batches = CollUtil.split(list, BATCH_SIZE); + + int batchCount = 0; + for (List batch : batches) { + sharedWorkerPool.submit(() -> accessDevSafely(batch, dictTreeMap, topicVersions)); + // 非最后一批时等待 + if (++batchCount < batches.size()) { + try { + Thread.sleep(BATCH_INTERVAL_MS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + log.warn("批次间隔被中断，停止后续调度"); + return; + } } } } + private void accessDevSafely(List list, + Map dictTreeMap, + Map topicVersions) { + for (CsEquipmentDeliveryPO item : list) { + try { + SysDicTreePO dicItem = dictTreeMap.get(item.getDevType()); + if (dicItem != null + && Objects.equals(dicItem.getCode(), DicDataEnum.PORTABLE.getCode()) + && Objects.equals(item.getStatus(), 1)) { + log.debug("设备 {} 需手动注册接入，跳过", item.getNdid()); + continue; + } + + String version = topicVersions.getOrDefault(item.getNdid(), "V1"); + + mqttSemaphore.acquire(); + try { + boolean result = csDeviceService.autoAccess(item.getNdid(), version, 1); + if (result) { + redisUtil.saveByKey(AppRedisKey.DEVICE_MID + item.getNdid(), 1); + } + } finally { + mqttSemaphore.release(); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + log.warn("启动恢复: 设备 {} 处理被中断", item.getNdid()); + return; + } catch (Exception e) { + log.error("启动恢复: 设备 {} 接入异常: {}", item.getNdid(), e.getMessage()); + } + } + } } + +//package com.njcn.access.runner; +// +//import cn.hutool.core.collection.CollUtil; +//import com.njcn.access.service.ICsEquipmentDeliveryService; +//import com.njcn.access.service.ICsTopicService; +//import com.njcn.access.service.impl.CsDeviceServiceImpl; +//import com.njcn.access.utils.ChannelObjectUtil; +//import com.njcn.csdevice.pojo.po.CsEquipmentDeliveryPO; +//import com.njcn.mq.message.LogMessage; +//import com.njcn.mq.template.LogMessageTemplate; +//import com.njcn.redis.pojo.enums.AppRedisKey; +//import com.njcn.redis.utils.RedisUtil; +//import com.njcn.system.enums.DicDataEnum; +//import com.njcn.system.pojo.po.SysDicTreePO; +//import lombok.RequiredArgsConstructor; +//import lombok.extern.slf4j.Slf4j; +//import org.springframework.boot.ApplicationArguments; +//import org.springframework.boot.ApplicationRunner; +//import org.springframework.stereotype.Component; +// +//import java.util.ArrayList; +//import java.util.List; +//import java.util.Map; +//import java.util.Objects; +//import java.util.concurrent.*; +//import java.util.stream.Collectors; +// +///** +// * 类的介绍：用来重新发起设备的接入，存在程序意外停止了，缓存失效导致无法更新装置的状态，所以需要在程序启动时发起设备的接入 +// * +// * @author xuyang +// * @version 1.0.0 +// * @createTime 2023/8/28 13:57 +// */ +//@Component +//@Slf4j +//@RequiredArgsConstructor +//public class AccessApplicationRunner implements ApplicationRunner { +// +// private final RedisUtil redisUtil; +// private final ICsEquipmentDeliveryService csEquipmentDeliveryService; +// private final ICsTopicService csTopicService; +// private final CsDeviceServiceImpl csDeviceService; +// private final ChannelObjectUtil channelObjectUtil; +// +// ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1); +// private static final long ACCESS_TIME = 60L; +// +// @Override +// public void run(ApplicationArguments args) { +// //项目启动60s后发起自动接入 +// Runnable task = () -> { +// log.info("系统重启,所有符合条件的装置发起接入!"); +// List list = csEquipmentDeliveryService.getOnlineDev(); +// if (CollUtil.isNotEmpty(list)) { +// //获取字典数据 +// List dictTreeKey = channelObjectUtil.objectToList(redisUtil.getObjectByKey(AppRedisKey.DICT_TREE), SysDicTreePO.class); +// Map dictTreeMap = dictTreeKey.stream().collect(Collectors.toMap(SysDicTreePO::getId, item -> item)); +// //获取主题版本信息 +// List nDidIds = list.stream().map(CsEquipmentDeliveryPO::getNdid).collect(Collectors.toList()); +// Map topicVersions = csTopicService.getVersion(nDidIds); +// +// ExecutorService executor = Executors.newFixedThreadPool(10); +// // 将任务平均分配给10个子列表 +// List> subLists = new ArrayList<>(); +// int partitionSize = list.size() / 10; +// for (int i = 0; i < 10; i++) { +// int start = i * partitionSize; +// int end = (i == 9) ? list.size() : start + partitionSize; +// subLists.add(list.subList(start, end)); +// } +// // 创建一个ExecutorService来处理这些任务 +// List> futures = new ArrayList<>(); +// // 提交任务给线程池执行 +// for (int i = 0; i < 10; i++) { +// int index = i; +// futures.add(executor.submit(new Callable() { +// @Override +// public Void call() { +// accessDev(subLists.get(index), dictTreeMap, topicVersions); +// return null; +// } +// })); +// } +// // 等待所有任务完成 +// for (Future future : futures) { +// try { +// future.get(); +// } catch (InterruptedException | ExecutionException e) { +// throw new RuntimeException(e); +// } +// } +// // 关闭ExecutorService +// executor.shutdown(); +// } +// }; +// scheduler.schedule(task,ACCESS_TIME,TimeUnit.SECONDS); +// scheduler.shutdown(); +// } +// +// public void accessDev(List list, Map dictTreeMap, Map topicVersions) { +// if (CollUtil.isNotEmpty(list)) { +// try { +// list.forEach(item->{ +// //System.out.println(Thread.currentThread().getName() + ": reboot : nDid : " + item.getNdid()); +// //判断设备类型 便携式设备需要特殊处理 未注册的要先注册、再接入;已注册的直接重新接入 +// if (Objects.equals(dictTreeMap.get(item.getDevType()).getCode(), DicDataEnum.PORTABLE.getCode()) && Objects.equals(item.getStatus(),1)) { +// //csDeviceService.wlDevRegister(item.getNdid()); +// log.info("请先手动注册、接入"); +// } else { +// String version = topicVersions.get(item.getNdid()); +// if (Objects.isNull(version)) { +// version = "V1"; +// } +// boolean result = csDeviceService.autoAccess(item.getNdid(),version,1); +// if (result) { +// redisUtil.saveByKey(AppRedisKey.DEVICE_MID + item.getNdid(), 1); +// } +// } +// }); +// } catch (Exception e) { +// log.error(e.getMessage()); +// } +// } +// } +// +//} diff --git a/iot-access/access-boot/src/main/java/com/njcn/access/runner/AutoAccessTimer.java b/iot-access/access-boot/src/main/java/com/njcn/access/runner/AutoAccessTimer.java index 84cbad0..53c9ef2 100644 --- a/iot-access/access-boot/src/main/java/com/njcn/access/runner/AutoAccessTimer.java +++ b/iot-access/access-boot/src/main/java/com/njcn/access/runner/AutoAccessTimer.java @@ -16,151 +16,280 @@ import org.springframework.boot.ApplicationArguments; import org.springframework.boot.ApplicationRunner; import org.springframework.stereotype.Component; -import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.Objects; -import java.util.concurrent.*; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; -/** - * @author xy - * 定时轮询离线设备接入 - */ @Component @Slf4j @RequiredArgsConstructor public class AutoAccessTimer implements ApplicationRunner { - ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1); + private static final long AUTO_TIME = 120L; + private static final int BATCH_SIZE = 20; + private static final long BATCH_INTERVAL_MS = 500L; + private final RedisUtil redisUtil; private final ICsEquipmentDeliveryService csEquipmentDeliveryService; private final ICsTopicService csTopicService; private final CsDeviceServiceImpl csDeviceService; private final ChannelObjectUtil channelObjectUtil; + // ✅ 注入公共资源 + private final ScheduledExecutorService sharedScheduler; + private final ExecutorService sharedWorkerPool; + private final Semaphore mqttSemaphore; + @Override public void run(ApplicationArguments args) { - if (scheduler.isShutdown() || scheduler.isTerminated()) { - scheduler = Executors.newScheduledThreadPool(1); - } - Runnable task = () -> { + sharedScheduler.scheduleWithFixedDelay(() -> { try { executeScheduledTask(); + } catch (Throwable t) { + log.error("定时轮询任务异常", t); } - // 捕获所有Throwable，包括Error - catch (Throwable t) { - log.error("定时任务发生严重异常，尝试恢复", t); - // 可以添加重启逻辑或告警 - } - }; - ScheduledFuture future = scheduler.scheduleWithFixedDelay(task, AUTO_TIME, AUTO_TIME, TimeUnit.SECONDS); - // 添加监控，如果任务被取消则重新调度 - monitorScheduledTask(future); - } + }, AUTO_TIME, AUTO_TIME, TimeUnit.SECONDS); - //10分钟检查一下调度任务 - private void monitorScheduledTask(ScheduledFuture future) { - Thread monitorThread = new Thread(() -> { - while (!Thread.currentThread().isInterrupted()) { - try { - //每10分钟检查一次 - Thread.sleep(600000); - if (future.isCancelled() || future.isDone()) { - log.warn("定时任务被取消或完成，重新调度..."); - // 重新启动任务 - run(null); - break; - } - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - log.warn("监控线程被中断"); - break; - } catch (Exception e) { - log.error("监控任务异常", e); - } - } - }, "Schedule-Monitor-Thread"); - - monitorThread.setDaemon(true); - monitorThread.start(); + log.info("自动接入定时任务已启动, 周期={}s", AUTO_TIME); } private void executeScheduledTask() { - log.info("轮询定时任务执行中!"); List list = csEquipmentDeliveryService.getOfflineDev(); - if (CollUtil.isNotEmpty(list)) { - //获取字典数据 - List dictTreeKey = channelObjectUtil.objectToList(redisUtil.getObjectByKey(AppRedisKey.DICT_TREE), SysDicTreePO.class); - Map dictTreeMap = dictTreeKey.stream().collect(Collectors.toMap(SysDicTreePO::getId, item -> item)); - //获取主题版本信息 - List nDidIds = list.stream().map(CsEquipmentDeliveryPO::getNdid).collect(Collectors.toList()); - Map topicVersions = csTopicService.getVersion(nDidIds); + if (CollUtil.isEmpty(list)) { + return; + } + log.info("轮询定时任务执行中, 待处理设备数={}", list.size()); - ExecutorService executor = Executors.newFixedThreadPool(10); - try { - List> subLists = CollUtil.split(list, 10); - List> futures = new ArrayList<>(); - for (List subList : subLists) { - futures.add(executor.submit(() -> { - try { - accessDevSafely(subList,dictTreeMap,topicVersions); - } catch (Exception e) { - log.error("处理设备子列表异常", e); - } - return null; - })); - } - for (Future future : futures) { - try { - future.get(5, TimeUnit.MINUTES); - } catch (TimeoutException e) { - log.error("任务执行超时", e); - } catch (Exception e) { - log.error("任务执行异常", e); - } - } - } finally { - executor.shutdown(); + List dictTreeKey = channelObjectUtil.objectToList( + redisUtil.getObjectByKey(AppRedisKey.DICT_TREE), SysDicTreePO.class); + Map dictTreeMap = dictTreeKey.stream() + .collect(Collectors.toMap(SysDicTreePO::getId, item -> item, (a, b) -> a)); + + List ndidIds = list.stream().map(CsEquipmentDeliveryPO::getNdid).collect(Collectors.toList()); + Map topicVersions = csTopicService.getVersion(ndidIds); + + List> batches = CollUtil.split(list, BATCH_SIZE); + + int batchCount = 0; + for (List batch : batches) { + sharedWorkerPool.submit(() -> accessDevSafely(batch, dictTreeMap, topicVersions)); + + // 非最后一批时等待 + if (++batchCount < batches.size()) { try { - if (!executor.awaitTermination(1, TimeUnit.MINUTES)) { - executor.shutdownNow(); - } + Thread.sleep(BATCH_INTERVAL_MS); } catch (InterruptedException e) { - executor.shutdownNow(); Thread.currentThread().interrupt(); + log.warn("批次间隔被中断，停止后续调度"); + return; } } } } - //安全的accessDev版本 - private void accessDevSafely(List list, Map dictTreeMap, Map topicVersions) { - if (CollUtil.isNotEmpty(list)) { - for (CsEquipmentDeliveryPO item : list) { - try { - if (Objects.equals(dictTreeMap.get(item.getDevType()).getCode(), DicDataEnum.PORTABLE.getCode()) && Objects.equals(item.getStatus(), 1)) { - log.info("设备 {} 需要手动注册、接入", item.getNdid()); - } else { - String version = topicVersions.get(item.getNdid()); - if (Objects.isNull(version)) { - version = "V1"; - } - // 使用try-catch确保单个设备失败不影响其他设备 - try { - boolean success = csDeviceService.autoAccess2(item.getNdid(), version, 1); - if (success) { - redisUtil.saveByKey(AppRedisKey.DEVICE_MID + item.getNdid(), 1); - } - } catch (Exception e) { - log.error("设备 {} 接入异常: {}", item.getNdid(), e.getMessage()); - } - } - } catch (Exception e) { - log.error("处理设备 {} 失败: {}", item.getNdid(), e.getMessage()); + private void accessDevSafely(List list, + Map dictTreeMap, + Map topicVersions) { + for (CsEquipmentDeliveryPO item : list) { + try { + SysDicTreePO dicItem = dictTreeMap.get(item.getDevType()); + if (dicItem != null + && Objects.equals(dicItem.getCode(), DicDataEnum.PORTABLE.getCode()) + && Objects.equals(item.getStatus(), 1)) { + log.debug("设备 {} 需手动注册接入，跳过", item.getNdid()); + continue; } + + String version = topicVersions.getOrDefault(item.getNdid(), "V1"); + + mqttSemaphore.acquire(); + try { + boolean success = csDeviceService.autoAccess2(item.getNdid(), version, 1); + if (success) { + redisUtil.saveByKey(AppRedisKey.DEVICE_MID + item.getNdid(), 1); + } + } finally { + mqttSemaphore.release(); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + log.warn("设备 {} 处理被中断", item.getNdid()); + return; + } catch (Exception e) { + log.error("设备 {} 接入异常: {}", item.getNdid(), e.getMessage()); } } - } } + +//package com.njcn.access.runner; +// +//import cn.hutool.core.collection.CollUtil; +//import com.njcn.access.service.ICsEquipmentDeliveryService; +//import com.njcn.access.service.ICsTopicService; +//import com.njcn.access.service.impl.CsDeviceServiceImpl; +//import com.njcn.access.utils.ChannelObjectUtil; +//import com.njcn.csdevice.pojo.po.CsEquipmentDeliveryPO; +//import com.njcn.redis.pojo.enums.AppRedisKey; +//import com.njcn.redis.utils.RedisUtil; +//import com.njcn.system.enums.DicDataEnum; +//import com.njcn.system.pojo.po.SysDicTreePO; +//import lombok.RequiredArgsConstructor; +//import lombok.extern.slf4j.Slf4j; +//import org.springframework.boot.ApplicationArguments; +//import org.springframework.boot.ApplicationRunner; +//import org.springframework.stereotype.Component; +// +//import java.util.ArrayList; +//import java.util.List; +//import java.util.Map; +//import java.util.Objects; +//import java.util.concurrent.*; +// import java.util.stream.Collectors; +// +///** +// * @author xy +// * 定时轮询离线设备接入 +// */ +//@Component +//@Slf4j +//@RequiredArgsConstructor +//public class AutoAccessTimer implements ApplicationRunner { +// ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1); +// private static final long AUTO_TIME = 120L; +// private final RedisUtil redisUtil; +// private final ICsEquipmentDeliveryService csEquipmentDeliveryService; +// private final ICsTopicService csTopicService; +// private final CsDeviceServiceImpl csDeviceService; +// private final ChannelObjectUtil channelObjectUtil; +// +// @Override +// public void run(ApplicationArguments args) { +// if (scheduler.isShutdown() || scheduler.isTerminated()) { +// scheduler = Executors.newScheduledThreadPool(1); +// } +// Runnable task = () -> { +// try { +// executeScheduledTask(); +// } +// // 捕获所有Throwable，包括Error +// catch (Throwable t) { +// log.error("定时任务发生严重异常，尝试恢复", t); +// // 可以添加重启逻辑或告警 +// } +// }; +// ScheduledFuture future = scheduler.scheduleWithFixedDelay(task, AUTO_TIME, AUTO_TIME, TimeUnit.SECONDS); +// // 添加监控，如果任务被取消则重新调度 +// monitorScheduledTask(future); +// } +// +// //10分钟检查一下调度任务 +// private void monitorScheduledTask(ScheduledFuture future) { +// Thread monitorThread = new Thread(() -> { +// while (!Thread.currentThread().isInterrupted()) { +// try { +// //每10分钟检查一次 +// Thread.sleep(600000); +// if (future.isCancelled() || future.isDone()) { +// log.warn("定时任务被取消或完成，重新调度..."); +// // 重新启动任务 +// run(null); +// break; +// } +// } catch (InterruptedException e) { +// Thread.currentThread().interrupt(); +// log.warn("监控线程被中断"); +// break; +// } catch (Exception e) { +// log.error("监控任务异常", e); +// } +// } +// }, "Schedule-Monitor-Thread"); +// +// monitorThread.setDaemon(true); +// monitorThread.start(); +// } +// +// private void executeScheduledTask() { +// log.info("轮询定时任务执行中!"); +// List list = csEquipmentDeliveryService.getOfflineDev(); +// if (CollUtil.isNotEmpty(list)) { +// //获取字典数据 +// List dictTreeKey = channelObjectUtil.objectToList(redisUtil.getObjectByKey(AppRedisKey.DICT_TREE), SysDicTreePO.class); +// Map dictTreeMap = dictTreeKey.stream().collect(Collectors.toMap(SysDicTreePO::getId, item -> item)); +// //获取主题版本信息 +// List nDidIds = list.stream().map(CsEquipmentDeliveryPO::getNdid).collect(Collectors.toList()); +// Map topicVersions = csTopicService.getVersion(nDidIds); +// +// ExecutorService executor = Executors.newFixedThreadPool(10); +// try { +// List> subLists = CollUtil.split(list, 10); +// List> futures = new ArrayList<>(); +// for (List subList : subLists) { +// futures.add(executor.submit(() -> { +// try { +// accessDevSafely(subList,dictTreeMap,topicVersions); +// } catch (Exception e) { +// log.error("处理设备子列表异常", e); +// } +// return null; +// })); +// } +// for (Future future : futures) { +// try { +// future.get(5, TimeUnit.MINUTES); +// } catch (TimeoutException e) { +// log.error("任务执行超时", e); +// } catch (Exception e) { +// log.error("任务执行异常", e); +// } +// } +// } finally { +// executor.shutdown(); +// try { +// if (!executor.awaitTermination(1, TimeUnit.MINUTES)) { +// executor.shutdownNow(); +// } +// } catch (InterruptedException e) { +// executor.shutdownNow(); +// Thread.currentThread().interrupt(); +// } +// } +// } +// } +// +// //安全的accessDev版本 +// private void accessDevSafely(List list, Map dictTreeMap, Map topicVersions) { +// if (CollUtil.isNotEmpty(list)) { +// for (CsEquipmentDeliveryPO item : list) { +// try { +// if (Objects.equals(dictTreeMap.get(item.getDevType()).getCode(), DicDataEnum.PORTABLE.getCode()) && Objects.equals(item.getStatus(), 1)) { +// log.info("设备 {} 需要手动注册、接入", item.getNdid()); +// } else { +// String version = topicVersions.get(item.getNdid()); +// if (Objects.isNull(version)) { +// version = "V1"; +// } +// // 使用try-catch确保单个设备失败不影响其他设备 +// try { +// boolean success = csDeviceService.autoAccess2(item.getNdid(), version, 1); +// if (success) { +// redisUtil.saveByKey(AppRedisKey.DEVICE_MID + item.getNdid(), 1); +// } +// } catch (Exception e) { +// log.error("设备 {} 接入异常: {}", item.getNdid(), e.getMessage()); +// } +// } +// } catch (Exception e) { +// log.error("处理设备 {} 失败: {}", item.getNdid(), e.getMessage()); +// } +// } +// } +// +// } +//} 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 new file mode 100644 index 0000000..5f21baa --- /dev/null +++ b/iot-access/access-boot/src/main/java/com/njcn/access/runner/UpdateDevStatusTimer.java @@ -0,0 +1,26 @@ +package com.njcn.access.runner; + +import com.njcn.access.service.ICsHeartService; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; + + +/** + * @author xy + * 定时轮询离线设备接入 + */ +@Component +@Slf4j +@RequiredArgsConstructor +public class UpdateDevStatusTimer { + + private final ICsHeartService csHeartService; + + @Scheduled(cron = "0 0/30 * * * ?") + public void timer() { + csHeartService.updateDevStatus(); + } + +} diff --git a/iot-access/access-boot/src/main/java/com/njcn/access/service/ICsEquipmentDeliveryService.java b/iot-access/access-boot/src/main/java/com/njcn/access/service/ICsEquipmentDeliveryService.java index 7b5a526..6165deb 100644 --- a/iot-access/access-boot/src/main/java/com/njcn/access/service/ICsEquipmentDeliveryService.java +++ b/iot-access/access-boot/src/main/java/com/njcn/access/service/ICsEquipmentDeliveryService.java @@ -24,6 +24,11 @@ public interface ICsEquipmentDeliveryService extends IService list); + /** * 根据网关id修改软件信息 * @param nDid 网关id diff --git a/iot-access/access-boot/src/main/java/com/njcn/access/service/ICsHeartService.java b/iot-access/access-boot/src/main/java/com/njcn/access/service/ICsHeartService.java index 637bc0d..5946d0e 100644 --- a/iot-access/access-boot/src/main/java/com/njcn/access/service/ICsHeartService.java +++ b/iot-access/access-boot/src/main/java/com/njcn/access/service/ICsHeartService.java @@ -2,6 +2,8 @@ package com.njcn.access.service; import com.njcn.mq.message.HeartbeatTimeoutMessage; +import java.util.List; + /** * @author xy */ @@ -9,4 +11,8 @@ public interface ICsHeartService { void handleHeartbeat(HeartbeatTimeoutMessage message); + List getMissHeartbeat(); + + void updateDevStatus(); + } diff --git a/iot-access/access-boot/src/main/java/com/njcn/access/service/impl/CsDeviceServiceImpl.java b/iot-access/access-boot/src/main/java/com/njcn/access/service/impl/CsDeviceServiceImpl.java index f786c7f..ee2851d 100644 --- a/iot-access/access-boot/src/main/java/com/njcn/access/service/impl/CsDeviceServiceImpl.java +++ b/iot-access/access-boot/src/main/java/com/njcn/access/service/impl/CsDeviceServiceImpl.java @@ -54,6 +54,7 @@ import org.springframework.transaction.annotation.Transactional; import java.time.LocalDateTime; import java.util.*; +import java.util.concurrent.CompletableFuture; import java.util.stream.Collectors; /** @@ -1053,9 +1054,9 @@ public class CsDeviceServiceImpl implements ICsDeviceService { lineList = csLineFeignClient.findByNdid(nDid).getData(); for (CsLinePO item : lineList) { if (item.getClDid() == 0) { - updateLineIds(modelMap.get(0),item.getClDid(),nDid); + updateLineIds(modelMap.get(0),item.getClDid(),item.getLineId()); } else { - updateLineIds(modelMap.get(1),item.getClDid(),nDid); + updateLineIds(modelMap.get(1),item.getClDid(),item.getLineId()); } } } @@ -1126,9 +1127,9 @@ public class CsDeviceServiceImpl implements ICsDeviceService { lineList = csLineFeignClient.findByNdid(nDid).getData(); for (CsLinePO item : lineList) { if (item.getClDid() == 0) { - updateLineIds(modelMap.get(0),item.getClDid(),nDid); + updateLineIds(modelMap.get(0),item.getClDid(),item.getLineId()); } else { - updateLineIds(modelMap.get(1),item.getClDid(),nDid); + updateLineIds(modelMap.get(1),item.getClDid(),item.getLineId()); } } } @@ -1150,64 +1151,189 @@ public class CsDeviceServiceImpl implements ICsDeviceService { return result; } - @Transactional(rollbackFor = Exception.class) + /** + * 自动接入方法 - 重构版 + * 移除 @Transactional，手动控制事务边界 + */ public boolean autoAccess2(String nDid, String version, Integer mid) { - boolean result = false; - try { - String clientName = "NJCN-" + nDid.substring(nDid.length() - 6); - boolean mqttClient = mqttUtil.judgeClientOnline(clientName); - if (!mqttClient) { - //log.warn("装置 {} 客户端不在线", nDid); - return false; - } - - Map modelMap = new HashMap<>(); - redisUtil.delete(AppRedisKey.MODEL + nDid); - //redisUtil.deleteKeysByString(AppRedisKey.DEV_MODEL); - publisher.send("/Pfm/DevCmd/"+version+"/"+nDid, new Gson().toJson(getJson(mid,TypeEnum.TYPE_3.getCode())), 1, false); - try { - Thread.sleep(2000); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - //log.warn("线程休眠被中断: {}", e.getMessage()); - return false; - } - List modelId = channelObjectUtil.objectToList(redisUtil.getObjectByKey(AppRedisKey.MODEL + nDid), CsModelDto.class); - if (CollUtil.isEmpty(modelId)) { - //log.warn("装置 {} 未获取到模板信息", nDid); - return false; - } - CsEquipmentDeliveryVO vo = equipmentFeignClient.queryEquipmentByndid(nDid).getData(); - for (CsModelDto item : modelId) { - CsDevModelRelationPO po = new CsDevModelRelationPO(); - po.setDevId(vo.getId()); - po.setModelId(item.getModelId()); - po.setDid(item.getDid()); - po.setUpdateTime(LocalDateTime.now()); - csDevModelRelationService.addRelation(po); - modelMap.put(item.getType(), item.getModelId()); - } - Object object = redisUtil.getObjectByKey("accessLineInfo:" + nDid); - if (Objects.isNull(object)) { - LineInfoParam param = new LineInfoParam(); - param.setNDid(nDid); - deviceMessageFeignClient.getLineInfo(param); - } - publisher.send("/Pfm/DevCmd/"+version+"/"+nDid, new Gson().toJson(getJson(mid,TypeEnum.TYPE_5.getCode())), 1, false); - result = true; - } catch (Exception e) { - LogMessage logDto = new LogMessage(); - logDto.setUserIndex("定时任务"); - logDto.setLoginName("定时任务"); - logDto.setResult(0); - logDto.setOperate(nDid + "装置接入"); - logDto.setFailReason(e.getMessage()); - logMessageTemplate.sendMember(logDto); - //log.error("装置 {} 接入失败: {}", nDid, e.getMessage()); + // ========== 阶段1: 前置检查 (无事务) ========== + String clientName = "NJCN-" + nDid.substring(nDid.length() - 6); + if (!mqttUtil.judgeClientOnline(clientName)) { + return false; } - return result; + + // ========== 阶段2: 发送TYPE_3指令获取模板 (无事务, 无sleep) ========== + try { + publisher.send( + "/Pfm/DevCmd/" + version + "/" + nDid, + new Gson().toJson(getJson(mid, TypeEnum.TYPE_3.getCode())), + 1, false + ); + } catch (Exception e) { + logAndRecord(nDid, "发送TYPE_3指令失败", e); + return false; + } + + // ========== 阶段3: 异步等待模板数据 + 后续流程 ========== + // 【关键改造】不再sleep阻塞线程，而是用异步回调/延迟任务处理设备响应 + // 方案A(推荐): 设备上报模板后通过MQTT订阅回调触发后续DB写入 + // 方案B(兼容): 用ScheduledExecutorService延迟执行，不占用当前线程 + + CompletableFuture.supplyAsync(() -> waitForModelData(nDid, 3000)) + .thenAccept(modelList -> { + if (CollUtil.isEmpty(modelList)) { + log.warn("装置 {} 超时未获取到模板信息", nDid); + return; + } + try { + // DB写入在独立短事务中完成 + saveDeviceModelsInTransaction(nDid, modelList); + + // 补充线路信息(远程调用，事务外) + supplementLineInfo(nDid); + + // 发送TYPE_5指令(事务外) + publisher.send( + "/Pfm/DevCmd/" + version + "/" + nDid, + new Gson().toJson(getJson(mid, TypeEnum.TYPE_5.getCode())), + 1, false + ); + + redisUtil.saveByKey(AppRedisKey.DEVICE_MID + nDid, 1); + } catch (Exception e) { + logAndRecord(nDid, "装置接入后续流程失败", e); + } + }) + .exceptionally(ex -> { + logAndRecord(nDid, "装置接入异步流程异常", ex); + return null; + }); + + return true; // TYPE_3已发出即视为接入流程启动成功 } + /** + * 等待Redis中模板数据就绪（非阻塞轮询） + */ + private List waitForModelData(String nDid, long timeoutMs) { + long deadline = System.currentTimeMillis() + timeoutMs; + while (System.currentTimeMillis() < deadline) { + List data = channelObjectUtil.objectToList( + redisUtil.getObjectByKey(AppRedisKey.MODEL + nDid), CsModelDto.class); + if (CollUtil.isNotEmpty(data)) { + return data; + } + try { Thread.sleep(200); } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return null; + } + } + return null; + } + + /** + * 【独立短事务】仅包含DB写操作 + */ + @Transactional(rollbackFor = Exception.class) + public void saveDeviceModelsInTransaction(String nDid, List modelIdList) { + CsEquipmentDeliveryVO vo = equipmentFeignClient.queryEquipmentByndid(nDid).getData(); + if (vo == null) { + throw new BusinessException("设备信息不存在: " + nDid); + } + + redisUtil.delete(AppRedisKey.MODEL + nDid); // 消费后清理 + + for (CsModelDto item : modelIdList) { + CsDevModelRelationPO po = new CsDevModelRelationPO(); + po.setDevId(vo.getId()); + po.setModelId(item.getModelId()); + po.setDid(item.getDid()); + po.setUpdateTime(LocalDateTime.now()); + csDevModelRelationService.addRelation(po); + } + } + + /** + * 补充线路信息（事务外远程调用） + */ + private void supplementLineInfo(String nDid) { + Object cached = redisUtil.getObjectByKey("accessLineInfo:" + nDid); + if (Objects.isNull(cached)) { + LineInfoParam param = new LineInfoParam(); + param.setNDid(nDid); + deviceMessageFeignClient.getLineInfo(param); + } + } + + private void logAndRecord(String nDid, String operate, Throwable e) { + log.error("装置 {} {}: {}", nDid, operate, e.getMessage(), e); + LogMessage logDto = new LogMessage(); + logDto.setUserIndex("定时任务"); + logDto.setLoginName("定时任务"); + logDto.setResult(0); + logDto.setOperate(nDid + "装置接入"); + logDto.setFailReason(e.getMessage()); + logMessageTemplate.sendMember(logDto); + } + +// @Transactional(rollbackFor = Exception.class) +// public boolean autoAccess2(String nDid, String version, Integer mid) { +// boolean result = false; +// try { +// String clientName = "NJCN-" + nDid.substring(nDid.length() - 6); +// boolean mqttClient = mqttUtil.judgeClientOnline(clientName); +// if (!mqttClient) { +// //log.warn("装置 {} 客户端不在线", nDid); +// return false; +// } +// +// Map modelMap = new HashMap<>(); +// redisUtil.delete(AppRedisKey.MODEL + nDid); +// //redisUtil.deleteKeysByString(AppRedisKey.DEV_MODEL); +// publisher.send("/Pfm/DevCmd/"+version+"/"+nDid, new Gson().toJson(getJson(mid,TypeEnum.TYPE_3.getCode())), 1, false); +// try { +// Thread.sleep(2000); +// } catch (InterruptedException e) { +// Thread.currentThread().interrupt(); +// //log.warn("线程休眠被中断: {}", e.getMessage()); +// return false; +// } +// List modelId = channelObjectUtil.objectToList(redisUtil.getObjectByKey(AppRedisKey.MODEL + nDid), CsModelDto.class); +// if (CollUtil.isEmpty(modelId)) { +// //log.warn("装置 {} 未获取到模板信息", nDid); +// return false; +// } +// CsEquipmentDeliveryVO vo = equipmentFeignClient.queryEquipmentByndid(nDid).getData(); +// for (CsModelDto item : modelId) { +// CsDevModelRelationPO po = new CsDevModelRelationPO(); +// po.setDevId(vo.getId()); +// po.setModelId(item.getModelId()); +// po.setDid(item.getDid()); +// po.setUpdateTime(LocalDateTime.now()); +// csDevModelRelationService.addRelation(po); +// modelMap.put(item.getType(), item.getModelId()); +// } +// Object object = redisUtil.getObjectByKey("accessLineInfo:" + nDid); +// if (Objects.isNull(object)) { +// LineInfoParam param = new LineInfoParam(); +// param.setNDid(nDid); +// deviceMessageFeignClient.getLineInfo(param); +// } +// publisher.send("/Pfm/DevCmd/"+version+"/"+nDid, new Gson().toJson(getJson(mid,TypeEnum.TYPE_5.getCode())), 1, false); +// result = true; +// } catch (Exception e) { +// LogMessage logDto = new LogMessage(); +// logDto.setUserIndex("定时任务"); +// logDto.setLoginName("定时任务"); +// logDto.setResult(0); +// logDto.setOperate(nDid + "装置接入"); +// logDto.setFailReason(e.getMessage()); +// logMessageTemplate.sendMember(logDto); +// //log.error("装置 {} 接入失败: {}", nDid, e.getMessage()); +// } +// return result; +// } + /** * 组装报文 */ @@ -1224,10 +1350,10 @@ public class CsDeviceServiceImpl implements ICsDeviceService { /** * 修改监测点的模板id和数据集id */ - public void updateLineIds(String modelId, Integer clDid, String nDid) { + public void updateLineIds(String modelId, Integer clDid, String lineId) { CsDataSet dataSet = dataSetFeignClient.getSetByModelId(modelId,clDid).getData().get(0); CsLineParam csLineParam = new CsLineParam(); - csLineParam.setLineId(nDid + clDid); + csLineParam.setLineId(lineId); csLineParam.setDataSetId(dataSet.getId()); csLineParam.setModelId(modelId); csLineFeignClient.updateIds(csLineParam); diff --git a/iot-access/access-boot/src/main/java/com/njcn/access/service/impl/CsEquipmentDeliveryServiceImpl.java b/iot-access/access-boot/src/main/java/com/njcn/access/service/impl/CsEquipmentDeliveryServiceImpl.java index 5b4a5c2..36b17e2 100644 --- a/iot-access/access-boot/src/main/java/com/njcn/access/service/impl/CsEquipmentDeliveryServiceImpl.java +++ b/iot-access/access-boot/src/main/java/com/njcn/access/service/impl/CsEquipmentDeliveryServiceImpl.java @@ -48,6 +48,16 @@ public class CsEquipmentDeliveryServiceImpl extends ServiceImpl list) { + LambdaUpdateWrapper lambdaUpdateWrapper = new LambdaUpdateWrapper<>(); + lambdaUpdateWrapper + .set(CsEquipmentDeliveryPO::getRunStatus,AccessEnum.OFFLINE.getCode()) + .set(CsEquipmentDeliveryPO::getStatus,AccessEnum.REGISTERED.getCode()) + .in(CsEquipmentDeliveryPO::getNdid,list); + this.update(lambdaUpdateWrapper); + } + @Override public void updateSoftInfoBynDid(String nDid, String id, Integer moduleNumber) { LambdaUpdateWrapper lambdaUpdateWrapper = new LambdaUpdateWrapper<>(); diff --git 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 index 35b9e0f..b2a914e 100644 --- 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 @@ -1,7 +1,9 @@ package com.njcn.access.service.impl; +import cn.hutool.core.collection.CollUtil; import cn.hutool.core.collection.CollectionUtil; import cn.hutool.core.date.DatePattern; +import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; import com.njcn.access.enums.AccessEnum; import com.njcn.access.pojo.dto.NoticeUserDto; import com.njcn.access.service.ICsEquipmentDeliveryService; @@ -26,8 +28,10 @@ import org.springframework.stereotype.Service; import javax.annotation.Resource; import java.time.LocalDateTime; import java.time.format.DateTimeFormatter; +import java.util.ArrayList; import java.util.List; import java.util.Objects; +import java.util.Set; import java.util.stream.Collectors; /** @@ -70,6 +74,49 @@ public class CsHeartServiceImpl implements ICsHeartService { handleDeviceOffline(nDid); } + @Override + public List getMissHeartbeat() { + List result = new ArrayList<>(); + LambdaQueryWrapper queryWrapper = new LambdaQueryWrapper<>(); + queryWrapper.eq(CsEquipmentDeliveryPO::getRunStatus, 2).eq(CsEquipmentDeliveryPO::getUsageStatus,1); + List list = csEquipmentDeliveryService.list(queryWrapper); + //获取redis的数据 + Set keys = redisUtil.scanKeysByPattern("HEARTBEAT:*",100); + if (CollUtil.isNotEmpty(list)) { + list.forEach(item -> { + if (Objects.equals(item.getDevAccessMethod(),"CLD")) { + return; + } + String nDid = item.getNdid(); + boolean hasHeartbeat = keys.stream().anyMatch(key -> key.contains(nDid)); + if (!hasHeartbeat) { + result.add(nDid); + } + }); + } + return result; + } + + @Override + public void updateDevStatus() { + List missHeartbeat = getMissHeartbeat(); + if (CollUtil.isNotEmpty(missHeartbeat)) { + List list = new ArrayList<>(); + //修改状态 + csEquipmentDeliveryService.updateStatusByList(missHeartbeat); + //记录掉线 + missHeartbeat.forEach(item->{ + PqsCommunicateDto dto = new PqsCommunicateDto(); + dto.setTime(LocalDateTime.now().format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_PATTERN))); + dto.setDevId(item); + dto.setType(0); + dto.setDescription("通讯中断"); + list.add(dto); + }); + csCommunicateFeignClient.insertionList(list); + } + } + private void handleDeviceOffline(String nDid) { //装置下线 csEquipmentDeliveryService.updateRunStatusBynDid(nDid, AccessEnum.OFFLINE.getCode()); 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 516ae86..172c61f 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 @@ -168,16 +168,16 @@ public class StatServiceImpl implements IStatService { csLineLatestDataFeignClient.addData(csLineLatestData); } //判断设备运行状态 -// if (!Objects.isNull(po.getRunStatus()) && po.getRunStatus() == 1) { -// csDeviceFeignClient.updateRunStatus(appAutoDataMessage.getId(), AccessEnum.ONLINE.getCode()); -// //记录设备上线 -// PqsCommunicateDto dto = new PqsCommunicateDto(); -// dto.setTime(LocalDateTime.now().format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_PATTERN))); -// dto.setDevId(appAutoDataMessage.getId()); -// dto.setType(1); -// dto.setDescription("通讯正常"); -// csCommunicateFeignClient.insertion(dto); -// } + if (!Objects.isNull(po.getRunStatus()) && po.getRunStatus() == 1) { + csDeviceFeignClient.updateRunStatus(appAutoDataMessage.getId(), AccessEnum.ONLINE.getCode()); + //记录设备上线 + PqsCommunicateDto dto = new PqsCommunicateDto(); + dto.setTime(LocalDateTime.now().format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_PATTERN))); + dto.setDevId(appAutoDataMessage.getId()); + dto.setType(1); + dto.setDescription("通讯正常"); + csCommunicateFeignClient.insertion(dto); + } } } 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 805830c..2d118c9 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 @@ -437,7 +437,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); } @@ -446,9 +446,11 @@ public class EventServiceImpl implements IEventService { if (Objects.equals(cldLogMessage.getLevel(),"terminal")) { po.setDeviceId(cldLogMessage.getBusinessId()); } else { - CsLinePO line = csLineFeignClient.getById(cldLogMessage.getBusinessId()).getData(); - po.setDeviceId(line.getDeviceId()); - po.setLineId(cldLogMessage.getBusinessId()); + if (!Objects.isNull(cldLogMessage.getBusinessId()) && !Objects.equals(cldLogMessage.getBusinessId(),"")) { + CsLinePO line = csLineFeignClient.getById(cldLogMessage.getBusinessId()).getData(); + po.setDeviceId(line.getDeviceId()); + po.setLineId(cldLogMessage.getBusinessId()); + } } po.setType(3); }