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

This commit is contained in:
tianyongbao
2026-06-30 10:32:19 +08:00
parent 257f19c16f
commit f0f6494574
2 changed files with 21 additions and 82 deletions

View File

@@ -2099,6 +2099,10 @@ public class IotController extends BaseController {
}
private boolean cachePendingSolarSalinity(Device device, Double salinityCompensation) {
if (device.getSerialNum() == null || device.getSerialNum().isBlank()) {
log.warn("太阳能网控盐度缓存失败: deviceId={}, serialNum为空", device.getId());
return false;
}
try {
Map<String, Object> cacheValue = new java.util.HashMap<>();
cacheValue.put("deviceId", device.getId());
@@ -2107,20 +2111,13 @@ public class IotController extends BaseController {
cacheValue.put("salinityCompensation", salinityCompensation);
cacheValue.put("updatedAt", System.currentTimeMillis());
String cacheJson = cn.hutool.json.JSONUtil.toJsonStr(cacheValue);
java.util.List<String> cacheKeys = buildPendingSolarSalinityKeys(device);
if (cacheKeys.isEmpty()) {
log.warn("太阳能网控盐度缓存失败: deviceId={}, serialNum和iotId均为空", device.getId());
return false;
}
String cacheKey = DeviceDataHandler.PENDING_SOLAR_SALINITY_KEY_PREFIX + device.getSerialNum();
// 忽略租户前缀HTTP线程写入带租户前缀AMQP消费线程读取不带前缀需统一
final String cacheJsonRef = cacheJson;
TenantHelper.ignore(() -> {
for (String key : cacheKeys) {
com.intc.common.redis.utils.RedisUtils.setCacheObject(key, cacheJsonRef, Duration.ofDays(7));
}
});
log.info("太阳能网控盐度已缓存,等待上线补发: deviceId={}, serialNum={}, iotId={}, cacheKeys={}, salinityCompensation={}",
device.getId(), device.getSerialNum(), device.getIotId(), cacheKeys, salinityCompensation);
TenantHelper.ignore(() ->
com.intc.common.redis.utils.RedisUtils.setCacheObject(cacheKey, cacheJson, Duration.ofDays(7))
);
log.info("太阳能网控盐度已缓存,等待上线补发: deviceId={}, serialNum={}, salinityCompensation={}",
device.getId(), device.getSerialNum(), salinityCompensation);
return true;
} catch (Exception e) {
log.error("缓存太阳能网控盐度失败: deviceId={}, serialNum={}, err={}",
@@ -2146,22 +2143,6 @@ public class IotController extends BaseController {
);
}
private java.util.List<String> buildPendingSolarSalinityKeys(Device device) {
java.util.List<String> keys = new java.util.ArrayList<>();
addPendingSolarSalinityKey(keys, device.getSerialNum());
addPendingSolarSalinityKey(keys, device.getIotId());
return keys;
}
private void addPendingSolarSalinityKey(java.util.List<String> keys, String cacheName) {
if (cacheName == null || cacheName.isBlank()) {
return;
}
String key = DeviceDataHandler.PENDING_SOLAR_SALINITY_KEY_PREFIX + cacheName;
if (!keys.contains(key)) {
keys.add(key);
}
}
/**
* 修改设备输入电压类型

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;
String redisKey = PENDING_SOLAR_SALINITY_KEY_PREFIX + serialNum;
Object rawPending = RedisUtils.getCacheObject(redisKey);
if (rawPending != null) {
return new PendingSolarSalinity(redisKey, rawPending);
}
}
return null;
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 {