qdintc/fishery-back
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: 盐度设置，设备上线后，直接下发，去掉原来的1.5秒延时。

Browse Source
This commit is contained in:
tianyongbao
2026-06-30 10:32:19 +08:00
parent 5425aba162
commit 386f571e77
2 changed files with 21 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
intc-modules/intc-iot/src/main/java/com/intc/iot/handler/DeviceDataHandler.java
View File

@@ -47,9 +47,6 @@ import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
/**
* 设备数据处理器
@@ -65,17 +62,6 @@ public class DeviceDataHandler {
public static final String PENDING_SOLAR_SALINITY_KEY_PREFIX = "device:pending:solar:salinity:";
private static final int DEVICE_TYPE_SOLAR_CONTROLLER = 3;
/**
* 太阳能盐度补发专用调度线程池（daemon线程，避免阻止JVM退出）
* 用于设备上线后延迟执行属性下发，确保设备已完成 MQTT 订阅
*/
private static final ScheduledExecutorService SOLAR_SALINITY_SCHEDULER =
Executors.newSingleThreadScheduledExecutor(r -> {
Thread t = new Thread(r, "solar-salinity-flush");
t.setDaemon(true);
return t;
});
/**
* TDengine 服务，用于存储设备时序数据
*/
@@ -260,7 +246,7 @@ public class DeviceDataHandler {
if (params != null) {
handleSalinitySet(deviceName, params);
}
flushPendingSolarSalinity(deviceName, "属性上报");
// flushPendingSolarSalinity(deviceName, "属性上报");
if (params != null && params.size() > 0) {
try {
@@ -974,12 +960,8 @@ public class DeviceDataHandler {
log.info("[离线告警] 设备已上线，取消离线告警: {}", deviceName);
}
// 异步延迟 1.5 秒补发缓存盐度：设备刚上线时尚未完成 MQTT 订阅，
// 立即下发属性会失败，等待 1.5 秒让设备完成订阅后再下发
SOLAR_SALINITY_SCHEDULER.schedule(
() -> flushPendingSolarSalinity(deviceName, "生命周期上线"),
1500L, TimeUnit.MILLISECONDS
);
// 设备上线后直接补发缓存盐度
flushPendingSolarSalinity(deviceName, "生命周期上线");
// 同步更新 Device.warnCode，清除离线和休眠标志位
setDeviceWarnCodeBit(deviceName, DefineDeviceWarnCode.DeviceOffline, false);
@@ -1060,22 +1042,17 @@ public class DeviceDataHandler {
}
private PendingSolarSalinity findPendingSolarSalinity(String deviceName, Device device) {
List<String> cacheNames = new java.util.ArrayList<>();
addCacheName(cacheNames, deviceName);
addCacheName(cacheNames, device.getSerialNum());
addCacheName(cacheNames, device.getIotId());
String serialNum = device.getSerialNum();
if (StrUtil.isBlank(serialNum)) {
return null;
}
// 太阳能盐度缓存为跨上下文共享数据（HTTP线程写入 + AMQP消费线程读取），
// 必须忽略租户前缀，避免写入带租户前缀而读取不带前缀导致 key 不匹配
return TenantHelper.ignore(() -> {
for (String cacheName : cacheNames) {
String redisKey = PENDING_SOLAR_SALINITY_KEY_PREFIX + cacheName;
Object rawPending = RedisUtils.getCacheObject(redisKey);
if (rawPending != null) {
return new PendingSolarSalinity(redisKey, rawPending);
}
}
return null;
String redisKey = PENDING_SOLAR_SALINITY_KEY_PREFIX + serialNum;
Object rawPending = RedisUtils.getCacheObject(redisKey);
return rawPending != null ? new PendingSolarSalinity(redisKey, rawPending) : null;
});
}
@@ -1093,28 +1070,9 @@ public class DeviceDataHandler {
trigger, deviceName, device.getId(), device.getSerialNum(), device.getIotId());
}
private void addCacheName(List<String> cacheNames, String cacheName) {
if (StrUtil.isBlank(cacheName) || cacheNames.contains(cacheName)) {
return;
}
cacheNames.add(cacheName);
}
private void clearPendingSolarSalinity(String deviceName, Device device, String matchedRedisKey) {
// 忽略租户前缀，确保与写入侧 key 一致（写入在 HTTP 线程，删除在 AMQP 线程）
TenantHelper.ignore(() -> {
RedisUtils.deleteObject(matchedRedisKey);
deletePendingSolarSalinityByName(deviceName);
deletePendingSolarSalinityByName(device.getSerialNum());
deletePendingSolarSalinityByName(device.getIotId());
});
}
private void deletePendingSolarSalinityByName(String cacheName) {
if (StrUtil.isBlank(cacheName)) {
return;
}
RedisUtils.deleteObject(PENDING_SOLAR_SALINITY_KEY_PREFIX + cacheName);
TenantHelper.ignore(() -> RedisUtils.deleteObject(matchedRedisKey));
}
private static class PendingSolarSalinity {