diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/controller/IotController.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/controller/IotController.java index d20df30..461538b 100644 --- a/intc-modules/intc-iot/src/main/java/com/intc/iot/controller/IotController.java +++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/controller/IotController.java @@ -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 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 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 buildPendingSolarSalinityKeys(Device device) { - java.util.List keys = new java.util.ArrayList<>(); - addPendingSolarSalinityKey(keys, device.getSerialNum()); - addPendingSolarSalinityKey(keys, device.getIotId()); - return keys; - } - - private void addPendingSolarSalinityKey(java.util.List 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); - } - } /** * 修改设备输入电压类型 diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/handler/DeviceDataHandler.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/handler/DeviceDataHandler.java index 97b3590..5a19fad 100644 --- a/intc-modules/intc-iot/src/main/java/com/intc/iot/handler/DeviceDataHandler.java +++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/handler/DeviceDataHandler.java @@ -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 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 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 {