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 c2bd2f9..a48f33f 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 @@ -2100,16 +2100,23 @@ public class IotController extends BaseController { private boolean cachePendingSolarSalinity(Device device, Double salinityCompensation) { try { - String key = DeviceDataHandler.PENDING_SOLAR_SALINITY_KEY_PREFIX + device.getSerialNum(); Map cacheValue = new java.util.HashMap<>(); cacheValue.put("deviceId", device.getId()); cacheValue.put("iotId", device.getIotId()); cacheValue.put("serialNum", device.getSerialNum()); cacheValue.put("salinityCompensation", salinityCompensation); cacheValue.put("updatedAt", System.currentTimeMillis()); - com.intc.common.redis.utils.RedisUtils.setCacheObject(key, cn.hutool.json.JSONUtil.toJsonStr(cacheValue), Duration.ofDays(7)); - log.info("太阳能网控盐度已缓存,等待上线补发: deviceId={}, serialNum={}, salinityCompensation={}", - device.getId(), device.getSerialNum(), salinityCompensation); + 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; + } + for (String key : cacheKeys) { + com.intc.common.redis.utils.RedisUtils.setCacheObject(key, cacheJson, Duration.ofDays(7)); + } + log.info("太阳能网控盐度已缓存,等待上线补发: deviceId={}, serialNum={}, iotId={}, cacheKeys={}, salinityCompensation={}", + device.getId(), device.getSerialNum(), device.getIotId(), cacheKeys, salinityCompensation); return true; } catch (Exception e) { log.error("缓存太阳能网控盐度失败: deviceId={}, serialNum={}, err={}", @@ -2133,6 +2140,23 @@ public class IotController extends BaseController { com.intc.common.redis.utils.RedisUtils.deleteObject(DeviceDataHandler.PENDING_SOLAR_SALINITY_KEY_PREFIX + serialNum); } + 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 7e9c898..9370aa8 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 @@ -59,6 +59,7 @@ import java.util.Map; 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; /** * TDengine 服务,用于存储设备时序数据 @@ -240,6 +241,12 @@ public class DeviceDataHandler { handleSensorErrorCode(deviceName, params); } + // 太阳能网控在线窗口很短,先处理盐度设定值,再尽快补发缓存盐度。 +// if (params != null) { +// handleSalinitySet(deviceName, params); +// } +// flushPendingSolarSalinity(deviceName, "属性上报"); + if (params != null && params.size() > 0) { try { // 如果是控制器,需要区分处理 @@ -298,11 +305,6 @@ public class DeviceDataHandler { } } - // 处理 salinitySet(盐度补偿设定值):同步到设备表,不入 TDengine - if (params != null) { - handleSalinitySet(deviceName, params); - } - // 检查是否触发报警 if (params != null && sensorData != null) { checkAndTriggerAlarm(productKey, deviceName, sensorData); @@ -957,22 +959,33 @@ public class DeviceDataHandler { log.info("[离线告警] 设备已上线,取消离线告警: {}", deviceName); } - flushPendingSolarSalinity(deviceName); + flushPendingSolarSalinity(deviceName, "生命周期上线"); // 同步更新 Device.warnCode,清除离线和休眠标志位 setDeviceWarnCodeBit(deviceName, DefineDeviceWarnCode.DeviceOffline, false); setDeviceWarnCodeBit(deviceName, DefineDeviceWarnCode.DeviceSleeping, false); } - private void flushPendingSolarSalinity(String deviceName) { - String redisKey = PENDING_SOLAR_SALINITY_KEY_PREFIX + deviceName; - Object rawPending = RedisUtils.getCacheObject(redisKey); - if (rawPending == null) { + private void flushPendingSolarSalinity(String deviceName, String trigger) { + Device device = findDeviceByOnlineName(deviceName); + if (device == null) { + log.debug("[太阳能盐度] 上线设备未绑定,跳过补发: deviceName={}", deviceName); + return; + } + if (!Integer.valueOf(DEVICE_TYPE_SOLAR_CONTROLLER).equals(device.getDeviceType())) { + log.debug("[太阳能盐度] 非太阳能网控设备,跳过补发: deviceName={}, deviceId={}, deviceType={}", + deviceName, device.getId(), device.getDeviceType()); return; } try { - JSONObject pending = JSONUtil.parseObj(rawPending); + PendingSolarSalinity pendingSalinity = findPendingSolarSalinity(deviceName, device); + if (pendingSalinity == null) { + logNoPendingSolarSalinity(trigger, deviceName, device); + return; + } + + JSONObject pending = JSONUtil.parseObj(pendingSalinity.rawPending); Double salinityCompensation = pending.getDouble("salinityCompensation"); String iotId = pending.getStr("iotId"); Long deviceId = pending.getLong("deviceId"); @@ -981,12 +994,15 @@ public class DeviceDataHandler { return; } + log.info("[太阳能盐度] 太阳能网控{},准备补发缓存盐度: deviceName={}, cacheKey={}, deviceId={}, salinityCompensation={}", + trigger, deviceName, pendingSalinity.redisKey, deviceId, salinityCompensation); + Map properties = new HashMap<>(); properties.put("salinitySet", salinityCompensation); boolean success = iotCloudService.setProperty(iotId, properties, true, 2); if (!success) { - log.warn("[太阳能盐度] 上线补发失败,保留缓存: deviceName={}, deviceId={}, salinityCompensation={}", - deviceName, deviceId, salinityCompensation); + log.warn("[太阳能盐度] {}补发失败,保留缓存: deviceName={}, deviceId={}, salinityCompensation={}", + trigger, deviceName, deviceId, salinityCompensation); return; } @@ -995,15 +1011,92 @@ public class DeviceDataHandler { updateDevice.setSalinityCompensation(salinityCompensation); int updated = deviceMapper.updateById(updateDevice); if (updated > 0) { - RedisUtils.deleteObject(redisKey); - log.info("[太阳能盐度] 上线补发成功并清除缓存: deviceName={}, deviceId={}, salinityCompensation={}", - deviceName, deviceId, salinityCompensation); + clearPendingSolarSalinity(deviceName, device, pendingSalinity.redisKey); + log.info("[太阳能盐度] {}补发成功并清除缓存: deviceName={}, cacheKey={}, deviceId={}, salinityCompensation={}", + trigger, deviceName, pendingSalinity.redisKey, deviceId, salinityCompensation); } else { - log.warn("[太阳能盐度] 上线补发成功但设备表更新失败,保留缓存以便重试: deviceName={}, deviceId={}", - deviceName, deviceId); + log.warn("[太阳能盐度] {}补发成功但设备表更新失败,保留缓存以便重试: deviceName={}, deviceId={}", + trigger, deviceName, deviceId); } } catch (Exception e) { - log.error("[太阳能盐度] 上线补发异常: deviceName={}, err={}", deviceName, e.getMessage(), e); + log.error("[太阳能盐度] {}补发异常: deviceName={}, err={}", trigger, deviceName, e.getMessage(), e); + } + } + + private Device findDeviceByOnlineName(String deviceName) { + if (StrUtil.isBlank(deviceName)) { + return null; + } + return deviceMapper.selectOne( + new com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper() + .and(wrapper -> wrapper + .eq(Device::getSerialNum, deviceName) + .or() + .eq(Device::getIotId, deviceName) + ) + .select(Device::getId, Device::getIotId, Device::getSerialNum, Device::getDeviceType) + .last("LIMIT 1") + ); + } + + private PendingSolarSalinity findPendingSolarSalinity(String deviceName, Device device) { + List cacheNames = new java.util.ArrayList<>(); + addCacheName(cacheNames, deviceName); + addCacheName(cacheNames, device.getSerialNum()); + addCacheName(cacheNames, device.getIotId()); + + 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; + } + + private boolean hasPendingSolarSalinity(String deviceName, Device device) { + return findPendingSolarSalinity(deviceName, device) != null; + } + + private void logNoPendingSolarSalinity(String trigger, String deviceName, Device device) { + if ("生命周期上线".equals(trigger)) { + log.info("[太阳能盐度] 太阳能网控{},未找到待补发缓存: deviceName={}, deviceId={}, serialNum={}, iotId={}", + trigger, deviceName, device.getId(), device.getSerialNum(), device.getIotId()); + return; + } + log.debug("[太阳能盐度] 太阳能网控{},未找到待补发缓存: deviceName={}, deviceId={}, serialNum={}, iotId={}", + 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) { + 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 final String redisKey; + private final Object rawPending; + + private PendingSolarSalinity(String redisKey, Object rawPending) { + this.redisKey = redisKey; + this.rawPending = rawPending; } } @@ -2677,6 +2770,11 @@ public class DeviceDataHandler { log.debug("设备不存在,无法更新盐度补偿设定值: {}", deviceName); return; } + if (hasPendingSolarSalinity(deviceName, device)) { + log.debug("[太阳能盐度] 存在待补发盐度,跳过设备旧salinitySet同步: deviceName={}, reportSalinitySet={}", + deviceName, salinitySet); + return; + } Device updateDevice = new Device(); updateDevice.setId(device.getId()); updateDevice.setSalinityCompensation(salinitySet);