fix: 控制器定时任务,同一个控制器的开关同一时间下发的话,间隔20秒下发一条指令。

This commit is contained in:
tianyongbao
2026-07-06 21:41:53 +08:00
parent 66d8dc2fe7
commit 074371da02

View File

@@ -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<Long, List<DueTask>> 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<Long, List<DueTask>> entry : tasksByDevice.entrySet()) {
Long deviceId = entry.getKey();
List<DueTask> 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<DeviceSwitch>()
.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<TimingCtrl> listEnabledDailyTimingCtrls() {
return timingCtrlMapper.selectList(