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

feat: 新增太阳能控制下发指令功能。

Browse Source
This commit is contained in:
tianyongbao
2026-07-01 01:02:17 +08:00
parent 573bac7505
commit 2930829ee9
4 changed files with 670 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
intc-modules/intc-fishery/src/main/java/com/intc/fishery/controller/DeviceController.java
View File

@@ -80,6 +80,16 @@ public class DeviceController extends BaseController {
return deviceService.queryPageList(bo, pageQuery);
}
/**
* 查询太阳能网控设备列表
*/
@SaCheckPermission("fishery:solarControl:list")
@GetMapping("/solar/list")
public TableDataInfo<DeviceVo> solarList(DeviceBo bo, PageQuery pageQuery) {
bo.setDeviceType(3);
return deviceService.queryPageList(bo, pageQuery);
}
/**
* 导出设备管理列表
*/

463
intc-modules/intc-iot/src/main/java/com/intc/iot/controller/IotController.java
View File

@@ -32,6 +32,7 @@ import com.intc.iot.domain.bo.AddDeviceDetectorBo;
import com.intc.iot.domain.bo.DeviceCalibrateBo;
import com.intc.iot.domain.bo.DeviceRealtimeDataBo;
import com.intc.iot.domain.bo.DeviceSalinityCompensationBo;
import com.intc.iot.domain.bo.DeviceSolarPropertyCommandBo;
import com.intc.iot.domain.bo.DeviceVoltageTypeBo;
import com.intc.iot.domain.vo.DeviceRealtimeDataVo;
import com.intc.iot.service.DeviceDataService;
@@ -66,8 +67,11 @@ import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.time.Duration;
import java.math.BigDecimal;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
/**
* 阿里云生活物联网平台(飞燕平台)控制器
@@ -82,6 +86,31 @@ import java.util.Map;
public class IotController extends BaseController {
private static final DateTimeFormatter DATETIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
private static final int DEVICE_TYPE_SOLAR_CONTROLLER = 3;
private static final int DEVICE_STATUS_ONLINE = 1;
private static final int DEVICE_STATUS_OFFLINE = 3;
private static final Set<String> SOLAR_WRITABLE_PROPERTY_KEYS = Set.of(
"correct",
"salinitySet",
"ammonia_nitrogen",
"phasedif_compensation",
"phase_compensation",
"temperature_compensation",
"stdsatphase",
"stdphaseinterval",
"SaturationPoint",
"AnaerobicPoint",
"LowtempCOEF",
"HightempCOEF",
"updatatime",
"PowerMode",
"WakeUpTime",
"BootToSleepTimeout",
"CorrectCharge",
"TempPobeOffSet1",
"TempPobeOffSet2",
"FirstBootToSleepTimeout"
);
private final DeviceTypeProperties deviceTypeProperties;
@@ -2043,7 +2072,15 @@ public class IotController extends BaseController {
boolean isSolarDevice = Integer.valueOf(3).equals(device.getDeviceType());
if (isSolarDevice) {
if (!cachePendingSolarSalinity(device, request.getSalinityCompensation())) {
if (device.getIotId() == null || device.getIotId().isBlank()) {
return R.fail("设备 iotId 为空，无法缓存盐度指令");
}
if (device.getSerialNum() == null || device.getSerialNum().isBlank()) {
return R.fail("设备编号为空，无法缓存盐度指令");
}
Map<String, Object> properties = new LinkedHashMap<>();
properties.put("salinitySet", request.getSalinityCompensation());
if (!cachePendingSolarProperties(device, properties)) {
return R.fail("盐度缓存失败，请稍后重试");
}
updateDeviceSalinityCompensation(device.getId(), request.getSalinityCompensation());
@@ -2083,7 +2120,6 @@ public class IotController extends BaseController {
}
updateDeviceSalinityCompensation(device.getId(), request.getSalinityCompensation());
clearPendingSolarSalinity(device.getSerialNum());
log.info("设置盐度补偿成功: userId={}, deviceId={}, deviceName={}, serialNum={}, salinityCompensation={}‰",
userId, device.getId(), device.getDeviceName(), device.getSerialNum(), request.getSalinityCompensation());
@@ -2103,34 +2139,6 @@ public class IotController extends BaseController {
return setResult != null && Boolean.TRUE.equals(setResult.get("success"));
}
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());
cacheValue.put("iotId", device.getIotId());
cacheValue.put("serialNum", device.getSerialNum());
cacheValue.put("salinityCompensation", salinityCompensation);
cacheValue.put("updatedAt", System.currentTimeMillis());
String cacheJson = cn.hutool.json.JSONUtil.toJsonStr(cacheValue);
String cacheKey = DeviceDataHandler.PENDING_SOLAR_SALINITY_KEY_PREFIX + device.getSerialNum();
// 忽略租户前缀：HTTP线程写入带租户前缀，AMQP消费线程读取不带前缀，需统一
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={}",
device.getId(), device.getSerialNum(), e.getMessage(), e);
return false;
}
}
private void updateDeviceSalinityCompensation(Long deviceId, Double salinityCompensation) {
deviceMapper.update(null,
new com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper<Device>()
@@ -2139,15 +2147,398 @@ public class IotController extends BaseController {
);
}
private void clearPendingSolarSalinity(String serialNum) {
if (serialNum == null || serialNum.isBlank()) {
return;
@Operation(summary = "太阳能网控属性下发")
@SaCheckPermission("fishery:solarControl:command")
@PostMapping("/device/solar/property-command")
public R<Map<String, Object>> setSolarPropertyCommand(@Validated @RequestBody DeviceSolarPropertyCommandBo request) {
try {
Long userId = LoginHelper.getUserId();
if (userId == null) {
return R.fail("未登录或登录已过期");
}
if (deviceMapper == null) {
return R.fail("设备服务未启用");
}
Device device = deviceMapper.selectOne(
new com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper<Device>()
.eq(Device::getId, request.getId())
.select(Device::getId, Device::getUserId, Device::getIotId, Device::getSerialNum,
Device::getDeviceName, Device::getDeviceType, Device::getDeadTime)
);
if (device == null) {
return R.fail("设备不存在");
}
if (!Integer.valueOf(DEVICE_TYPE_SOLAR_CONTROLLER).equals(device.getDeviceType())) {
return R.fail("仅支持太阳能网控设备");
}
if (device.getDeadTime() != null && new Date().after(device.getDeadTime())) {
return R.fail("设备已到期");
}
if (device.getIotId() == null || device.getIotId().isBlank()) {
return R.fail("设备 iotId 为空，无法下发指令");
}
if (device.getSerialNum() == null || device.getSerialNum().isBlank()) {
return R.fail("设备编号为空，无法缓存指令");
}
Map<String, Object> properties = normalizeSolarProperties(request.getProperties());
boolean cache = Boolean.TRUE.equals(request.getCache());
Integer statusCode = queryDeviceStatusCode(device.getIotId());
boolean online = Integer.valueOf(DEVICE_STATUS_ONLINE).equals(statusCode);
Map<String, Object> result = new LinkedHashMap<>();
result.put("deviceId", device.getId());
result.put("serialNum", device.getSerialNum());
result.put("online", online);
result.put("statusCode", statusCode);
result.put("properties", properties);
if (cache) {
if (online) {
return R.fail("设备当前在线，请使用直接下发");
}
if (!cachePendingSolarProperties(device, properties)) {
return R.fail("指令缓存失败，请稍后重试");
}
syncDeviceFieldsAfterSolarCommand(device.getId(), properties);
result.put("mode", "cached");
result.put("cached", true);
result.put("sent", false);
MessageOpRecordUtil.addMessageOpRecordUser(
device.getUserId(),
userId,
"设备操作",
String.format("%s(%s)，太阳能网控属性缓存下发：%s。",
device.getDeviceName(), device.getSerialNum(), properties)
);
log.info("太阳能网控属性已缓存: userId={}, deviceId={}, serialNum={}, properties={}",
userId, device.getId(), device.getSerialNum(), properties);
return R.ok("缓存成功，设备上线后自动下发", result);
}
if (!online) {
return R.fail("设备离线或未激活，请使用缓存下发");
}
if (iotCloudService == null) {
return R.fail("飞燕平台配置未启用");
}
boolean success = iotCloudService.setProperty(device.getIotId(), properties, true, 2);
if (!success) {
return R.fail("属性下发失败，请重试");
}
syncDeviceFieldsAfterSolarCommand(device.getId(), properties);
removePendingSolarProperties(device, properties.keySet());
result.put("mode", "direct");
result.put("cached", false);
result.put("sent", true);
MessageOpRecordUtil.addMessageOpRecordUser(
device.getUserId(),
userId,
"设备操作",
String.format("%s(%s)，太阳能网控属性直接下发：%s。",
device.getDeviceName(), device.getSerialNum(), properties)
);
log.info("太阳能网控属性下发成功: userId={}, deviceId={}, serialNum={}, properties={}",
userId, device.getId(), device.getSerialNum(), properties);
return R.ok("下发成功", result);
} catch (IllegalArgumentException e) {
return R.fail(e.getMessage());
} catch (Exception e) {
log.error("太阳能网控属性下发失败: {}", e.getMessage(), e);
return R.fail("太阳能网控属性下发失败: " + e.getMessage());
}
TenantHelper.ignore(() ->
com.intc.common.redis.utils.RedisUtils.deleteObject(DeviceDataHandler.PENDING_SOLAR_SALINITY_KEY_PREFIX + serialNum)
);
}
private Map<String, Object> normalizeSolarProperties(Map<String, Object> source) {
if (source == null || source.isEmpty()) {
throw new IllegalArgumentException("下发属性不能为空");
}
Map<String, Object> properties = new LinkedHashMap<>();
for (Map.Entry<String, Object> entry : source.entrySet()) {
String identifier = entry.getKey();
if (identifier == null || identifier.isBlank()) {
throw new IllegalArgumentException("属性标识符不能为空");
}
if (!SOLAR_WRITABLE_PROPERTY_KEYS.contains(identifier)) {
throw new IllegalArgumentException("属性不支持下发: " + identifier);
}
properties.put(identifier, normalizeSolarPropertyValue(identifier, entry.getValue()));
}
return properties;
}
private Object normalizeSolarPropertyValue(String identifier, Object value) {
if (value == null || (value instanceof String str && str.isBlank())) {
throw new IllegalArgumentException(identifier + " 的值不能为空");
}
return switch (identifier) {
case "correct" -> {
Integer intValue = parseInteger(value, identifier);
if (!Set.of(1, 2, 11, 12, 13, 14, 31, 32).contains(intValue)) {
throw new IllegalArgumentException("校准指令值不合法");
}
yield intValue;
}
case "PowerMode" -> requireIntegerInRange(value, identifier, 0, 2);
case "updatatime" -> requireIntegerInRange(value, identifier, 0, 65535);
case "WakeUpTime" -> requireIntegerInRange(value, identifier, 1, Integer.MAX_VALUE);
case "BootToSleepTimeout" -> requireIntegerInRange(value, identifier, 2, Integer.MAX_VALUE);
case "FirstBootToSleepTimeout" -> requireIntegerInRange(value, identifier, 3, Integer.MAX_VALUE);
case "CorrectCharge" -> normalizeCorrectCharge(value);
case "salinitySet" -> requireDecimalInRange(value, identifier, "0", "100");
case "ammonia_nitrogen" -> requireDecimalInRange(value, identifier, "0", "1000");
case "phasedif_compensation" -> requireDecimalInRange(value, identifier, "-10", "100");
case "phase_compensation" -> requireDecimalInRange(value, identifier, "0", "10");
case "temperature_compensation" -> requireDecimalInRange(value, identifier, "-10", "10");
case "stdsatphase", "stdphaseinterval" -> requireDecimalInRange(value, identifier, "0", "360");
case "SaturationPoint", "AnaerobicPoint", "LowtempCOEF", "HightempCOEF" ->
requireDecimalInRange(value, identifier, "-1000", "1000");
case "TempPobeOffSet1", "TempPobeOffSet2" -> requireDecimalInRange(value, identifier, "-12.7", "12.7");
default -> throw new IllegalArgumentException("属性不支持下发: " + identifier);
};
}
@SuppressWarnings("unchecked")
private Map<String, Object> normalizeCorrectCharge(Object value) {
if (!(value instanceof Map<?, ?> rawMap)) {
throw new IllegalArgumentException("电量校准参数格式不正确");
}
Map<String, Object> map = (Map<String, Object>) rawMap;
Map<String, Object> result = new LinkedHashMap<>();
result.put("lowvoltage", requireDecimalInRange(map.get("lowvoltage"), "lowvoltage", "0", "10"));
result.put("highvoltage", requireDecimalInRange(map.get("highvoltage"), "highvoltage", "0", "10"));
return result;
}
private Integer requireIntegerInRange(Object value, String identifier, int min, int max) {
Integer intValue = parseInteger(value, identifier);
if (intValue < min || intValue > max) {
throw new IllegalArgumentException(identifier + " 超出范围: " + min + " - " + max);
}
return intValue;
}
private Integer parseInteger(Object value, String identifier) {
try {
if (value instanceof Number number) {
double doubleValue = number.doubleValue();
if (doubleValue % 1 != 0) {
throw new IllegalArgumentException(identifier + " 必须为整数");
}
return number.intValue();
}
return Integer.valueOf(value.toString());
} catch (IllegalArgumentException e) {
throw e;
} catch (Exception e) {
throw new IllegalArgumentException(identifier + " 必须为整数");
}
}
private Double requireDecimalInRange(Object value, String identifier, String min, String max) {
BigDecimal decimal = parseDecimal(value, identifier);
BigDecimal minDecimal = new BigDecimal(min);
BigDecimal maxDecimal = new BigDecimal(max);
if (decimal.compareTo(minDecimal) < 0 || decimal.compareTo(maxDecimal) > 0) {
throw new IllegalArgumentException(identifier + " 超出范围: " + min + " - " + max);
}
return decimal.doubleValue();
}
private BigDecimal parseDecimal(Object value, String identifier) {
if (value == null || (value instanceof String str && str.isBlank())) {
throw new IllegalArgumentException(identifier + " 的值不能为空");
}
try {
return new BigDecimal(value.toString());
} catch (Exception e) {
throw new IllegalArgumentException(identifier + " 必须为数字");
}
}
private Integer queryDeviceStatusCode(String iotId) {
if (iotDeviceService == null || iotId == null || iotId.isBlank()) {
return null;
}
try {
Map<String, Object> deviceDetail = iotDeviceService.queryDeviceInfo(iotId);
return extractDeviceStatusCode(deviceDetail);
} catch (Exception e) {
log.warn("查询设备状态失败: iotId={}, err={}", iotId, e.getMessage());
return null;
}
}
private Integer extractDeviceStatusCode(Map<String, Object> deviceDetail) {
if (deviceDetail == null || deviceDetail.get("data") == null) {
return 0;
}
Object detailData = deviceDetail.get("data");
try {
java.lang.reflect.Method getStatusMethod = detailData.getClass().getMethod("getStatus");
Object statusObj = getStatusMethod.invoke(detailData);
if (statusObj == null) {
return 0;
}
String statusStr = statusObj.toString();
if ("ONLINE".equalsIgnoreCase(statusStr) || "online".equalsIgnoreCase(statusStr)) {
return DEVICE_STATUS_ONLINE;
}
if ("OFFLINE".equalsIgnoreCase(statusStr) || "offline".equalsIgnoreCase(statusStr)) {
return DEVICE_STATUS_OFFLINE;
}
if ("DISABLE".equalsIgnoreCase(statusStr) || "disable".equalsIgnoreCase(statusStr)) {
return 8;
}
return 0;
} catch (Exception ex) {
log.warn("无法从 SDK 对象中获取 status: {}", ex.getMessage());
return 0;
}
}
private boolean cachePendingSolarProperties(Device device, Map<String, Object> properties) {
try {
Map<String, Object> mergedProperties = new LinkedHashMap<>();
String cacheKey = DeviceDataHandler.PENDING_SOLAR_PROPERTY_KEY_PREFIX + device.getSerialNum();
TenantHelper.ignore(() -> {
Object rawPending = com.intc.common.redis.utils.RedisUtils.getCacheObject(cacheKey);
if (rawPending != null) {
try {
cn.hutool.json.JSONObject pending = cn.hutool.json.JSONUtil.parseObj(rawPending);
Object propertiesObj = pending.get("properties");
if (propertiesObj != null) {
cn.hutool.json.JSONObject propertyJson = cn.hutool.json.JSONUtil.parseObj(propertiesObj);
for (String key : propertyJson.keySet()) {
mergedProperties.put(key, propertyJson.get(key));
}
}
} catch (Exception e) {
log.warn("解析旧太阳能属性缓存失败，将用新值覆盖: serialNum={}, err={}",
device.getSerialNum(), e.getMessage());
}
}
});
mergedProperties.putAll(properties);
Map<String, Object> cacheValue = new LinkedHashMap<>();
cacheValue.put("deviceId", device.getId());
cacheValue.put("iotId", device.getIotId());
cacheValue.put("serialNum", device.getSerialNum());
cacheValue.put("properties", mergedProperties);
cacheValue.put("updatedAt", System.currentTimeMillis());
String cacheJson = cn.hutool.json.JSONUtil.toJsonStr(cacheValue);
TenantHelper.ignore(() ->
com.intc.common.redis.utils.RedisUtils.setCacheObject(cacheKey, cacheJson, Duration.ofDays(7))
);
return true;
} catch (Exception e) {
log.error("缓存太阳能网控属性失败: deviceId={}, serialNum={}, err={}",
device.getId(), device.getSerialNum(), e.getMessage(), e);
return false;
}
}
private void removePendingSolarProperties(Device device, Set<String> identifiers) {
if (device == null || device.getSerialNum() == null || device.getSerialNum().isBlank()
|| identifiers == null || identifiers.isEmpty()) {
return;
}
try {
String cacheKey = DeviceDataHandler.PENDING_SOLAR_PROPERTY_KEY_PREFIX + device.getSerialNum();
TenantHelper.ignore(() -> {
Object rawPending = com.intc.common.redis.utils.RedisUtils.getCacheObject(cacheKey);
if (rawPending == null) {
return;
}
cn.hutool.json.JSONObject pending = cn.hutool.json.JSONUtil.parseObj(rawPending);
Object propertiesObj = pending.get("properties");
if (propertiesObj == null) {
com.intc.common.redis.utils.RedisUtils.deleteObject(cacheKey);
return;
}
cn.hutool.json.JSONObject propertyJson = cn.hutool.json.JSONUtil.parseObj(propertiesObj);
Map<String, Object> remainingProperties = new LinkedHashMap<>();
for (String key : propertyJson.keySet()) {
if (!identifiers.contains(key)) {
remainingProperties.put(key, propertyJson.get(key));
}
}
if (remainingProperties.isEmpty()) {
com.intc.common.redis.utils.RedisUtils.deleteObject(cacheKey);
return;
}
Map<String, Object> cacheValue = new LinkedHashMap<>();
cacheValue.put("deviceId", device.getId());
cacheValue.put("iotId", device.getIotId());
cacheValue.put("serialNum", device.getSerialNum());
cacheValue.put("properties", remainingProperties);
cacheValue.put("updatedAt", System.currentTimeMillis());
com.intc.common.redis.utils.RedisUtils.setCacheObject(
cacheKey,
cn.hutool.json.JSONUtil.toJsonStr(cacheValue),
Duration.ofDays(7)
);
});
} catch (Exception e) {
log.warn("清理太阳能网控已下发属性缓存失败: serialNum={}, identifiers={}, err={}",
device.getSerialNum(), identifiers, e.getMessage());
}
}
private void syncDeviceFieldsAfterSolarCommand(Long deviceId, Map<String, Object> properties) {
if (deviceId == null || properties == null || properties.isEmpty()) {
return;
}
Device updateDevice = new Device();
updateDevice.setId(deviceId);
boolean needUpdate = false;
if (properties.containsKey("salinitySet")) {
updateDevice.setSalinityCompensation(toDouble(properties.get("salinitySet")));
needUpdate = true;
}
if (properties.containsKey("phase_compensation")) {
updateDevice.setPhaseCompensation(toDouble(properties.get("phase_compensation")));
needUpdate = true;
}
if (properties.containsKey("phasedif_compensation")) {
updateDevice.setPhasedifCompensation(toDouble(properties.get("phasedif_compensation")));
needUpdate = true;
}
if (properties.containsKey("temperature_compensation")) {
updateDevice.setTemperatureCompensation(toDouble(properties.get("temperature_compensation")));
needUpdate = true;
}
if (needUpdate) {
deviceMapper.updateById(updateDevice);
}
}
private Double toDouble(Object value) {
if (value == null) {
return null;
}
if (value instanceof Number number) {
return number.doubleValue();
}
try {
return Double.valueOf(value.toString());
} catch (Exception e) {
return null;
}
}
/**
* 修改设备输入电压类型

43
intc-modules/intc-iot/src/main/java/com/intc/iot/domain/bo/DeviceSolarPropertyCommandBo.java Normal file
View File

@@ -0,0 +1,43 @@
package com.intc.iot.domain.bo;
import io.swagger.v3.oas.annotations.media.Schema;
import jakarta.validation.constraints.NotEmpty;
import jakarta.validation.constraints.NotNull;
import lombok.Data;
import java.io.Serial;
import java.io.Serializable;
import java.util.Map;
/**
* 太阳能网控属性下发请求对象
*
* @author intc
*/
@Data
@Schema(description = "太阳能网控属性下发请求对象")
public class DeviceSolarPropertyCommandBo implements Serializable {
@Serial
private static final long serialVersionUID = 1L;
/**
* 设备ID
*/
@Schema(description = "设备ID")
@NotNull(message = "设备ID不能为空")
private Long id;
/**
* 待下发属性，key 为物模型属性标识符
*/
@Schema(description = "待下发属性")
@NotEmpty(message = "下发属性不能为空")
private Map<String, Object> properties;
/**
* 是否缓存下发。true=写入缓存，设备上线后自动补发；false=立即下发
*/
@Schema(description = "是否缓存下发")
private Boolean cache;
}

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

@@ -45,8 +45,11 @@ import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.Date;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
/**
* 设备数据处理器
@@ -59,9 +62,15 @@ import java.util.Map;
@ConditionalOnBean(IDeviceSensorDataService.class)
public class DeviceDataHandler {
public static final String PENDING_SOLAR_SALINITY_KEY_PREFIX = "device:pending:solar:salinity:";
public static final String PENDING_SOLAR_PROPERTY_KEY_PREFIX = "device:pending:solar:property:";
private static final int DEVICE_TYPE_SOLAR_CONTROLLER = 3;
/**
* 太阳能网控上线后缓存属性下发的延迟时间（毫秒）
* 设备上线后 MQTT 订阅需要一定时间完成，延迟后再下发设置避免失败
*/
private static final long SOLAR_PROPERTY_DELAY_MS = 400L;
/**
* TDengine 服务，用于存储设备时序数据
*/
@@ -127,6 +136,11 @@ public class DeviceDataHandler {
*/
private final AliyunIotProperties aliyunIotProperties;
/**
* 调度线程池，用于延迟执行任务
*/
private final ScheduledExecutorService scheduledExecutorService;
private static final DateTimeFormatter DATETIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
/**
@@ -246,7 +260,7 @@ public class DeviceDataHandler {
if (params != null) {
handleSalinitySet(deviceName, params);
}
// flushPendingSolarSalinity(deviceName, "属性上报");
// flushPendingSolarProperties(deviceName, "属性上报");
if (params != null && params.size() > 0) {
try {
@@ -960,71 +974,16 @@ public class DeviceDataHandler {
log.info("[离线告警] 设备已上线，取消离线告警: {}", deviceName);
}
// 设备上线后直接补发缓存盐度
flushPendingSolarSalinity(deviceName, "生命周期上线");
// 太阳能网控上线后，延迟下发缓存属性（设备MQTT订阅需要一定时间完成，立即下发可能失败）
scheduledExecutorService.schedule(
() -> flushPendingSolarProperties(deviceName, "生命周期上线"),
SOLAR_PROPERTY_DELAY_MS, TimeUnit.MILLISECONDS);
// 同步更新 Device.warnCode，清除离线和休眠标志位
setDeviceWarnCodeBit(deviceName, DefineDeviceWarnCode.DeviceOffline, false);
setDeviceWarnCodeBit(deviceName, DefineDeviceWarnCode.DeviceSleeping, false);
}
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 {
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");
if (salinityCompensation == null || iotId == null || deviceId == null) {
log.warn("[太阳能盐度] 缓存数据不完整，跳过补发: deviceName={}, pending={}", deviceName, pending);
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={}",
trigger, deviceName, deviceId, salinityCompensation);
return;
}
Device updateDevice = new Device();
updateDevice.setId(deviceId);
updateDevice.setSalinityCompensation(salinityCompensation);
int updated = deviceMapper.updateById(updateDevice);
if (updated > 0) {
clearPendingSolarSalinity(deviceName, device, pendingSalinity.redisKey);
log.info("[太阳能盐度] {}补发成功并清除缓存: deviceName={}, cacheKey={}, deviceId={}, salinityCompensation={}",
trigger, deviceName, pendingSalinity.redisKey, deviceId, salinityCompensation);
} else {
log.warn("[太阳能盐度] {}补发成功但设备表更新失败，保留缓存以便重试: deviceName={}, deviceId={}",
trigger, deviceName, deviceId);
}
} catch (Exception e) {
log.error("[太阳能盐度] {}补发异常: deviceName={}, err={}", trigger, deviceName, e.getMessage(), e);
}
}
private Device findDeviceByOnlineName(String deviceName) {
if (StrUtil.isBlank(deviceName)) {
return null;
@@ -1041,47 +1000,196 @@ public class DeviceDataHandler {
);
}
private PendingSolarSalinity findPendingSolarSalinity(String deviceName, Device device) {
private void flushPendingSolarProperties(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 {
PendingSolarProperty pendingProperty = findPendingSolarProperty(device);
if (pendingProperty == null) {
logNoPendingSolarProperty(trigger, deviceName, device);
return;
}
if (pendingProperty.iotId == null || pendingProperty.deviceId == null || pendingProperty.properties == null) {
log.warn("[太阳能属性] 缓存数据不完整，跳过补发: deviceName={}, cacheKey={}",
deviceName, pendingProperty.redisKey);
return;
}
if (pendingProperty.properties.isEmpty()) {
log.warn("[太阳能属性] 缓存属性为空，跳过补发: deviceName={}, cacheKey={}",
deviceName, pendingProperty.redisKey);
return;
}
log.info("[太阳能属性] 太阳能网控{}，准备补发缓存属性: deviceName={}, cacheKey={}, deviceId={}, properties={}",
trigger, deviceName, pendingProperty.redisKey, pendingProperty.deviceId, pendingProperty.properties);
boolean success = iotCloudService.setProperty(pendingProperty.iotId, pendingProperty.properties, true, 2);
if (!success) {
log.warn("[太阳能属性] {}补发失败，保留缓存: deviceName={}, deviceId={}, properties={}",
trigger, deviceName, pendingProperty.deviceId, pendingProperty.properties);
return;
}
syncDeviceFieldsAfterPropertySet(pendingProperty.deviceId, pendingProperty.properties);
clearPendingSolarProperty(pendingProperty);
log.info("[太阳能属性] {}补发成功并清除缓存: deviceName={}, cacheKey={}, deviceId={}, properties={}",
trigger, deviceName, pendingProperty.redisKey, pendingProperty.deviceId, pendingProperty.properties);
} catch (Exception e) {
log.error("[太阳能属性] {}补发异常: deviceName={}, err={}", trigger, deviceName, e.getMessage(), e);
}
}
private PendingSolarProperty findPendingSolarProperty(Device device) {
String serialNum = device.getSerialNum();
if (StrUtil.isBlank(serialNum)) {
return null;
}
// 太阳能盐度缓存为跨上下文共享数据（HTTP线程写入 + AMQP消费线程读取），
// 必须忽略租户前缀，避免写入带租户前缀而读取不带前缀导致 key 不匹配
return TenantHelper.ignore(() -> {
String redisKey = PENDING_SOLAR_SALINITY_KEY_PREFIX + serialNum;
String redisKey = PENDING_SOLAR_PROPERTY_KEY_PREFIX + serialNum;
Object rawPending = RedisUtils.getCacheObject(redisKey);
return rawPending != null ? new PendingSolarSalinity(redisKey, rawPending) : null;
if (rawPending == null) {
return null;
}
Long deviceId = device.getId();
String iotId = device.getIotId();
Map<String, Object> properties = new LinkedHashMap<>();
try {
JSONObject pending = JSONUtil.parseObj(rawPending);
Long cachedDeviceId = pending.getLong("deviceId");
String cachedIotId = pending.getStr("iotId");
if (cachedDeviceId != null) {
deviceId = cachedDeviceId;
}
if (StrUtil.isNotBlank(cachedIotId)) {
iotId = cachedIotId;
}
Object propertiesObj = pending.get("properties");
if (propertiesObj != null) {
JSONObject propertiesJson = JSONUtil.parseObj(propertiesObj);
for (String key : propertiesJson.keySet()) {
properties.put(key, propertiesJson.get(key));
}
}
} catch (Exception e) {
log.warn("[太阳能属性] 解析缓存属性失败: serialNum={}, err={}", serialNum, e.getMessage());
}
return new PendingSolarProperty(redisKey, deviceId, iotId, properties);
});
}
private boolean hasPendingSolarSalinity(String deviceName, Device device) {
return findPendingSolarSalinity(deviceName, device) != null;
private boolean hasPendingSolarProperty(Device device, String identifier) {
if (device == null || StrUtil.isBlank(identifier)) {
return false;
}
PendingSolarProperty pendingProperty = findPendingSolarProperty(device);
return pendingProperty != null
&& pendingProperty.properties != null
&& pendingProperty.properties.containsKey(identifier);
}
private void logNoPendingSolarSalinity(String trigger, String deviceName, Device device) {
private void logNoPendingSolarProperty(String trigger, String deviceName, Device device) {
if ("生命周期上线".equals(trigger)) {
log.info("[太阳能盐度] 太阳能网控{}，未找到待补发缓存: deviceName={}, deviceId={}, serialNum={}, iotId={}",
log.info("[太阳能属性] 太阳能网控{}，未找到待补发缓存: deviceName={}, deviceId={}, serialNum={}, iotId={}",
trigger, deviceName, device.getId(), device.getSerialNum(), device.getIotId());
return;
}
log.debug("[太阳能盐度] 太阳能网控{}，未找到待补发缓存: deviceName={}, deviceId={}, serialNum={}, iotId={}",
log.debug("[太阳能属性] 太阳能网控{}，未找到待补发缓存: deviceName={}, deviceId={}, serialNum={}, iotId={}",
trigger, deviceName, device.getId(), device.getSerialNum(), device.getIotId());
}
private void clearPendingSolarSalinity(String deviceName, Device device, String matchedRedisKey) {
// 忽略租户前缀，确保与写入侧 key 一致（写入在 HTTP 线程，删除在 AMQP 线程）
TenantHelper.ignore(() -> RedisUtils.deleteObject(matchedRedisKey));
private void clearPendingSolarProperty(PendingSolarProperty pendingProperty) {
if (pendingProperty == null) {
return;
}
TenantHelper.ignore(() -> {
if (StrUtil.isNotBlank(pendingProperty.redisKey)) {
RedisUtils.deleteObject(pendingProperty.redisKey);
}
});
}
private static class PendingSolarSalinity {
private final String redisKey;
private final Object rawPending;
private void syncDeviceFieldsAfterPropertySet(Long deviceId, Map<String, Object> properties) {
if (deviceId == null || properties == null || properties.isEmpty()) {
return;
}
Device updateDevice = new Device();
updateDevice.setId(deviceId);
boolean needUpdate = false;
private PendingSolarSalinity(String redisKey, Object rawPending) {
if (properties.containsKey("salinitySet")) {
Double value = toDouble(properties.get("salinitySet"));
if (value != null) {
updateDevice.setSalinityCompensation(value);
needUpdate = true;
}
}
if (properties.containsKey("phase_compensation")) {
Double value = toDouble(properties.get("phase_compensation"));
if (value != null) {
updateDevice.setPhaseCompensation(value);
needUpdate = true;
}
}
if (properties.containsKey("phasedif_compensation")) {
Double value = toDouble(properties.get("phasedif_compensation"));
if (value != null) {
updateDevice.setPhasedifCompensation(value);
needUpdate = true;
}
}
if (properties.containsKey("temperature_compensation")) {
Double value = toDouble(properties.get("temperature_compensation"));
if (value != null) {
updateDevice.setTemperatureCompensation(value);
needUpdate = true;
}
}
if (needUpdate) {
deviceMapper.updateById(updateDevice);
}
}
private Double toDouble(Object value) {
if (value == null) {
return null;
}
if (value instanceof Number number) {
return number.doubleValue();
}
try {
return Double.valueOf(value.toString());
} catch (Exception e) {
return null;
}
}
private static class PendingSolarProperty {
private final String redisKey;
private final Long deviceId;
private final String iotId;
private final Map<String, Object> properties;
private PendingSolarProperty(String redisKey, Long deviceId, String iotId, Map<String, Object> properties) {
this.redisKey = redisKey;
this.rawPending = rawPending;
this.deviceId = deviceId;
this.iotId = iotId;
this.properties = properties;
}
}
@@ -2748,15 +2856,15 @@ public class DeviceDataHandler {
Device device = deviceMapper.selectOne(
new com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper<Device>()
.eq(Device::getSerialNum, deviceName)
.select(Device::getId)
.select(Device::getId, Device::getIotId, Device::getSerialNum)
.last("LIMIT 1")
);
if (device == null) {
log.debug("设备不存在，无法更新盐度补偿设定值: {}", deviceName);
return;
}
if (hasPendingSolarSalinity(deviceName, device)) {
log.debug("[太阳能盐度] 存在待补发盐度，跳过设备旧salinitySet同步: deviceName={}, reportSalinitySet={}",
if (hasPendingSolarProperty(device, "salinitySet")) {
log.debug("[太阳能属性] 存在待补发salinitySet，跳过设备旧salinitySet同步: deviceName={}, reportSalinitySet={}",
deviceName, salinitySet);
return;
}