Microservice/iot
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(device): 重构设备监测点信息缓存逻辑并优化数据处理

Browse Source 
- 引入 DeviceMessageFeignClient 替代原有的 Redis 缓存监测点信息逻辑
- 移除 CsDeviceServiceImpl 中重复的监测点信息缓存代码，统一调用远程服务
- 将 phase 默认值从 "M" 修改为 "T"，更新所有相关数据处理逻辑
- 优化 RtServiceImpl 中 CT 和 PT 变比计算逻辑，支持双变比处理
- 更新实时数据分析中的相别标识符，从 "M" 改为 "T"
- 在 StatServiceImpl 中完善数据标签处理和质量标志设置逻辑
This commit is contained in:
xy
2026-05-20 14:16:36 +08:00
parent 6983cd39fe
commit 2eeabddf5c
6 changed files with 76 additions and 139 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
iot-analysis/analysis-stat/stat-boot/src/main/java/com/njcn/stat/service/impl/StatServiceImpl.java
View File

@@ -11,6 +11,7 @@ import com.njcn.access.utils.ChannelObjectUtil;
import com.njcn.common.pojo.exception.BusinessException;
import com.njcn.common.utils.PubUtils;
import com.njcn.csdevice.api.CsCommunicateFeignClient;
import com.njcn.csdevice.api.CsLineFeignClient;
import com.njcn.csdevice.api.DataArrayFeignClient;
import com.njcn.csdevice.api.DeviceMessageFeignClient;
import com.njcn.csdevice.pojo.dto.PqsCommunicateDto;
@@ -61,6 +62,7 @@ public class StatServiceImpl implements IStatService {
private final CsDeviceFeignClient csDeviceFeignClient;
private final DeviceMessageFeignClient deviceMessageFeignClient;
private final CsCommunicateFeignClient csCommunicateFeignClient;
private final CsLineFeignClient csLineFeignClient;
@Override
@Transactional(rollbackFor = Exception.class)
@@ -82,7 +84,7 @@ public class StatServiceImpl implements IStatService {
String lineId = null;
Object object1 = redisUtil.getObjectByKey(AppRedisKey.LINE_POSITION+appAutoDataMessage.getId());
if (Objects.isNull(object1)){
deviceMessageFeignClient.getLineInfo(appAutoDataMessage.getId());
deviceMessageFeignClient.getLineInfo(appAutoDataMessage.getId(),null);
}
//获取当前设备信息判断装置型号，来筛选监测点
List<CsEquipmentDeliveryPO> poList = channelObjectUtil.objectToList(redisUtil.getObjectByKey(AppRedisKey.DEVICE_LIST),CsEquipmentDeliveryPO.class);
@@ -109,6 +111,7 @@ public class StatServiceImpl implements IStatService {
//获取当前设备信息
if (CollectionUtil.isNotEmpty(list)) {
Map<String,String> map = new Gson().fromJson(String.valueOf(redisUtil.getObjectByKey(AppRedisKey.ELE_EPD_PQD)), Map.class);
List<String> recordList = new ArrayList<>();
for (AppAutoDataMessage.DataArray item : list) {
switch (item.getDataAttr()) {
@@ -141,7 +144,7 @@ public class StatServiceImpl implements IStatService {
} else {
dataArrayList = objectToList(object);
}
List<String> result = assembleData(lineId,dataArrayList,item,appAutoDataMessage.getMsg().getClDid(),dataArrayParam.getStatMethod(),po.getProcess(),code,po.getDevAccessMethod());
List<String> result = assembleData(lineId,dataArrayList,item,appAutoDataMessage.getMsg().getClDid(),dataArrayParam.getStatMethod(),po.getProcess(),code,po.getDevAccessMethod(),map);
recordList.addAll(result);
//获取时间
boolean timeFlag = Objects.equals(DicDataEnum.DEV_CLD.getCode(), code) && Objects.equals(po.getDevAccessMethod(), "CLD");
@@ -190,7 +193,7 @@ public class StatServiceImpl implements IStatService {
/**
* influxDB数据组装
*/
public List<String> assembleData(String lineId,List<CsDataArray> dataArrayList,AppAutoDataMessage.DataArray item,Integer clDid,String statMethod,Integer process,String devType,String accessMethod) {
public List<String> assembleData(String lineId,List<CsDataArray> dataArrayList,AppAutoDataMessage.DataArray item,Integer clDid,String statMethod,Integer process,String devType,String accessMethod, Map<String,String> map) {
List<String> records = new ArrayList<String>();
List<Float> floats = PubUtils.byteArrayToFloatList(Base64.getDecoder().decode(item.getData()));
if (CollectionUtil.isEmpty(floats)){
@@ -199,14 +202,10 @@ public class StatServiceImpl implements IStatService {
if (!Objects.equals(dataArrayList.size(),floats.size())){
throw new BusinessException(StatResponseEnum.ARRAY_DATA_NOT_MATCH);
}
Map<String,String> map = new Gson().fromJson(String.valueOf(redisUtil.getObjectByKey(AppRedisKey.ELE_EPD_PQD)), Map.class);
boolean flag = Objects.equals(DicDataEnum.DEV_CLD.getCode(), devType) && Objects.equals(accessMethod, "CLD");
//fixme 捂脸设备上送的是北京时间，时序数据库录入时 需要utc时间，减去8小时
long originalTimeSec = flag ? item.getDataTimeSec() : item.getDataTimeSec() - 8 * 3600;
if (originalTimeSec < 0) {
System.out.println("originalTimeSec==:" + originalTimeSec);
}
for (int i = 0; i < dataArrayList.size(); i++) {
String tableName = map.get(dataArrayList.get(i).getName());
@@ -227,7 +226,11 @@ public class StatServiceImpl implements IStatService {
tags.put(InfluxDBTableConstant.LINE_ID,lineId);
tags.put(InfluxDBTableConstant.PHASIC_TYPE,dataArrayList.get(i).getPhase());
tags.put(InfluxDBTableConstant.VALUE_TYPE,statMethod.toUpperCase());
tags.put(InfluxDBTableConstant.QUALITY_FLAG,"0");
if (Objects.isNull(item.getDataTag())) {
tags.put(InfluxDBTableConstant.QUALITY_FLAG,"0");
} else {
tags.put(InfluxDBTableConstant.QUALITY_FLAG,String.valueOf(item.getDataTag()));
}
Map<String,Object> fields = new HashMap<>();
//这边特殊处理，如果数据为3.14159，则将数据置为null
if (Objects.isNull(dataArrayList.get(i).getInfluxDbName())) {
@@ -237,7 +240,6 @@ public class StatServiceImpl implements IStatService {
}
fields.put(InfluxDBTableConstant.CL_DID,clDid.toString());
fields.put(InfluxDBTableConstant.PROCESS,process.toString());
fields.put(InfluxDBTableConstant.IS_ABNORMAL,item.getDataTag());
Point point = influxDbUtils.pointBuilder(tableName, adjustedTimeSec, TimeUnit.SECONDS, tags, fields);
BatchPoints batchPoints = BatchPoints.database(influxDbUtils.getDbName()).retentionPolicy("").consistency(InfluxDB.ConsistencyLevel.ALL).build();
@@ -258,4 +260,5 @@ public class StatServiceImpl implements IStatService {
}
return urlList;
}
}