From 6983cd39fe165739eb1264da8222d865fccd8f02 Mon Sep 17 00:00:00 2001 From: xy <748613696@qq.com> Date: Mon, 18 May 2026 18:36:42 +0800 Subject: [PATCH] =?UTF-8?q?fix(data):=20=E8=A7=A3=E5=86=B3=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?=E7=A9=BA=E6=8C=87=E9=92=88=E5=BC=82=E5=B8=B8=E5=92=8C=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E5=86=99=E5=85=A5=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 在基础数据处理中添加redis对象空值检查，避免空指针异常 - 在谐波数据处理中添加redis对象空值检查，确保数据安全处理 - 将统计方法名称转换为大写以确保数据一致性 - 根据influxDbName字段决定数据字段名称，提高数据映射准确性 - 移除已注释的废弃代码块，清理代码结构 --- .../njcn/rt/service/impl/RtServiceImpl.java | 52 +++++++++++-------- .../stat/service/impl/StatServiceImpl.java | 49 +++-------------- 2 files changed, 37 insertions(+), 64 deletions(-) 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 d3c4fc7..8329a88 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 @@ -82,16 +82,19 @@ public class RtServiceImpl implements IRtService { //基础数据 if (dataSet.getName().contains("Ds$Pqd$Rt$Basic$")) { //用户Id - String userId = redisUtil.getObjectByKey("rtDataUserId:"+lineId).toString(); - BaseRealDataSet baseRealDataSet = assembleData(dataArrayList,item,po.getConType(),po1.getDevAccessMethod()); - baseRealDataSet.setUserId(userId); - baseRealDataSet.setLineId(lineId); - baseRealDataSet.setPt(po.getPtRatio().floatValue()); - baseRealDataSet.setCt(po.getCtRatio().floatValue()); - baseRealDataSet.setDataLevel(dataSet.getDataLevel()); - long timestamp = item.getDataTimeSec() - 8*3600; - baseRealDataSet.setDataTime(getTime(timestamp)); - publisher.send("/Web/RealData/" + lineId, new Gson().toJson(baseRealDataSet), 1, false); + Object redisObject = redisUtil.getObjectByKey("rtDataUserId:"+lineId); + if (ObjectUtil.isNotNull(redisObject)) { + String userId = redisObject.toString(); + BaseRealDataSet baseRealDataSet = assembleData(dataArrayList,item,po.getConType(),po1.getDevAccessMethod()); + baseRealDataSet.setUserId(userId); + baseRealDataSet.setLineId(lineId); + baseRealDataSet.setPt(po.getPtRatio().floatValue()); + baseRealDataSet.setCt(po.getCtRatio().floatValue()); + baseRealDataSet.setDataLevel(dataSet.getDataLevel()); + long timestamp = item.getDataTimeSec() - 8*3600; + baseRealDataSet.setDataTime(getTime(timestamp)); + publisher.send("/Web/RealData/" + lineId, new Gson().toJson(baseRealDataSet), 1, false); + } } else if (dataSet.getName().contains("实时数据") || dataSet.getName().contains("Ds$Pqd$Rt$01")) { //用户Id Object redisObject = redisUtil.getObjectByKey("rtDataUserId:"+lineId); @@ -114,20 +117,23 @@ public class RtServiceImpl implements IRtService { else { long timestamp; //用户Id - String userId = redisUtil.getObjectByKey("rtDataUserId:"+lineId).toString(); - HarmRealDataSet harmRealDataSet = harmData(dataArrayList,item,dataSet.getDataLevel(),po.getCtRatio()); - harmRealDataSet.setUserId(userId); - harmRealDataSet.setLineId(lineId); - harmRealDataSet.setPt(po.getPtRatio().floatValue()); - harmRealDataSet.setCt(po.getCtRatio().floatValue()); - harmRealDataSet.setDataLevel(dataSet.getDataLevel()); - if (ObjectUtil.isNotNull(po.getLineNo())) { - timestamp = item.getDataTimeSec(); - } else { - timestamp = item.getDataTimeSec() - 8*3600; + Object redisObject = redisUtil.getObjectByKey("rtDataUserId:"+lineId); + if (ObjectUtil.isNotNull(redisObject)) { + String userId = redisObject.toString(); + HarmRealDataSet harmRealDataSet = harmData(dataArrayList,item,dataSet.getDataLevel(),po.getCtRatio()); + harmRealDataSet.setUserId(userId); + harmRealDataSet.setLineId(lineId); + harmRealDataSet.setPt(po.getPtRatio().floatValue()); + harmRealDataSet.setCt(po.getCtRatio().floatValue()); + harmRealDataSet.setDataLevel(dataSet.getDataLevel()); + if (ObjectUtil.isNotNull(po.getLineNo())) { + timestamp = item.getDataTimeSec(); + } else { + timestamp = item.getDataTimeSec() - 8*3600; + } + harmRealDataSet.setDataTime(getTime(timestamp)); + publisher.send("/Web/RealData/" + lineId, new Gson().toJson(harmRealDataSet), 1, false); } - harmRealDataSet.setDataTime(getTime(timestamp)); - publisher.send("/Web/RealData/" + lineId, new Gson().toJson(harmRealDataSet), 1, false); } } diff --git a/iot-analysis/analysis-stat/stat-boot/src/main/java/com/njcn/stat/service/impl/StatServiceImpl.java b/iot-analysis/analysis-stat/stat-boot/src/main/java/com/njcn/stat/service/impl/StatServiceImpl.java index ca0ffe9..4db2b56 100644 --- a/iot-analysis/analysis-stat/stat-boot/src/main/java/com/njcn/stat/service/impl/StatServiceImpl.java +++ b/iot-analysis/analysis-stat/stat-boot/src/main/java/com/njcn/stat/service/impl/StatServiceImpl.java @@ -226,13 +226,17 @@ public class StatServiceImpl implements IStatService { Map tags = new HashMap<>(); tags.put(InfluxDBTableConstant.LINE_ID,lineId); tags.put(InfluxDBTableConstant.PHASIC_TYPE,dataArrayList.get(i).getPhase()); - tags.put(InfluxDBTableConstant.VALUE_TYPE,statMethod); - tags.put(InfluxDBTableConstant.CL_DID,clDid.toString()); - tags.put(InfluxDBTableConstant.PROCESS,process.toString()); + tags.put(InfluxDBTableConstant.VALUE_TYPE,statMethod.toUpperCase()); tags.put(InfluxDBTableConstant.QUALITY_FLAG,"0"); Map fields = new HashMap<>(); //这边特殊处理，如果数据为3.14159，则将数据置为null - fields.put(dataArrayList.get(i).getName(),Objects.equals(floats.get(i),3.14159f) ? null:floats.get(i)); + if (Objects.isNull(dataArrayList.get(i).getInfluxDbName())) { + fields.put(dataArrayList.get(i).getName(),Objects.equals(floats.get(i),3.14159f) ? null:floats.get(i)); + } else { + fields.put(dataArrayList.get(i).getInfluxDbName(),Objects.equals(floats.get(i),3.14159f) ? null:floats.get(i)); + } + 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); @@ -243,43 +247,6 @@ public class StatServiceImpl implements IStatService { return records; } -// public List assembleData(String lineId,List dataArrayList,AppAutoDataMessage.DataArray item,Integer clDid,String statMethod,Integer process,String devType,String accessMethod) { -// List records = new ArrayList(); -// //解码 -// List floats = PubUtils.byteArrayToFloatList(Base64.getDecoder().decode(item.getData())); -// if (CollectionUtil.isEmpty(floats)){ -// throw new BusinessException(StatResponseEnum.AUTO_DATA_NULL); -// } -// //校验模板和解码数据数量能否对应上 -// if (!Objects.equals(dataArrayList.size(),floats.size())){ -// throw new BusinessException(StatResponseEnum.ARRAY_DATA_NOT_MATCH); -// } -// Map map = new Gson().fromJson(String.valueOf(redisUtil.getObjectByKey(AppRedisKey.ELE_EPD_PQD)), Map.class); -// for (int i = 0; i < dataArrayList.size(); i++) { -// String tableName = map.get(dataArrayList.get(i).getName()); -// Map tags = new HashMap<>(); -// tags.put(InfluxDBTableConstant.LINE_ID,lineId); -// tags.put(InfluxDBTableConstant.PHASIC_TYPE,dataArrayList.get(i).getPhase()); -// //todo 不清楚之前为啥要修改相别，这边按字典配置相别无法查询到数据，先改回来 -// //tags.put(InfluxDBTableConstant.PHASIC_TYPE,Objects.isNull(PHASE_MAPPING.get(dataArrayList.get(i).getPhase()))?dataArrayList.get(i).getPhase():PHASE_MAPPING.get(dataArrayList.get(i).getPhase())); -// tags.put(InfluxDBTableConstant.VALUE_TYPE,statMethod); -// tags.put(InfluxDBTableConstant.CL_DID,clDid.toString()); -// tags.put(InfluxDBTableConstant.PROCESS,process.toString()); -// tags.put(InfluxDBTableConstant.QUALITY_FLAG,"0"); -// Map fields = new HashMap<>(); -// //这边特殊处理，如果数据为3.14159，则将数据置为null -// fields.put(dataArrayList.get(i).getName(),Objects.equals(floats.get(i),3.14159f) ? null:floats.get(i)); -// fields.put(InfluxDBTableConstant.IS_ABNORMAL,item.getDataTag()); -// //fixme 设备上送的是北京时间，时序数据库录入时 需要utc时间，减去8小时 -// boolean flag = Objects.equals(DicDataEnum.DEV_CLD.getCode(), devType) && Objects.equals(accessMethod, "CLD"); -// Point point = influxDbUtils.pointBuilder(tableName, flag?item.getDataTimeSec():item.getDataTimeSec()-8*3600, TimeUnit.SECONDS, tags, fields); -// BatchPoints batchPoints = BatchPoints.database(influxDbUtils.getDbName()).retentionPolicy("").consistency(InfluxDB.ConsistencyLevel.ALL).build(); -// batchPoints.point(point); -// records.add(batchPoints.lineProtocol()); -// } -// return records; -// } - public List objectToList(Object object) { List urlList = new ArrayList<>(); if (object != null) {