From f1ea5f4b44e6c3cd0913cb24d60a5cbff45d0bad Mon Sep 17 00:00:00 2001
From: xy <748613696@qq.com>
Date: Mon, 29 Jun 2026 20:20:34 +0800
Subject: [PATCH] =?UTF-8?q?fix(realtime):=20=E4=BF=AE=E5=A4=8D=E5=AE=9E?=
 =?UTF-8?q?=E6=97=B6=E6=95=B0=E6=8D=AE=E5=A4=84=E7=90=86=E4=B8=AD=E7=9A=84?=
 =?UTF-8?q?Redis=E7=BC=93=E5=AD=98=E5=92=8C=E8=AE=BE=E5=A4=87=E6=B3=A8?=
 =?UTF-8?q?=E5=86=8C=E9=80=BB=E8=BE=91?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- 移除MqttMessageHandler中重复的Redis缓存保存操作
- 移除RealDataConsumer中冗余的Redis缓存保存逻辑
- 在RtServiceImpl中添加必要的Redis缓存保存以确保设备响应状态同步
- 集成设备注册表查询以正确获取线路ID
- 修复APF实时数据分析中的模块工作状态键名错误
---
 .../access/handler/MqttMessageHandler.java     |  1 -
 .../njcn/rt/service/impl/RtServiceImpl.java    | 18 +++++++++++++-----
 .../message/consumer/RealDataConsumer.java     |  2 --
 3 files changed, 13 insertions(+), 8 deletions(-)

diff --git a/iot-access/access-boot/src/main/java/com/njcn/access/handler/MqttMessageHandler.java b/iot-access/access-boot/src/main/java/com/njcn/access/handler/MqttMessageHandler.java
index a96770d..5358dc1 100644
--- a/iot-access/access-boot/src/main/java/com/njcn/access/handler/MqttMessageHandler.java
+++ b/iot-access/access-boot/src/main/java/com/njcn/access/handler/MqttMessageHandler.java
@@ -612,7 +612,6 @@ public class MqttMessageHandler {
                         JSONObject jsonObject2 = JSONObject.parseObject(JSON.toJSONString(dataDto));
                         AppAutoDataMessage appAutoDataMessage = JSONObject.toJavaObject(jsonObject2, AppAutoDataMessage.class);
                         appAutoDataMessage.setId(nDid);
-                        redisUtil.saveByKeyWithExpire("devResponse:" + nDid ,200,5L);
                         rtFeignClient.analysis(appAutoDataMessage);
                         break;
                     //处理主动上送的统计数据、电度数据
diff --git a/iot-analysis/analysis-rt/rt-boot/src/main/java/com/njcn/rt/service/impl/RtServiceImpl.java b/iot-analysis/analysis-rt/rt-boot/src/main/java/com/njcn/rt/service/impl/RtServiceImpl.java
index d233f9a..a52ff2e 100644
--- a/iot-analysis/analysis-rt/rt-boot/src/main/java/com/njcn/rt/service/impl/RtServiceImpl.java
+++ b/iot-analysis/analysis-rt/rt-boot/src/main/java/com/njcn/rt/service/impl/RtServiceImpl.java
@@ -97,6 +97,7 @@ public class RtServiceImpl implements IRtService {
                 long timestamp = item.getDataTimeSec() - 8*3600;
                 baseRealDataSet.setDataTime(getTime(timestamp));
                 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")) {
             //用户Id
@@ -113,6 +114,7 @@ public class RtServiceImpl implements IRtService {
                     long timestamp = item.getDataTimeSec();
                     baseRealDataSet.setDataTime(getTime(timestamp));
                     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));
                 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
     public void apfRtAnalysis(AppAutoDataMessage appAutoDataMessage) {
         List<CsDataArray> dataArrayList;
-        String lineId;
         //监测点id
-        if (appAutoDataMessage.getDid() == 1){
-            lineId = appAutoDataMessage.getId() + "0";
+        String lineId;
+        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 {
-            lineId = appAutoDataMessage.getId() + appAutoDataMessage.getMsg().getClDid();
+            lineId = csDeviceRegistry.getId();
         }
         //获取监测点基础信息
         CsLinePO po = csLineFeignClient.getById(lineId).getData();
@@ -166,7 +174,7 @@ public class RtServiceImpl implements IRtService {
         //根据dataArray解析数据
         AppAutoDataMessage.DataArray item = appAutoDataMessage.getMsg().getDataArray().get(0);
         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);
     }
 
diff --git a/iot-message/message-boot/src/main/java/com/njcn/message/consumer/RealDataConsumer.java b/iot-message/message-boot/src/main/java/com/njcn/message/consumer/RealDataConsumer.java
index c49979a..26dbea8 100644
--- a/iot-message/message-boot/src/main/java/com/njcn/message/consumer/RealDataConsumer.java
+++ b/iot-message/message-boot/src/main/java/com/njcn/message/consumer/RealDataConsumer.java
@@ -47,8 +47,6 @@ public class RealDataConsumer extends EnhanceConsumerMessageHandler<AppAutoDataM
     @Override
     protected void handleMessage(AppAutoDataMessage appAutoDataMessage) {
         log.info("分发至实时数据");
-        String lineId = appAutoDataMessage.getId() + appAutoDataMessage.getMsg().getClDid();
-        redisUtil.saveByKeyWithExpire("devResponse:" + lineId,200,5L);
         rtFeignClient.analysis(appAutoDataMessage);
     }