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

This commit is contained in:
tianyongbao
2026-06-30 10:32:19 +08:00
parent 9f9c4fc4a2
commit fa2cc8e77a
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) { private boolean cachePendingSolarSalinity(Device device, Double salinityCompensation) {
if (device.getSerialNum() == null || device.getSerialNum().isBlank()) {
log.warn("太阳能网控盐度缓存失败: deviceId={}, serialNum为空", device.getId());
return false;
}
try { try {
Map<String, Object> cacheValue = new java.util.HashMap<>(); Map<String, Object> cacheValue = new java.util.HashMap<>();
cacheValue.put("deviceId", device.getId()); cacheValue.put("deviceId", device.getId());
@@ -2107,20 +2111,13 @@ public class IotController extends BaseController {
cacheValue.put("salinityCompensation", salinityCompensation); cacheValue.put("salinityCompensation", salinityCompensation);
cacheValue.put("updatedAt", System.currentTimeMillis()); cacheValue.put("updatedAt", System.currentTimeMillis());
String cacheJson = cn.hutool.json.JSONUtil.toJsonStr(cacheValue); String cacheJson = cn.hutool.json.JSONUtil.toJsonStr(cacheValue);
java.util.List<String> cacheKeys = buildPendingSolarSalinityKeys(device); String cacheKey = DeviceDataHandler.PENDING_SOLAR_SALINITY_KEY_PREFIX + device.getSerialNum();
if (cacheKeys.isEmpty()) {
log.warn("太阳能网控盐度缓存失败: deviceId={}, serialNum和iotId均为空", device.getId());
return false;
}
// 忽略租户前缀HTTP线程写入带租户前缀AMQP消费线程读取不带前缀需统一 // 忽略租户前缀HTTP线程写入带租户前缀AMQP消费线程读取不带前缀需统一
final String cacheJsonRef = cacheJson; TenantHelper.ignore(() ->
TenantHelper.ignore(() -> { com.intc.common.redis.utils.RedisUtils.setCacheObject(cacheKey, cacheJson, Duration.ofDays(7))
for (String key : cacheKeys) { );
com.intc.common.redis.utils.RedisUtils.setCacheObject(key, cacheJsonRef, Duration.ofDays(7)); log.info("太阳能网控盐度已缓存,等待上线补发: deviceId={}, serialNum={}, salinityCompensation={}",
} device.getId(), device.getSerialNum(), salinityCompensation);
});
log.info("太阳能网控盐度已缓存,等待上线补发: deviceId={}, serialNum={}, iotId={}, cacheKeys={}, salinityCompensation={}",
device.getId(), device.getSerialNum(), device.getIotId(), cacheKeys, salinityCompensation);
return true; return true;
} catch (Exception e) { } catch (Exception e) {
log.error("缓存太阳能网控盐度失败: deviceId={}, serialNum={}, err={}", 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.HashMap;
import java.util.List; import java.util.List;
import java.util.Map; 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:"; public static final String PENDING_SOLAR_SALINITY_KEY_PREFIX = "device:pending:solar:salinity:";
private static final int DEVICE_TYPE_SOLAR_CONTROLLER = 3; 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 服务,用于存储设备时序数据 * TDengine 服务,用于存储设备时序数据
*/ */
@@ -260,7 +246,7 @@ public class DeviceDataHandler {
if (params != null) { if (params != null) {
handleSalinitySet(deviceName, params); handleSalinitySet(deviceName, params);
} }
flushPendingSolarSalinity(deviceName, "属性上报"); // flushPendingSolarSalinity(deviceName, "属性上报");
if (params != null && params.size() > 0) { if (params != null && params.size() > 0) {
try { try {
@@ -974,12 +960,8 @@ public class DeviceDataHandler {
log.info("[离线告警] 设备已上线,取消离线告警: {}", deviceName); log.info("[离线告警] 设备已上线,取消离线告警: {}", deviceName);
} }
// 异步延迟 1.5 秒补发缓存盐度:设备刚上线时尚未完成 MQTT 订阅, // 设备上线后直接补发缓存盐度
// 立即下发属性会失败,等待 1.5 秒让设备完成订阅后再下发 flushPendingSolarSalinity(deviceName, "生命周期上线");
SOLAR_SALINITY_SCHEDULER.schedule(
() -> flushPendingSolarSalinity(deviceName, "生命周期上线"),
1500L, TimeUnit.MILLISECONDS
);
// 同步更新 Device.warnCode清除离线和休眠标志位 // 同步更新 Device.warnCode清除离线和休眠标志位
setDeviceWarnCodeBit(deviceName, DefineDeviceWarnCode.DeviceOffline, false); setDeviceWarnCodeBit(deviceName, DefineDeviceWarnCode.DeviceOffline, false);
@@ -1060,22 +1042,17 @@ public class DeviceDataHandler {
} }
private PendingSolarSalinity findPendingSolarSalinity(String deviceName, Device device) { private PendingSolarSalinity findPendingSolarSalinity(String deviceName, Device device) {
List<String> cacheNames = new java.util.ArrayList<>(); String serialNum = device.getSerialNum();
addCacheName(cacheNames, deviceName); if (StrUtil.isBlank(serialNum)) {
addCacheName(cacheNames, device.getSerialNum()); return null;
addCacheName(cacheNames, device.getIotId()); }
// 太阳能盐度缓存为跨上下文共享数据HTTP线程写入 + AMQP消费线程读取 // 太阳能盐度缓存为跨上下文共享数据HTTP线程写入 + AMQP消费线程读取
// 必须忽略租户前缀,避免写入带租户前缀而读取不带前缀导致 key 不匹配 // 必须忽略租户前缀,避免写入带租户前缀而读取不带前缀导致 key 不匹配
return TenantHelper.ignore(() -> { return TenantHelper.ignore(() -> {
for (String cacheName : cacheNames) { String redisKey = PENDING_SOLAR_SALINITY_KEY_PREFIX + serialNum;
String redisKey = PENDING_SOLAR_SALINITY_KEY_PREFIX + cacheName; Object rawPending = RedisUtils.getCacheObject(redisKey);
Object rawPending = RedisUtils.getCacheObject(redisKey); return rawPending != null ? new PendingSolarSalinity(redisKey, rawPending) : null;
if (rawPending != null) {
return new PendingSolarSalinity(redisKey, rawPending);
}
}
return null;
}); });
} }
@@ -1093,28 +1070,9 @@ public class DeviceDataHandler {
trigger, deviceName, device.getId(), device.getSerialNum(), device.getIotId()); 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) { private void clearPendingSolarSalinity(String deviceName, Device device, String matchedRedisKey) {
// 忽略租户前缀,确保与写入侧 key 一致(写入在 HTTP 线程,删除在 AMQP 线程) // 忽略租户前缀,确保与写入侧 key 一致(写入在 HTTP 线程,删除在 AMQP 线程)
TenantHelper.ignore(() -> { TenantHelper.ignore(() -> RedisUtils.deleteObject(matchedRedisKey));
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);
} }
private static class PendingSolarSalinity { private static class PendingSolarSalinity {