Revert "1 测试 上电后通信 和 最终完成 定时任务功能"

zhmes-agecal-system/src/main/java/com/zehong/system/modbus/business/DeviceStatusReaderAndTimeSetter.java

@@ -13,6 +13,7 @@ import com.zehong.system.domain.TEquipmentAlarmData;
 import com.zehong.system.modbus.handler.dto.DeviceStatusReaderDto;
 import com.zehong.system.modbus.util.Modbus4jUtils;
 import com.zehong.system.service.ITEquipmentAlarmDataService;
+import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Component;
 
 import javax.annotation.Resource;
@@ -20,6 +21,7 @@ import java.lang.reflect.Field;
 import java.net.Socket;
 import java.util.List;
 import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 import java.util.function.Predicate;
 
@@ -30,42 +32,40 @@ import java.util.function.Predicate;
  */
 @Component
 public class DeviceStatusReaderAndTimeSetter {
-    // 常量配置（优化重试次数和超时，避免任务堆积）
+    // 常量配置（优化重试，减少耗时）
     public static final int START_ADDRESS = 0;
     public static final int REGISTER_COUNT = 10;
-    public static final int TARGET_VALUE = 1;
-    public static final int MAX_RETRIES = 2; // 减少重试次数，缩短总耗时
-    public static final int RETRY_DELAY = 300; // 重试间隔300ms
-    public static final int TASK_TIMEOUT_SECONDS = 20; // 单设备任务总超时
-    // 工厂
-    private static final ModbusFactory modbusFactory = new ModbusFactory();
-
-    // 1. 关键优化：固定线程池大小（不随设备数变化，根据CPU核数设置，如10）
-
-    // 设备通信线程池（Spring风格，避免静态线程池问题）
-    private final ExecutorService deviceExecutor = new ThreadPoolExecutor(
-            8, // 核心线程8个（72个设备，8线程，每线程处理9个）
-            10, // 最大线程10个
+    public static final int MAX_RETRIES = 1; // 仅1次重试（减少总耗时）
+    public static final int RETRY_DELAY = 200; // 重试间隔200ms（缩短等待）
+    // 线程池：Spring管理，非静态（避免资源泄漏）
+    private final ExecutorService deviceExecutor;
+
+    // 构造函数：初始化线程池（核心线程=设备数/10，避免过多）
+    public DeviceStatusReaderAndTimeSetter() {
+        this.deviceExecutor = new ThreadPoolExecutor(
+                4, // 核心线程4个（72个设备，4线程，每线程处理18个，效率适中）
+                8, // 最大线程8个（峰值扩容）
                 60,
                 TimeUnit.SECONDS,
-            new LinkedBlockingQueue<>(50),
+                new LinkedBlockingQueue<>(30), // 队列30个，避免任务堆积
                 new ThreadFactory() {
                     private int count = 0;
                     @Override
                     public Thread newThread(Runnable r) {
                         Thread thread = new Thread(r, "modbus-device-pool-" + (count++));
-                    thread.setDaemon(true);
+                        thread.setDaemon(true); // 守护线程，不阻塞JVM退出
                         return thread;
                     }
                 },
-            new ThreadPoolExecutor.CallerRunsPolicy() // 队列满时调用线程执行，避免任务丢失
+                new ThreadPoolExecutor.CallerRunsPolicy() // 队列满时调用线程执行（避免任务丢失）
         );
+    }
 
-
+    // 工厂（单例）
+    private static final ModbusFactory modbusFactory = new ModbusFactory();
     @Resource
     private ITEquipmentAlarmDataService alarmDataService;
-
-    private static final org.slf4j.Logger log = org.slf4j.LoggerFactory.getLogger(DeviceStatusReaderAndTimeSetter.class);
+    private static final org.slf4j.Logger log = LoggerFactory.getLogger(DeviceStatusReaderAndTimeSetter.class);
     /**
      * 读取设备寄存器（线程安全版）
      */
@@ -219,67 +219,77 @@ public class DeviceStatusReaderAndTimeSetter {
         }
     }
     /**
-     * 启动多设备监控（优化版：资源管控、超时控制）
+     * 启动多设备监控（新增portTimeout参数，统一超时）
      */
     public void startMultiDeviceMonitoring(
             String ip, int port, List<Integer> deviceIds,
             Consumer<DeviceStatusReaderDto> resultHandler,
-            Predicate<int[]> stopCondition) {
+            Predicate<int[]> stopCondition,
+            int portTimeout) { // 单个端口总超时（秒）
         if (deviceIds == null || deviceIds.isEmpty()) {
-            log.warn("设备ID列表为空，不执行监控");
+            log.warn("设备ID列表为空，不执行监控：IP={}, 端口={}", ip, port);
             return;
         }
 
         final CountDownLatch latch = new CountDownLatch(deviceIds.size());
-        log.info("启动多设备监控：IP={}, 端口={}, 设备数={}，活跃线程={}",
-                ip, port, deviceIds.size(), ((ThreadPoolExecutor) deviceExecutor).getActiveCount());
+        log.debug("启动端口{}监控：IP={}，设备数={}，超时={}s", ip, port, deviceIds.size(), portTimeout);
 
         for (int deviceId : deviceIds) {
             final int devId = deviceId;
-            // 提交设备任务（带超时控制）
             deviceExecutor.submit(() -> {
-                ModbusMaster threadMaster = null;
+                AtomicReference<ModbusMaster> threadMaster = null;
                 try {
-                    // 创建独立连接（每个设备一个连接，避免复用泄漏）
-                    threadMaster = createModbusMaster(ip, port);
-                    // 读取数据（带重试）
-                    int[] result = readWithConditionalRetry(threadMaster, ip, port, devId, stopCondition);
-                    // 回调结果（判空避免NPE）
+                    // 单个设备任务超时控制（portTimeout/设备数，避免整体超时）
+                    int devTimeout = Math.max(1, portTimeout * 1000 / deviceIds.size());
+                    ExecutorService devExecutor = Executors.newSingleThreadExecutor();
+                    Future<?> devFuture = devExecutor.submit(() -> {
+                        try {
+                            // 1. 创建独立连接（每个设备一个，避免复用泄漏）
+                            threadMaster.set(createModbusMaster(ip, port));
+                            // 2. 读取数据（带重试）
+                            int[] result = readWithConditionalRetry(threadMaster.get(), ip, port, devId, stopCondition);
+                            // 3. 回调结果
                             if (resultHandler != null) {
                                 resultHandler.accept(new DeviceStatusReaderDto(ip, port, devId, result));
                             }
+                        } catch (Exception e) {
+                            throw new RuntimeException("设备" + devId + "执行异常", e);
+                        }
+                    });
 
-                } catch (ModbusInitException e) {
-                    log.error("设备{}: 连接初始化失败", devId, e);
-                    recordAlarm("03", "ip:" + ip + ",port:" + port + ",deviceId:" + devId,
-                            "连接初始化失败：" + e.getMessage());
-                } catch (Throwable e) {
-                    log.error("设备{}: 监控任务异常", devId, e);
-                    String alarmMsg = e instanceof OutOfMemoryError ? "内存溢出：" + e.getMessage() : "监控任务异常：" + e.getMessage();
-                    recordAlarm("03", "ip:" + ip + ",port:" + port + ",deviceId:" + devId, alarmMsg);
+                    // 单个设备任务超时等待
+                    devFuture.get(devTimeout, TimeUnit.MILLISECONDS);
+                    devExecutor.shutdown();
 
+                } catch (TimeoutException e) {
+                    String errMsg = "设备" + devId + "超时：IP=" + ip + ", 端口=" + port;
+                    log.error(errMsg, e);
+                    recordAlarm("03", "ip:" + ip + ",port:" + port + ",deviceId:" + devId, errMsg);
+                } catch (Exception e) {
+                    String errMsg = "设备" + devId + "执行异常：IP=" + ip + ", 端口=" + port;
+                    log.error(errMsg, e);
+                    recordAlarm("03", "ip:" + ip + ",port:" + port + ",deviceId:" + devId, errMsg + "：" + e.getMessage());
                 } finally {
-                    // 强制销毁连接（无论成功/失败）
-                    destroyModbusMaster(threadMaster, devId);
+                    // 强制释放资源（无论成功/失败）
+                    destroyModbusMaster(threadMaster.get(), devId);
                     latch.countDown();
-                    log.debug("设备{}: 任务完成，剩余任务数：{}", devId, latch.getCount());
+                    log.debug("设备{}完成：IP={}, 端口={}，剩余任务={}", devId, ip, port, latch.getCount());
                 }
             });
         }
 
-        // 等待任务完成（超时后记录告警，不关闭线程池）
+        // 端口总超时等待（确保不阻塞上层Job）
         try {
-            if (!latch.await(TASK_TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
-                log.warn("IP{}: 部分设备监控超时（未完成：{}）", ip, latch.getCount());
-                recordAlarm("03", "ip:" + ip + ",port:" + port,
-                        "部分设备监控超时（未完成：" + latch.getCount() + "）");
+            if (!latch.await(portTimeout, TimeUnit.SECONDS)) {
+                String errMsg = "端口" + port + "部分设备超时：IP=" + ip + "，未完成=" + latch.getCount();
+                log.warn(errMsg);
+                recordAlarm("03", "ip:" + ip + ",port:" + port, errMsg);
             }
         } catch (InterruptedException e) {
-            log.error("IP{}: 监控任务被中断", ip, e);
+            log.error("端口{}监控被中断：IP={}", port, ip, e);
             Thread.currentThread().interrupt();
         }
     }
-
     /**
      * 统一告警记录（抽离避免重复代码）
      */

zhmes-agecal-system/src/main/java/com/zehong/system/service/impl/TStoreyInfoServiceImpl.java

@@ -342,7 +342,7 @@ public class TStoreyInfoServiceImpl implements ITStoreyInfoService
         } else {
             registerOffset = Arrays.asList(55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71, 72);
         }
-        deviceStatusReaderAndTimeSetter.startMultiDeviceMonitoring(split[0], Integer.parseInt(split[1]),registerOffset, modbusResultHandler, ModbusResultHandler.createDefaultStopCondition());
+//        deviceStatusReaderAndTimeSetter.startMultiDeviceMonitoring(split[0], Integer.parseInt(split[1]),registerOffset, modbusResultHandler, ModbusResultHandler.createDefaultStopCondition());
 
 
         return null;

zhmes-agecal-system/src/main/java/com/zehong/system/task/DeviceTaskScheduler.java

@@ -31,8 +31,12 @@ public class DeviceTaskScheduler {
 
     // 新增：每个设备任务的Quartz线程隔离（避免任务间干扰）
     private static final String THREAD_GROUP = "DEVICE_THREAD_GROUP";
+
+    // 常量：Cron周期（5分钟）、任务有效期（7天）
+    private static final String CRON_EXPRESSION = "0 0/5 * * * ?";
+    private static final int TASK_VALID_DAYS = 7;
     /**
-     * 创建设备监控任务（入口方法）
+     * 创建设备监控任务（入口：增加调度器健康检查、任务去重）
      */
     public void scheduleDeviceMonitoring(Long fStoreyId, String fPowerOutageIp, Integer fPowerOutagePort) {
         if (fStoreyId == null || fPowerOutageIp == null || fPowerOutagePort == null) {
@@ -43,63 +47,64 @@ public class DeviceTaskScheduler {
         try {
             log.info("=== 开始创建设备监控任务：fStoreyId={} ===", fStoreyId);
 
-            // 1. 确保调度器已启动（若依框架可能延迟启动）
-            if (!scheduler.isStarted()) {
-                log.warn("调度器未启动，手动启动...");
-                scheduler.start();
-            }
+            // 1. 调度器健康检查（确保线程池可用）
+            checkSchedulerHealth();
 
-            // 关键：添加调度器状态监控，确保线程池可用
-            SchedulerMetaData metaData = scheduler.getMetaData();
-            log.info("Quartz线程池状态：核心线程={} 任务总数={}",
-                    metaData.getThreadPoolSize(),
-                    metaData.getNumberOfJobsExecuted());
+            // 2. 任务去重（避免重复创建导致资源竞争）
+            if (isTaskExists("COMM_" + fStoreyId)) {
+                log.info("通信任务[COMM_{}]已存在，无需重复创建", fStoreyId);
+                // 检查现有触发器状态，若为ERROR则重建
+                if (isTriggerError("COMM_" + fStoreyId)) {
+                    log.warn("通信任务[COMM_{}]触发器状态为ERROR，重建触发器", fStoreyId);
+                    createHourlyCommunicationJob(fStoreyId);
+                }
+                return;
+            }
 
-            // 2. 创建两个核心任务
+            // 3. 创建核心任务
             createHourlyCommunicationJob(fStoreyId);
             createFinalExecutionJob(fStoreyId, fPowerOutageIp, fPowerOutagePort);
-
-            // 3. 验证任务是否创建成功（关键：打印触发器状态）
             checkTaskStatus(fStoreyId);
 
             log.info("=== 设备监控任务创建完成：fStoreyId={} ===", fStoreyId);
 
         } catch (SchedulerException e) {
             log.error("=== 创建设备监控任务失败：fStoreyId={} ===", fStoreyId, e);
+            cleanInvalidTask(fStoreyId); // 失败时清理残留
             throw new RuntimeException("Quartz任务调度失败", e);
         }
     }
 
     /**
-     * 1. 创建每2分钟执行的通信任务（CronTrigger）
+     * 1. 创建每5分钟执行的通信任务（核心优化：简化调度逻辑、调整Misfire策略）
      */
     private void createHourlyCommunicationJob(Long fStoreyId) throws SchedulerException {
         String jobId = "COMM_" + fStoreyId;
         JobKey jobKey = new JobKey(jobId, JOB_GROUP);
         TriggerKey triggerKey = new TriggerKey(jobId + "_TRIGGER", TRIGGER_GROUP);
 
-        // 1. 构建JobDetail（仅定义任务元数据，不涉及调度规则）
+        // 1. 构建JobDetail（仅元数据，禁止并发执行通过注解控制）
         JobDetail job = JobBuilder.newJob(DeviceCommunicationJob.class)
                 .withIdentity(jobKey)
                 .withDescription("设备" + fStoreyId + "每5分钟Modbus通信任务")
-                .usingJobData("fStoreyId", fStoreyId.toString())
-                .storeDurably(false)
+                .usingJobData("fStoreyId", fStoreyId.toString()) // 传递参数（String避免类型问题）
+                .storeDurably(false) // 触发器删除后Job自动失效
                 .requestRecovery(true) // 服务重启后恢复未完成任务
-                .build(); // 移除错误的 withSchedule() 调用
+                .build();
 
-        // 2. 构建CronTrigger（调度规则在此配置）
+        // 2. 构建CronTrigger（移除withMisfireThreshold，兼容低版本）
         CronTrigger trigger = TriggerBuilder.newTrigger()
                 .withIdentity(triggerKey)
                 .forJob(jobKey)
                 .withDescription("设备" + fStoreyId + "通信任务触发器（每5分钟）")
-                .withSchedule(CronScheduleBuilder.cronSchedule("0 0/5 * * * ?")
-                        // 关键：Misfire策略（错过触发后忽略，避免集中执行）
-                        .withMisfireHandlingInstructionDoNothing())
+                .withSchedule(CronScheduleBuilder.cronSchedule(CRON_EXPRESSION)
+                        // Misfire策略：错过触发后立即执行，再按原计划（低版本支持）
+                        .withMisfireHandlingInstructionFireAndProceed())
                 .startNow()
-                .endAt(Date.from(Instant.now().plus(7, ChronoUnit.DAYS))) // 7天有效期
+                .endAt(Date.from(Instant.now().plus(TASK_VALID_DAYS, ChronoUnit.DAYS)))
                 .build();
 
-        // 3. 原子操作：创建/更新任务
+        // 3. 原子操作：创建/更新（优先更新，避免删除重建）
         if (scheduler.checkExists(jobKey)) {
             Date nextFireTime = scheduler.rescheduleJob(triggerKey, trigger);
             log.info("通信任务[{}]更新触发器成功，下次执行时间：{}", jobId, nextFireTime);
@@ -109,15 +114,15 @@ public class DeviceTaskScheduler {
         }
     }
 
+
     /**
-     * 2. 创建5分钟后执行的最终任务（SimpleTrigger，仅执行一次）
+     * 2. 创建15分钟后执行的最终任务（保持原逻辑，优化超时）
      */
     private void createFinalExecutionJob(Long fStoreyId, String fPowerOutageIp, Integer fPowerOutagePort) throws SchedulerException {
         String jobId = "FINAL_" + fStoreyId;
         JobKey jobKey = new JobKey(jobId, JOB_GROUP);
         TriggerKey triggerKey = new TriggerKey(jobId + "_TRIGGER", TRIGGER_GROUP);
 
-        // 构建JobDetail
         JobDetail job = JobBuilder.newJob(FinalExecutionJob.class)
                 .withIdentity(jobKey)
                 .withDescription("设备" + fStoreyId + "最终执行任务（仅一次）")
@@ -128,7 +133,6 @@ public class DeviceTaskScheduler {
                 .requestRecovery(true)
                 .build();
 
-        // 构建SimpleTrigger
         Date executeTime = Date.from(Instant.now().plus(15, ChronoUnit.MINUTES));
         SimpleTrigger trigger = TriggerBuilder.newTrigger()
                 .withIdentity(triggerKey)
@@ -136,11 +140,10 @@ public class DeviceTaskScheduler {
                 .withDescription("设备" + fStoreyId + "最终任务触发器")
                 .startAt(executeTime)
                 .withSchedule(SimpleScheduleBuilder.simpleSchedule()
-                        .withMisfireHandlingInstructionFireNow() // 错过触发立即执行
-                        .withRepeatCount(0)) // 仅执行一次
+                        .withMisfireHandlingInstructionFireNow() // 错过立即执行
+                        .withRepeatCount(0)) // 仅一次
                 .build();
 
-        // 原子操作：创建/更新
         if (scheduler.checkExists(jobKey)) {
             Date nextFireTime = scheduler.rescheduleJob(triggerKey, trigger);
             log.info("最终任务[{}]更新触发器成功，执行时间：{}", jobId, nextFireTime);
@@ -149,6 +152,38 @@ public class DeviceTaskScheduler {
             log.info("最终任务[{}]创建成功，执行时间：{}", jobId, nextFireTime);
         }
     }
+    // ------------------------------ 工具方法：增强稳定性 ------------------------------
+    /**
+     * 检查Quartz调度器健康状态（避免线程池耗尽）
+     */
+    private void checkSchedulerHealth() throws SchedulerException {
+        if (!scheduler.isStarted()) {
+            log.warn("调度器未启动，手动启动...");
+            scheduler.start();
+        }
+
+        SchedulerMetaData metaData = scheduler.getMetaData();
+        // 低版本兼容：获取线程池大小和已执行任务数（替代活跃线程数）
+        int poolSize = metaData.getThreadPoolSize();
+        long executedJobs = metaData.getNumberOfJobsExecuted();
+        log.info("Quartz健康状态：线程池大小={}, 已执行任务数={}", poolSize, executedJobs);
+
+        // 线程池大小预警（根据实际需求调整阈值）
+        if (poolSize < 5) {
+            log.warn("Quartz线程池过小（当前={}），可能导致任务延迟", poolSize);
+        }
+    }
+    /**
+     * 检查触发器是否为ERROR状态
+     */
+    private boolean isTriggerError(String jobId) throws SchedulerException {
+        TriggerKey triggerKey = new TriggerKey(jobId + "_TRIGGER", TRIGGER_GROUP);
+        if (!scheduler.checkExists(triggerKey)) {
+            return false;
+        }
+        Trigger.TriggerState state = scheduler.getTriggerState(triggerKey);
+        return state == Trigger.TriggerState.ERROR;
+    }
     /**
      * 检查任务是否已存在（避免重复创建）
      */