fix(realtime): 修复实时数据处理中的Redis缓存和设备注册逻辑

- 移除MqttMessageHandler中重复的Redis缓存保存操作
- 移除RealDataConsumer中冗余的Redis缓存保存逻辑
- 在RtServiceImpl中添加必要的Redis缓存保存以确保设备响应状态同步
- 集成设备注册表查询以正确获取线路ID
- 修复APF实时数据分析中的模块工作状态键名错误
This commit is contained in:
xy
2026-06-29 20:20:34 +08:00
parent 47d58e589b
commit f1ea5f4b44
3 changed files with 13 additions and 8 deletions

View File

@@ -612,7 +612,6 @@ public class MqttMessageHandler {
JSONObject jsonObject2 = JSONObject.parseObject(JSON.toJSONString(dataDto)); JSONObject jsonObject2 = JSONObject.parseObject(JSON.toJSONString(dataDto));
AppAutoDataMessage appAutoDataMessage = JSONObject.toJavaObject(jsonObject2, AppAutoDataMessage.class); AppAutoDataMessage appAutoDataMessage = JSONObject.toJavaObject(jsonObject2, AppAutoDataMessage.class);
appAutoDataMessage.setId(nDid); appAutoDataMessage.setId(nDid);
redisUtil.saveByKeyWithExpire("devResponse:" + nDid ,200,5L);
rtFeignClient.analysis(appAutoDataMessage); rtFeignClient.analysis(appAutoDataMessage);
break; break;
//处理主动上送的统计数据、电度数据 //处理主动上送的统计数据、电度数据

View File

@@ -97,6 +97,7 @@ public class RtServiceImpl implements IRtService {
long timestamp = item.getDataTimeSec() - 8*3600; long timestamp = item.getDataTimeSec() - 8*3600;
baseRealDataSet.setDataTime(getTime(timestamp)); baseRealDataSet.setDataTime(getTime(timestamp));
publisher.send("/Web/RealData/" + lineId, new Gson().toJson(baseRealDataSet), 1, false); publisher.send("/Web/RealData/" + lineId, new Gson().toJson(baseRealDataSet), 1, false);
redisUtil.saveByKeyWithExpire("devResponse:" + po1.getNdid() ,200,5L);
} }
} else if (dataSet.getName().contains("实时数据") || dataSet.getName().contains("Ds$Pqd$Rt$01")) { } else if (dataSet.getName().contains("实时数据") || dataSet.getName().contains("Ds$Pqd$Rt$01")) {
//用户Id //用户Id
@@ -113,6 +114,7 @@ public class RtServiceImpl implements IRtService {
long timestamp = item.getDataTimeSec(); long timestamp = item.getDataTimeSec();
baseRealDataSet.setDataTime(getTime(timestamp)); baseRealDataSet.setDataTime(getTime(timestamp));
publisher.send("/Web/RealData/" + lineId, new Gson().toJson(baseRealDataSet), 1, false); publisher.send("/Web/RealData/" + lineId, new Gson().toJson(baseRealDataSet), 1, false);
redisUtil.saveByKeyWithExpire("devResponse:" + lineId,200,5L);
}); });
} }
} }
@@ -136,6 +138,7 @@ public class RtServiceImpl implements IRtService {
} }
harmRealDataSet.setDataTime(getTime(timestamp)); harmRealDataSet.setDataTime(getTime(timestamp));
publisher.send("/Web/RealData/" + lineId, new Gson().toJson(harmRealDataSet), 1, false); publisher.send("/Web/RealData/" + lineId, new Gson().toJson(harmRealDataSet), 1, false);
redisUtil.saveByKeyWithExpire("devResponse:" + lineId,200,5L);
} }
} }
} }
@@ -143,12 +146,17 @@ public class RtServiceImpl implements IRtService {
@Override @Override
public void apfRtAnalysis(AppAutoDataMessage appAutoDataMessage) { public void apfRtAnalysis(AppAutoDataMessage appAutoDataMessage) {
List<CsDataArray> dataArrayList; List<CsDataArray> dataArrayList;
String lineId;
//监测点id //监测点id
if (appAutoDataMessage.getDid() == 1){ String lineId;
lineId = appAutoDataMessage.getId() + "0"; CsDeviceRegistry csDeviceRegistry = csDeviceRegistryFeignClient.queryByCurrentNdidAndClDid(appAutoDataMessage.getId(),appAutoDataMessage.getMsg().getClDid()).getData();
if (Objects.isNull(csDeviceRegistry)) {
if (appAutoDataMessage.getDid() == 1){
lineId = appAutoDataMessage.getId() + "0";
} else {
lineId = appAutoDataMessage.getId() + appAutoDataMessage.getMsg().getClDid();
}
} else { } else {
lineId = appAutoDataMessage.getId() + appAutoDataMessage.getMsg().getClDid(); lineId = csDeviceRegistry.getId();
} }
//获取监测点基础信息 //获取监测点基础信息
CsLinePO po = csLineFeignClient.getById(lineId).getData(); CsLinePO po = csLineFeignClient.getById(lineId).getData();
@@ -166,7 +174,7 @@ public class RtServiceImpl implements IRtService {
//根据dataArray解析数据 //根据dataArray解析数据
AppAutoDataMessage.DataArray item = appAutoDataMessage.getMsg().getDataArray().get(0); AppAutoDataMessage.DataArray item = appAutoDataMessage.getMsg().getDataArray().get(0);
Map<String,Float> map = getData(dataArrayList,item); Map<String,Float> map = getData(dataArrayList,item);
int data = Math.round(map.get("Apf_ModWorkingSts" + "M")); int data = Math.round(map.get("Apf_ModWorkingSts" + "T"));
redisUtil.saveByKeyWithExpire("ApfRtData:" + appAutoDataMessage.getMid(),data,10L); redisUtil.saveByKeyWithExpire("ApfRtData:" + appAutoDataMessage.getMid(),data,10L);
} }

View File

@@ -47,8 +47,6 @@ public class RealDataConsumer extends EnhanceConsumerMessageHandler<AppAutoDataM
@Override @Override
protected void handleMessage(AppAutoDataMessage appAutoDataMessage) { protected void handleMessage(AppAutoDataMessage appAutoDataMessage) {
log.info("分发至实时数据"); log.info("分发至实时数据");
String lineId = appAutoDataMessage.getId() + appAutoDataMessage.getMsg().getClDid();
redisUtil.saveByKeyWithExpire("devResponse:" + lineId,200,5L);
rtFeignClient.analysis(appAutoDataMessage); rtFeignClient.analysis(appAutoDataMessage);
} }