From 074371da025b9f0a449ea7d3bc9d39b44555e80d Mon Sep 17 00:00:00 2001 From: tianyongbao Date: Mon, 6 Jul 2026 21:41:53 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E6=8E=A7=E5=88=B6=E5=99=A8=E5=AE=9A?= =?UTF-8?q?=E6=97=B6=E4=BB=BB=E5=8A=A1=EF=BC=8C=E5=90=8C=E4=B8=80=E4=B8=AA?= =?UTF-8?q?=E6=8E=A7=E5=88=B6=E5=99=A8=E7=9A=84=E5=BC=80=E5=85=B3=E5=90=8C?= =?UTF-8?q?=E4=B8=80=E6=97=B6=E9=97=B4=E4=B8=8B=E5=8F=91=E7=9A=84=E8=AF=9D?= =?UTF-8?q?=EF=BC=8C=E9=97=B4=E9=9A=9420=E7=A7=92=E4=B8=8B=E5=8F=91?= =?UTF-8?q?=E4=B8=80=E6=9D=A1=E6=8C=87=E4=BB=A4=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../impl/TimingCtrlExecuteServiceImpl.java | 142 ++++++++++++++++-- 1 file changed, 127 insertions(+), 15 deletions(-) diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/service/impl/TimingCtrlExecuteServiceImpl.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/service/impl/TimingCtrlExecuteServiceImpl.java index 30709b4..7c4699d 100644 --- a/intc-modules/intc-iot/src/main/java/com/intc/iot/service/impl/TimingCtrlExecuteServiceImpl.java +++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/service/impl/TimingCtrlExecuteServiceImpl.java @@ -25,10 +25,15 @@ import java.time.LocalDate; import java.time.LocalDateTime; import java.time.LocalTime; import java.time.format.DateTimeFormatter; +import java.util.ArrayList; import java.util.Date; import java.util.HashMap; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; /** * 定时控制执行服务实现 @@ -43,6 +48,11 @@ public class TimingCtrlExecuteServiceImpl implements TimingCtrlExecuteService { private static final DateTimeFormatter DATE_KEY_FORMATTER = DateTimeFormatter.ofPattern("yyyyMMdd"); private static final long SCAN_WINDOW_MINUTES = 4L; + /** + * 同一控制器多个开关同时到期时，每条指令之间的下发间隔（秒） + */ + private static final long SAME_DEVICE_DELAY_SECONDS = 20L; + /** * 启动时打印日志，用于确认 Bean 是否成功创建 */ @@ -52,11 +62,34 @@ public class TimingCtrlExecuteServiceImpl implements TimingCtrlExecuteService { SCAN_WINDOW_MINUTES, DAILY_EXEC_LOCK_PREFIX); } + @jakarta.annotation.PreDestroy + public void destroy() { + log.info("[TimingCtrl] 关闭定时控制指令调度线程池..."); + commandScheduler.shutdown(); + try { + if (!commandScheduler.awaitTermination(5, TimeUnit.SECONDS)) { + commandScheduler.shutdownNow(); + } + } catch (InterruptedException e) { + commandScheduler.shutdownNow(); + Thread.currentThread().interrupt(); + } + } + private final DeviceMapper deviceMapper; private final DeviceSwitchMapper deviceSwitchMapper; private final TimingCtrlMapper timingCtrlMapper; private final IotCloudService iotCloudService; + /** + * 延迟下发指令的调度线程池（守护线程，不阻塞应用关闭） + */ + private final ScheduledExecutorService commandScheduler = Executors.newSingleThreadScheduledExecutor(r -> { + Thread t = new Thread(r, "timing-ctrl-cmd-scheduler"); + t.setDaemon(true); + return t; + }); + @Override public SwitchTurnResult executeSwitch(Long switchId, Integer targetStatus, String opTypeLabel, String opMessagePrefix, boolean timingOp) { @@ -160,7 +193,9 @@ public class TimingCtrlExecuteServiceImpl implements TimingCtrlExecuteService { LocalDateTime now = LocalDateTime.now(); log.info("[TimingCtrl] 扫描开始，当前时间={}，查到 {} 条启用的定时控制", now, list.size()); - int executedCount = 0; + + // 第一轮：收集所有到期任务，按设备ID分组 + Map> tasksByDevice = new LinkedHashMap<>(); for (TimingCtrl timingCtrl : list) { try { if (timingCtrl.getSwitchId() == null || timingCtrl.getOpenTime() == null || timingCtrl.getCloseTime() == null) { @@ -181,15 +216,17 @@ public class TimingCtrlExecuteServiceImpl implements TimingCtrlExecuteService { continue; } + // 查询开关所属设备ID，用于按设备分组错峰下发 + Long deviceId = getDeviceIdBySwitchId(timingCtrl.getSwitchId()); + if (deviceId == null) { + log.warn("[TimingCtrl] 开关不存在，跳过，switchId={}", timingCtrl.getSwitchId()); + continue; + } + if (openDue) { if (tryLock(timingCtrl.getId(), now.toLocalDate(), "open")) { - SwitchTurnResult result = executeSwitch(timingCtrl.getSwitchId(), 1, - "定时控制管理", "定时任务到达开启时间", true); - if (result.isSuccess()) { - executedCount++; - } else { - log.warn("[TimingCtrl] 开启执行失败，timingCtrlId={}, msg={}", timingCtrl.getId(), result.getMessage()); - } + tasksByDevice.computeIfAbsent(deviceId, k -> new ArrayList<>()) + .add(new DueTask(timingCtrl.getId(), timingCtrl.getSwitchId(), 1)); } else { log.debug("[TimingCtrl] 开启: 今天已执行过，跳过，id={}", timingCtrl.getId()); } @@ -197,13 +234,8 @@ public class TimingCtrlExecuteServiceImpl implements TimingCtrlExecuteService { if (closeDue) { if (tryLock(timingCtrl.getId(), now.toLocalDate(), "close")) { - SwitchTurnResult result = executeSwitch(timingCtrl.getSwitchId(), 0, - "定时控制管理", "定时任务到达关闭时间", true); - if (result.isSuccess()) { - executedCount++; - } else { - log.warn("[TimingCtrl] 关闭执行失败，timingCtrlId={}, msg={}", timingCtrl.getId(), result.getMessage()); - } + tasksByDevice.computeIfAbsent(deviceId, k -> new ArrayList<>()) + .add(new DueTask(timingCtrl.getId(), timingCtrl.getSwitchId(), 0)); } else { log.debug("[TimingCtrl] 关闭: 今天已执行过，跳过，id={}", timingCtrl.getId()); } @@ -213,9 +245,89 @@ public class TimingCtrlExecuteServiceImpl implements TimingCtrlExecuteService { } } + if (tasksByDevice.isEmpty()) { + log.info("[TimingCtrl] 本次扫描没有需要执行的定时控制"); + return 0; + } + + // 第二轮：按设备分组下发指令，同一设备的多个开关每隔 SAME_DEVICE_DELAY_SECONDS 秒下发一条 + int executedCount = 0; + for (Map.Entry> entry : tasksByDevice.entrySet()) { + Long deviceId = entry.getKey(); + List tasks = entry.getValue(); + log.info("[TimingCtrl] 设备{} 共有 {} 条定时控制待执行", deviceId, tasks.size()); + + for (int i = 0; i < tasks.size(); i++) { + DueTask task = tasks.get(i); + long delaySeconds = i * SAME_DEVICE_DELAY_SECONDS; + + if (delaySeconds == 0) { + // 第一条立即下发 + SwitchTurnResult result = executeSwitch(task.switchId, task.targetStatus, + "定时控制管理", "定时任务到达" + (task.targetStatus == 1 ? "开启" : "关闭") + "时间", true); + if (result.isSuccess()) { + executedCount++; + } else { + log.warn("[TimingCtrl] 执行失败，timingCtrlId={}, msg={}", task.timingCtrlId, result.getMessage()); + } + } else { + // 同一设备后续开关延迟下发，避免短时间内集中下发多条指令导致硬件无法及时处理 + final Long tcId = task.timingCtrlId; + final Long switchId = task.switchId; + final int targetStatus = task.targetStatus; + final long delay = delaySeconds; + log.info("[TimingCtrl] 定时控制{} 将延迟{}秒后执行（设备{}排队）", tcId, delay, deviceId); + commandScheduler.schedule(() -> { + try { + TenantHelper.ignore(() -> { + log.info("[TimingCtrl] 开始延迟执行定时控制，timingCtrlId={}", tcId); + SwitchTurnResult result = executeSwitch(switchId, targetStatus, + "定时控制管理", "定时任务到达" + (targetStatus == 1 ? "开启" : "关闭") + "时间", true); + if (result.isSuccess()) { + log.info("[TimingCtrl] 延迟执行成功，timingCtrlId={}", tcId); + } else { + log.warn("[TimingCtrl] 延迟执行失败，timingCtrlId={}, msg={}", tcId, result.getMessage()); + } + }); + } catch (Exception e) { + log.error("[TimingCtrl] 延迟执行异常，timingCtrlId={}: {}", tcId, e.getMessage(), e); + } + }, delay, TimeUnit.SECONDS); + executedCount++; + } + } + } + return executedCount; } + /** + * 根据开关ID查询所属设备ID + */ + private Long getDeviceIdBySwitchId(Long switchId) { + DeviceSwitch ds = deviceSwitchMapper.selectOne( + new LambdaQueryWrapper() + .eq(DeviceSwitch::getId, switchId) + .select(DeviceSwitch::getDeviceId) + ); + return ds != null ? ds.getDeviceId() : null; + } + + /** + * 到期的定时控制任务 + */ + private static class DueTask { + final Long timingCtrlId; + final Long switchId; + final int targetStatus; + + DueTask(Long timingCtrlId, Long switchId, int targetStatus) { + this.timingCtrlId = timingCtrlId; + this.switchId = switchId; + this.targetStatus = targetStatus; + } + } + @Override public List listEnabledDailyTimingCtrls() { return timingCtrlMapper.selectList(