fix: 控制器定时任务,同一个控制器的开关同一时间下发的话,间隔20秒下发一条指令。
This commit is contained in:
@@ -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(
|
||||
|
||||
Reference in New Issue
Block a user