refactor(access): 重构设备接入任务和优化线程管理

- 移除废弃的日志消息相关依赖和代码
- 优化 AccessApplicationRunner 和 AutoAccessTimer 的线程池使用
- 采用共享线程池和信号量控制并发访问设备
- 批量处理设备接入任务并添加批次间隔控制
- 重构 autoAccess2 方法移除事务注解并改用异步处理
- 修复更新线路ID时参数传递错误的问题
- 统一错误处理和异常捕获机制
- 添加更详细的日志记录和错误追踪功能

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);
+    }
+}

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<List<String>> getMissHeartbeat(){
+        String methodDescribe = getMethodDescribe("getMissHeartbeat");
+        List<String> result = csHeartService.getMissHeartbeat();
+        return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, result, methodDescribe);
+    }
+
+    @OperateInfo(info = LogEnum.BUSINESS_COMMON)
+    @PostMapping("/updateDevStatus")
+    @ApiOperation("手动调整状态")
+    public HttpResult<String> updateDevStatus(){
+        String methodDescribe = getMethodDescribe("updateDevStatus");
+        csHeartService.updateDevStatus();
+        return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, null, methodDescribe);
+    }
 }

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后发起自动接入
+        sharedScheduler.schedule(() -> {
-        Runnable task = () -> {
+            try {
-            log.info("系统重启,所有符合条件的装置发起接入!");
+                executeStartupRecovery();
+            } catch (Throwable t) {
+                log.error("启动恢复任务异常", t);
+            }
+        }, STARTUP_DELAY, TimeUnit.SECONDS);
+
+        log.info("启动恢复任务已调度, 延迟={}s", STARTUP_DELAY);
+    }
+
+    private void executeStartupRecovery() {
         List<CsEquipmentDeliveryPO> list = csEquipmentDeliveryService.getOnlineDev();
-            if (CollUtil.isNotEmpty(list)) {
+        if (CollUtil.isEmpty(list)) {
-                //获取字典数据
+            log.info("启动恢复: 无需恢复的设备");
-                List<SysDicTreePO> dictTreeKey = channelObjectUtil.objectToList(redisUtil.getObjectByKey(AppRedisKey.DICT_TREE), SysDicTreePO.class);
+            return;
-                Map<String, SysDicTreePO> dictTreeMap = dictTreeKey.stream().collect(Collectors.toMap(SysDicTreePO::getId, item -> item));
+        }
-                //获取主题版本信息
+        log.info("启动恢复: 待恢复设备数={}", list.size());
-                List<String> nDidIds = list.stream().map(CsEquipmentDeliveryPO::getNdid).collect(Collectors.toList());
-                Map<String,String> topicVersions = csTopicService.getVersion(nDidIds);
 
-                ExecutorService executor = Executors.newFixedThreadPool(10);
+        List<SysDicTreePO> dictTreeKey = channelObjectUtil.objectToList(
-                // 将任务平均分配给10个子列表
+                redisUtil.getObjectByKey(AppRedisKey.DICT_TREE), SysDicTreePO.class);
-                List<List<CsEquipmentDeliveryPO>> subLists = new ArrayList<>();
+        Map<String, SysDicTreePO> dictTreeMap = dictTreeKey.stream()
-                int partitionSize = list.size() / 10;
+                .collect(Collectors.toMap(SysDicTreePO::getId, item -> item, (a, b) -> a));
-                for (int i = 0; i < 10; i++) {
+
-                    int start = i * partitionSize;
+        List<String> ndidIds = list.stream().map(CsEquipmentDeliveryPO::getNdid).collect(Collectors.toList());
-                    int end = (i == 9) ? list.size() : start + partitionSize;
+        Map<String, String> topicVersions = csTopicService.getVersion(ndidIds);
-                    subLists.add(list.subList(start, end));
+
-                }
+        List<List<CsEquipmentDeliveryPO>> batches = CollUtil.split(list, BATCH_SIZE);
-                // 创建一个ExecutorService来处理这些任务
+
-                List<Future<Void>> futures = new ArrayList<>();
+        int batchCount = 0;
-                // 提交任务给线程池执行
+        for (List<CsEquipmentDeliveryPO> batch : batches) {
-                for (int i = 0; i < 10; i++) {
+            sharedWorkerPool.submit(() -> accessDevSafely(batch, dictTreeMap, topicVersions));
-                    int index = i;
+            // 非最后一批时等待
-                    futures.add(executor.submit(new Callable<Void>() {
+            if (++batchCount < batches.size()) {
-                        @Override
-                        public Void call() {
-                            accessDev(subLists.get(index), dictTreeMap, topicVersions);
-                            return null;
-                        }
-                    }));
-                }
-                // 等待所有任务完成
-                for (Future<Void> future : futures) {
                 try {
-                        future.get();
+                    Thread.sleep(BATCH_INTERVAL_MS);
-                    } catch (InterruptedException | ExecutionException e) {
+                } catch (InterruptedException e) {
-                        throw new RuntimeException(e);
+                    Thread.currentThread().interrupt();
+                    log.warn("批次间隔被中断，停止后续调度");
+                    return;
                 }
             }
-                // 关闭ExecutorService
-                executor.shutdown();
         }
-        };
-        scheduler.schedule(task,ACCESS_TIME,TimeUnit.SECONDS);
-        scheduler.shutdown();
     }
 
-    public void accessDev(List<CsEquipmentDeliveryPO> list, Map<String, SysDicTreePO> dictTreeMap,  Map<String,String> topicVersions) {
+    private void accessDevSafely(List<CsEquipmentDeliveryPO> list,
-        if (CollUtil.isNotEmpty(list)) {
+                                 Map<String, SysDicTreePO> dictTreeMap,
+                                 Map<String, String> topicVersions) {
+        for (CsEquipmentDeliveryPO item : list) {
             try {
-                list.forEach(item->{
+                SysDicTreePO dicItem = dictTreeMap.get(item.getDevType());
-                    //System.out.println(Thread.currentThread().getName() + ": reboot : nDid : " + item.getNdid());
+                if (dicItem != null
-                    //判断设备类型 便携式设备需要特殊处理 未注册的要先注册、再接入;已注册的直接重新接入
+                        && Objects.equals(dicItem.getCode(), DicDataEnum.PORTABLE.getCode())
-                    if (Objects.equals(dictTreeMap.get(item.getDevType()).getCode(), DicDataEnum.PORTABLE.getCode()) && Objects.equals(item.getStatus(),1)) {
+                        && Objects.equals(item.getStatus(), 1)) {
-                        //csDeviceService.wlDevRegister(item.getNdid());
+                    log.debug("设备 {} 需手动注册接入，跳过", item.getNdid());
-                        log.info("请先手动注册、接入");
+                    continue;
-                    } else {
-                        String version = topicVersions.get(item.getNdid());
-                        if (Objects.isNull(version)) {
-                            version = "V1";
                 }
+
+                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(e.getMessage());
+                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<CsEquipmentDeliveryPO> list = csEquipmentDeliveryService.getOnlineDev();
+//            if (CollUtil.isNotEmpty(list)) {
+//                //获取字典数据
+//                List<SysDicTreePO> dictTreeKey = channelObjectUtil.objectToList(redisUtil.getObjectByKey(AppRedisKey.DICT_TREE), SysDicTreePO.class);
+//                Map<String, SysDicTreePO> dictTreeMap = dictTreeKey.stream().collect(Collectors.toMap(SysDicTreePO::getId, item -> item));
+//                //获取主题版本信息
+//                List<String> nDidIds = list.stream().map(CsEquipmentDeliveryPO::getNdid).collect(Collectors.toList());
+//                Map<String,String> topicVersions = csTopicService.getVersion(nDidIds);
+//
+//                ExecutorService executor = Executors.newFixedThreadPool(10);
+//                // 将任务平均分配给10个子列表
+//                List<List<CsEquipmentDeliveryPO>> 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<Future<Void>> futures = new ArrayList<>();
+//                // 提交任务给线程池执行
+//                for (int i = 0; i < 10; i++) {
+//                    int index = i;
+//                    futures.add(executor.submit(new Callable<Void>() {
+//                        @Override
+//                        public Void call() {
+//                            accessDev(subLists.get(index), dictTreeMap, topicVersions);
+//                            return null;
+//                        }
+//                    }));
+//                }
+//                // 等待所有任务完成
+//                for (Future<Void> 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<CsEquipmentDeliveryPO> list, Map<String, SysDicTreePO> dictTreeMap,  Map<String,String> 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());
+//            }
+//        }
+//    }
+//
+//}

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()) {
+        sharedScheduler.scheduleWithFixedDelay(() -> {
-            scheduler = Executors.newScheduledThreadPool(1);
-        }
-        Runnable task = () -> {
             try {
                 executeScheduledTask();
+            } catch (Throwable t) {
+                log.error("定时轮询任务异常", t);
             }
-            // 捕获所有Throwable，包括Error
+        }, AUTO_TIME, AUTO_TIME, TimeUnit.SECONDS);
-            catch (Throwable t) {
-                log.error("定时任务发生严重异常，尝试恢复", t);
-                // 可以添加重启逻辑或告警
-            }
-        };
-        ScheduledFuture<?> future = scheduler.scheduleWithFixedDelay(task, AUTO_TIME, AUTO_TIME, TimeUnit.SECONDS);
-        // 添加监控，如果任务被取消则重新调度
-        monitorScheduledTask(future);
-    }
 
-    //10分钟检查一下调度任务
+        log.info("自动接入定时任务已启动, 周期={}s", AUTO_TIME);
-    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<CsEquipmentDeliveryPO> list = csEquipmentDeliveryService.getOfflineDev();
-        if (CollUtil.isNotEmpty(list)) {
+        if (CollUtil.isEmpty(list)) {
-            //获取字典数据
+            return;
-            List<SysDicTreePO> dictTreeKey = channelObjectUtil.objectToList(redisUtil.getObjectByKey(AppRedisKey.DICT_TREE), SysDicTreePO.class);
+        }
-            Map<String, SysDicTreePO> dictTreeMap = dictTreeKey.stream().collect(Collectors.toMap(SysDicTreePO::getId, item -> item));
+        log.info("轮询定时任务执行中, 待处理设备数={}", list.size());
-            //获取主题版本信息
-            List<String> nDidIds = list.stream().map(CsEquipmentDeliveryPO::getNdid).collect(Collectors.toList());
-            Map<String,String> topicVersions = csTopicService.getVersion(nDidIds);
 
-            ExecutorService executor = Executors.newFixedThreadPool(10);
+        List<SysDicTreePO> dictTreeKey = channelObjectUtil.objectToList(
+                redisUtil.getObjectByKey(AppRedisKey.DICT_TREE), SysDicTreePO.class);
+        Map<String, SysDicTreePO> dictTreeMap = dictTreeKey.stream()
+                .collect(Collectors.toMap(SysDicTreePO::getId, item -> item, (a, b) -> a));
+
+        List<String> ndidIds = list.stream().map(CsEquipmentDeliveryPO::getNdid).collect(Collectors.toList());
+        Map<String, String> topicVersions = csTopicService.getVersion(ndidIds);
+
+        List<List<CsEquipmentDeliveryPO>> batches = CollUtil.split(list, BATCH_SIZE);
+
+        int batchCount = 0;
+        for (List<CsEquipmentDeliveryPO> batch : batches) {
+            sharedWorkerPool.submit(() -> accessDevSafely(batch, dictTreeMap, topicVersions));
+
+            // 非最后一批时等待
+            if (++batchCount < batches.size()) {
                 try {
-                List<List<CsEquipmentDeliveryPO>> subLists = CollUtil.split(list, 10);
+                    Thread.sleep(BATCH_INTERVAL_MS);
-                List<Future<Void>> futures = new ArrayList<>();
-                for (List<CsEquipmentDeliveryPO> subList : subLists) {
-                    futures.add(executor.submit(() -> {
-                        try {
-                            accessDevSafely(subList,dictTreeMap,topicVersions);
-                        } catch (Exception e) {
-                            log.error("处理设备子列表异常", e);
-                        }
-                        return null;
-                    }));
-                }
-                for (Future<Void> 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();
+                    log.warn("批次间隔被中断，停止后续调度");
+                    return;
                 }
             }
         }
     }
 
-    //安全的accessDev版本
+    private void accessDevSafely(List<CsEquipmentDeliveryPO> list,
-    private void accessDevSafely(List<CsEquipmentDeliveryPO> list, Map<String, SysDicTreePO> dictTreeMap, Map<String,String> topicVersions) {
+                                 Map<String, SysDicTreePO> dictTreeMap,
-        if (CollUtil.isNotEmpty(list)) {
+                                 Map<String, String> topicVersions) {
         for (CsEquipmentDeliveryPO item : list) {
             try {
-                    if (Objects.equals(dictTreeMap.get(item.getDevType()).getCode(), DicDataEnum.PORTABLE.getCode()) && Objects.equals(item.getStatus(), 1)) {
+                SysDicTreePO dicItem = dictTreeMap.get(item.getDevType());
-                        log.info("设备 {} 需要手动注册、接入", item.getNdid());
+                if (dicItem != null
-                    } else {
+                        && Objects.equals(dicItem.getCode(), DicDataEnum.PORTABLE.getCode())
-                        String version = topicVersions.get(item.getNdid());
+                        && Objects.equals(item.getStatus(), 1)) {
-                        if (Objects.isNull(version)) {
+                    log.debug("设备 {} 需手动注册接入，跳过", item.getNdid());
-                            version = "V1";
+                    continue;
                 }
-                        // 使用try-catch确保单个设备失败不影响其他设备
+
+                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());
             }
         }
-                } 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<CsEquipmentDeliveryPO> list = csEquipmentDeliveryService.getOfflineDev();
+//        if (CollUtil.isNotEmpty(list)) {
+//            //获取字典数据
+//            List<SysDicTreePO> dictTreeKey = channelObjectUtil.objectToList(redisUtil.getObjectByKey(AppRedisKey.DICT_TREE), SysDicTreePO.class);
+//            Map<String, SysDicTreePO> dictTreeMap = dictTreeKey.stream().collect(Collectors.toMap(SysDicTreePO::getId, item -> item));
+//            //获取主题版本信息
+//            List<String> nDidIds = list.stream().map(CsEquipmentDeliveryPO::getNdid).collect(Collectors.toList());
+//            Map<String,String> topicVersions = csTopicService.getVersion(nDidIds);
+//
+//            ExecutorService executor = Executors.newFixedThreadPool(10);
+//            try {
+//                List<List<CsEquipmentDeliveryPO>> subLists = CollUtil.split(list, 10);
+//                List<Future<Void>> futures = new ArrayList<>();
+//                for (List<CsEquipmentDeliveryPO> subList : subLists) {
+//                    futures.add(executor.submit(() -> {
+//                        try {
+//                            accessDevSafely(subList,dictTreeMap,topicVersions);
+//                        } catch (Exception e) {
+//                            log.error("处理设备子列表异常", e);
+//                        }
+//                        return null;
+//                    }));
+//                }
+//                for (Future<Void> 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<CsEquipmentDeliveryPO> list, Map<String, SysDicTreePO> dictTreeMap, Map<String,String> 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());
+//                }
+//            }
+//        }
+//
+//    }
+//}

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();
+    }
+
+}

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

@@ -24,6 +24,11 @@ public interface ICsEquipmentDeliveryService extends IService<CsEquipmentDeliver
      */
     void updateStatusBynDid(String nDid,Integer status,String engineeringId, String projectId);
 
+    /**
+     * 批量修改设备通讯状态和接入状态
+     */
+    void updateStatusByList(List<String> list);
+
     /**
      * 根据网关id修改软件信息
      * @param nDid 网关id

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<String> getMissHeartbeat();
+
+    void updateDevStatus();
+
 }

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,52 +1151,122 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
         return result;
     }
 
-    @Transactional(rollbackFor = Exception.class)
+    /**
+     * 自动接入方法 - 重构版
+     * 移除 @Transactional，手动控制事务边界
+     */
     public boolean autoAccess2(String nDid, String version, Integer mid) {
-        boolean result = false;
+        // ========== 阶段1: 前置检查 (无事务) ==========
-        try {
         String clientName = "NJCN-" + nDid.substring(nDid.length() - 6);
-            boolean mqttClient = mqttUtil.judgeClientOnline(clientName);
+        if (!mqttUtil.judgeClientOnline(clientName)) {
-            if (!mqttClient) {
-                //log.warn("装置 {} 客户端不在线", nDid);
             return false;
         }
 
-            Map<Integer, String> modelMap = new HashMap<>();
+        // ========== 阶段2: 发送TYPE_3指令获取模板 (无事务, 无sleep) ==========
-            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);
+            publisher.send(
-            } catch (InterruptedException e) {
+                    "/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<CsModelDto> waitForModelData(String nDid, long timeoutMs) {
+        long deadline = System.currentTimeMillis() + timeoutMs;
+        while (System.currentTimeMillis() < deadline) {
+            List<CsModelDto> 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();
-                //log.warn("线程休眠被中断: {}", e.getMessage());
+                return null;
-                return false;
             }
-            List<CsModelDto> modelId = channelObjectUtil.objectToList(redisUtil.getObjectByKey(AppRedisKey.MODEL + nDid), CsModelDto.class);
-            if (CollUtil.isEmpty(modelId)) {
-                //log.warn("装置 {} 未获取到模板信息", nDid);
-                return false;
         }
+        return null;
+    }
+
+    /**
+     * 【独立短事务】仅包含DB写操作
+     */
+    @Transactional(rollbackFor = Exception.class)
+    public void saveDeviceModelsInTransaction(String nDid, List<CsModelDto> modelIdList) {
         CsEquipmentDeliveryVO vo = equipmentFeignClient.queryEquipmentByndid(nDid).getData();
-            for (CsModelDto item : modelId) {
+        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);
-                modelMap.put(item.getType(), item.getModelId());
         }
-            Object object = redisUtil.getObjectByKey("accessLineInfo:" + nDid);
+    }
-            if (Objects.isNull(object)) {
+
+    /**
+     * 补充线路信息（事务外远程调用）
+     */
+    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);
         }
-            publisher.send("/Pfm/DevCmd/"+version+"/"+nDid, new Gson().toJson(getJson(mid,TypeEnum.TYPE_5.getCode())), 1, false);
+    }
-            result = true;
+
-        } catch (Exception e) {
+    private void logAndRecord(String nDid, String operate, Throwable e) {
+        log.error("装置 {} {}: {}", nDid, operate, e.getMessage(), e);
         LogMessage logDto = new LogMessage();
         logDto.setUserIndex("定时任务");
         logDto.setLoginName("定时任务");
@@ -1203,11 +1274,66 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
         logDto.setOperate(nDid + "装置接入");
         logDto.setFailReason(e.getMessage());
         logMessageTemplate.sendMember(logDto);
-            //log.error("装置 {} 接入失败: {}", nDid, e.getMessage());
-        }
-        return result;
     }
 
+//    @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<Integer, String> 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<CsModelDto> 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);

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

@@ -48,6 +48,16 @@ public class CsEquipmentDeliveryServiceImpl extends ServiceImpl<CsEquipmentDeliv
         this.update(lambdaUpdateWrapper);
     }
 
+    @Override
+    public void updateStatusByList(List<String> list) {
+        LambdaUpdateWrapper<CsEquipmentDeliveryPO> 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<CsEquipmentDeliveryPO> lambdaUpdateWrapper = new LambdaUpdateWrapper<>();

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<String> getMissHeartbeat() {
+        List<String> result = new ArrayList<>();
+        LambdaQueryWrapper<CsEquipmentDeliveryPO> queryWrapper = new LambdaQueryWrapper<>();
+        queryWrapper.eq(CsEquipmentDeliveryPO::getRunStatus, 2).eq(CsEquipmentDeliveryPO::getUsageStatus,1);
+        List<CsEquipmentDeliveryPO> list = csEquipmentDeliveryService.list(queryWrapper);
+        //获取redis的数据
+        Set<String> 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<String> missHeartbeat = getMissHeartbeat();
+        if (CollUtil.isNotEmpty(missHeartbeat)) {
+            List<PqsCommunicateDto> 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());

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) {
+            if (!Objects.isNull(po.getRunStatus()) && po.getRunStatus() == 1) {
-//                csDeviceFeignClient.updateRunStatus(appAutoDataMessage.getId(), AccessEnum.ONLINE.getCode());
+                csDeviceFeignClient.updateRunStatus(appAutoDataMessage.getId(), AccessEnum.ONLINE.getCode());
-//                //记录设备上线
+                //记录设备上线
-//                PqsCommunicateDto dto = new PqsCommunicateDto();
+                PqsCommunicateDto dto = new PqsCommunicateDto();
-//                dto.setTime(LocalDateTime.now().format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_PATTERN)));
+                dto.setTime(LocalDateTime.now().format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_PATTERN)));
-//                dto.setDevId(appAutoDataMessage.getId());
+                dto.setDevId(appAutoDataMessage.getId());
-//                dto.setType(1);
+                dto.setType(1);
-//                dto.setDescription("通讯正常");
+                dto.setDescription("通讯正常");
-//                csCommunicateFeignClient.insertion(dto);
+                csCommunicateFeignClient.insertion(dto);
-//            }
+            }
         }
     }
 

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

@@ -446,10 +446,12 @@ public class EventServiceImpl implements IEventService {
             if (Objects.equals(cldLogMessage.getLevel(),"terminal")) {
                 po.setDeviceId(cldLogMessage.getBusinessId());
             } else {
+                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);
         }
         csEventService.save(po);