fix(mqtt): 修复MQTT消息处理中的设备注册查询逻辑

- 添加了通过逻辑子设备ID获取监测点信息的查询逻辑
- 实现了当设备注册信息为空时的备用查询方案
- 增加了对装置NDID和逻辑子设备ID的校验验证
- 统一了Redis缓存键值的生成方式为使用监测点ID
- 添加了相关业务异常处理以提高系统稳定性

iot-access/access-boot/src/main/java/com/njcn/access/handler/MqttMessageHandler.java

@@ -439,7 +439,22 @@ public class MqttMessageHandler {
                                 logDto.setOperate("系统端收到装置端"+nDid+"询问项目列表报文，code = " + res.getCode());
                                 logDto.setResult(1);
                                 List<RspDataDto.ProjectInfo> projectInfoList = JSON.parseArray(JSON.toJSONString(rspDataDto.getDataArray()), RspDataDto.ProjectInfo.class);
-                                String key3 = AppRedisKey.PROJECT_INFO + nDid + rspDataDto.getClDid();
+                                CsDeviceRegistry csDeviceRegistry = csDeviceRegistryFeignClient.queryByCurrentNdidAndClDid(nDid, rspDataDto.getClDid()).getData();
+                                String lineId;
+                                if (Objects.isNull(csDeviceRegistry)) {
+                                    List<CsLinePO> lines = csLineFeignClient.findByNdid(nDid).getData();
+                                    if (CollectionUtil.isEmpty(lines)) {
+                                        throw new BusinessException("通过装置NDID获取监测点信息失败");
+                                    }
+                                    CsLinePO line = lines.stream().filter(item->Objects.equals(item.getClDid(),rspDataDto.getClDid())).findFirst().orElse(null);
+                                    if (Objects.isNull(line)) {
+                                        throw new BusinessException("通过逻辑子设备ID获取监测点信息失败");
+                                    }
+                                    lineId = line.getLineId();
+                                } else {
+                                    lineId = csDeviceRegistry.getId();
+                                }
+                                String key3 = AppRedisKey.PROJECT_INFO + lineId;
                                 redisUtil.saveByKeyWithExpire(key3,projectInfoList,60L);
                                 logMessageTemplate.sendMember(logDto);
                                 break;
@@ -477,7 +492,15 @@ public class MqttMessageHandler {
         CsDeviceRegistry csDeviceRegistry = csDeviceRegistryFeignClient.queryByCurrentNdidAndClDid(nDid, item.getClDid()).getData();
         String lineId;
         if (Objects.isNull(csDeviceRegistry)) {
-            lineId = nDid + item.getClDid();
+            List<CsLinePO> lines = csLineFeignClient.findByNdid(nDid).getData();
+            if (CollectionUtil.isEmpty(lines)) {
+                throw new BusinessException("通过装置NDID获取监测点信息失败");
+            }
+            CsLinePO line = lines.stream().filter(item2->Objects.equals(item2.getClDid(),item.getClDid())).findFirst().orElse(null);
+            if (Objects.isNull(line)) {
+                throw new BusinessException("通过逻辑子设备ID获取监测点信息失败");
+            }
+            lineId = line.getLineId();
         } else {
             lineId = csDeviceRegistry.getId();
         }