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

fix: 盐度设置，功能优化，打印日志。

Browse Source
This commit is contained in:
tianyongbao
2026-06-29 10:39:50 +08:00
parent fe73e158bd
commit 4b7a1e0b58
2 changed files with 145 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<String, Object> 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<Device>()
.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<String> 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<String> 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);