From e629c8b42c65202b7bae2e019f57d05d427b4f8a Mon Sep 17 00:00:00 2001 From: xy <748613696@qq.com> Date: Thu, 2 Jul 2026 19:06:58 +0800 Subject: [PATCH] =?UTF-8?q?feat(api):=20=E6=B7=BB=E5=8A=A0=E6=89=B9?= =?UTF-8?q?=E9=87=8F=E6=8F=92=E5=85=A5=E9=80=9A=E4=BF=A1=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 在CsCommunicateFeignClient中增加insertionList接口方法 - 实现批量插入的fallback处理逻辑 - 在ICsCommunicateService中添加insertionList服务接口 - 完成InfluxdbCsCommunicateServiceImpl中的批量插入实现 - 优化数据处理逻辑，支持运行状态转换和CT/PT计算 - 添加数据排序功能，按项目名、设备名、测点名排序 - 在控制器中增加批量插入数据的REST接口 --- .../api/CsCommunicateFeignClient.java | 5 ++ ...CommunicateFeignClientFallbackFactory.java | 8 ++++ .../data/PqsCommunicateController.java | 9 ++++ .../service/ICsCommunicateService.java | 2 + .../InfluxdbCsCommunicateServiceImpl.java | 47 +++++++++++++++++++ .../service/impl/DataServiceImpl.java | 15 ++++-- 6 files changed, 82 insertions(+), 4 deletions(-) diff --git a/cs-device/cs-device-api/src/main/java/com/njcn/csdevice/api/CsCommunicateFeignClient.java b/cs-device/cs-device-api/src/main/java/com/njcn/csdevice/api/CsCommunicateFeignClient.java index 7b8af61..0042969 100644 --- a/cs-device/cs-device-api/src/main/java/com/njcn/csdevice/api/CsCommunicateFeignClient.java +++ b/cs-device/cs-device-api/src/main/java/com/njcn/csdevice/api/CsCommunicateFeignClient.java @@ -8,6 +8,8 @@ import org.springframework.cloud.openfeign.FeignClient; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestBody; +import java.util.List; + /** * @author xy */ @@ -17,4 +19,7 @@ public interface CsCommunicateFeignClient { @PostMapping("/insertion") HttpResult insertion(@RequestBody PqsCommunicateDto pqsCommunicateDto); + + @PostMapping("/insertionList") + HttpResult insertionList(@RequestBody List list); } diff --git a/cs-device/cs-device-api/src/main/java/com/njcn/csdevice/api/fallback/CsCommunicateFeignClientFallbackFactory.java b/cs-device/cs-device-api/src/main/java/com/njcn/csdevice/api/fallback/CsCommunicateFeignClientFallbackFactory.java index 834fa24..01efa38 100644 --- a/cs-device/cs-device-api/src/main/java/com/njcn/csdevice/api/fallback/CsCommunicateFeignClientFallbackFactory.java +++ b/cs-device/cs-device-api/src/main/java/com/njcn/csdevice/api/fallback/CsCommunicateFeignClientFallbackFactory.java @@ -10,6 +10,8 @@ import feign.hystrix.FallbackFactory; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; +import java.util.List; + /** * 类的介绍： * @@ -35,6 +37,12 @@ public class CsCommunicateFeignClientFallbackFactory implements FallbackFactory< log.error("{}异常，降级处理，异常为：{}","新增记录",cause.toString()); throw new BusinessException(finalExceptionEnum); } + + @Override + public HttpResult insertionList(List list) { + log.error("{}异常，降级处理，异常为：{}","批量插入数据",cause.toString()); + throw new BusinessException(finalExceptionEnum); + } }; } } diff --git a/cs-device/cs-device-boot/src/main/java/com/njcn/csdevice/controller/data/PqsCommunicateController.java b/cs-device/cs-device-boot/src/main/java/com/njcn/csdevice/controller/data/PqsCommunicateController.java index 9e4e0b5..a6e992e 100644 --- a/cs-device/cs-device-boot/src/main/java/com/njcn/csdevice/controller/data/PqsCommunicateController.java +++ b/cs-device/cs-device-boot/src/main/java/com/njcn/csdevice/controller/data/PqsCommunicateController.java @@ -75,4 +75,13 @@ public class PqsCommunicateController extends BaseController { return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, null, methodDescribe); } + @OperateInfo(info = LogEnum.BUSINESS_COMMON,operateType = OperateType.ADD) + @PostMapping("/insertionList") + @ApiOperation("批量插入数据") + public HttpResult insertionList(@RequestBody List list) { + String methodDescribe = getMethodDescribe("insertionList"); + pqsCommunicateCvtQuery.insertionList(list); + return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, null, methodDescribe); + } + } diff --git a/cs-device/cs-device-boot/src/main/java/com/njcn/csdevice/service/ICsCommunicateService.java b/cs-device/cs-device-boot/src/main/java/com/njcn/csdevice/service/ICsCommunicateService.java index 4a21d72..a30d3c1 100644 --- a/cs-device/cs-device-boot/src/main/java/com/njcn/csdevice/service/ICsCommunicateService.java +++ b/cs-device/cs-device-boot/src/main/java/com/njcn/csdevice/service/ICsCommunicateService.java @@ -45,4 +45,6 @@ public interface ICsCommunicateService { List getRawDataEnd(LineCountEvaluateParam lineParam); void insertion(PqsCommunicateDto pqsCommunicateDto); + + void insertionList(List list); } diff --git a/cs-device/cs-device-boot/src/main/java/com/njcn/csdevice/service/impl/InfluxdbCsCommunicateServiceImpl.java b/cs-device/cs-device-boot/src/main/java/com/njcn/csdevice/service/impl/InfluxdbCsCommunicateServiceImpl.java index bd40f37..ef919f8 100644 --- a/cs-device/cs-device-boot/src/main/java/com/njcn/csdevice/service/impl/InfluxdbCsCommunicateServiceImpl.java +++ b/cs-device/cs-device-boot/src/main/java/com/njcn/csdevice/service/impl/InfluxdbCsCommunicateServiceImpl.java @@ -27,6 +27,7 @@ import java.time.ZoneId; import java.time.format.DateTimeFormatter; import java.util.*; import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; /** @@ -321,6 +322,52 @@ public class InfluxdbCsCommunicateServiceImpl implements ICsCommunicateService { } } + @Override + public void insertionList(List list) { + if (CollectionUtils.isEmpty(list)) { + return; + } + Map> groupByDevId = list.stream().collect(Collectors.groupingBy(PqsCommunicateDto::getDevId)); + List lineProtocols = new ArrayList<>(); + groupByDevId.forEach((ndid, dtoList) -> { + LambdaQueryWrapper lambdaQueryWrapper = new LambdaQueryWrapper<>(); + lambdaQueryWrapper.eq(CsEquipmentDeliveryPO::getNdid, ndid).ne(CsEquipmentDeliveryPO::getRunStatus, 0); + CsEquipmentDeliveryPO po = csEquipmentDeliveryMapper.selectOne(lambdaQueryWrapper); + if (po == null) { + log.warn("未找到ndid={}对应的装置信息，跳过", ndid); + return; + } + + InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(PqsCommunicate.class); + influxQueryWrapper.eq(PqsCommunicate::getDevId, po.getId()).timeDesc().limit(1); + List pqsCommunicates = pqsCommunicateMapper.selectByQueryWrapper(influxQueryWrapper); + + Integer lastType = CollectionUtils.isEmpty(pqsCommunicates) ? null : pqsCommunicates.get(0).getType(); + + for (PqsCommunicateDto dto : dtoList) { + if (lastType != null && Objects.equals(lastType, dto.getType())) { + continue; + } + Map tags = new HashMap<>(); + tags.put("dev_id", po.getId()); + Map fields = new HashMap<>(); + fields.put("type", dto.getType()); + fields.put("description", dto.getDescription()); + long time = LocalDateTime.parse(dto.getTime(), DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_PATTERN)) + .atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(); + Point point = influxDbUtils.pointBuilder("pqs_communicate", time, TimeUnit.MILLISECONDS, tags, fields); + BatchPoints batchPoints = BatchPoints.database(influxDbUtils.getDbName()).retentionPolicy("") + .consistency(InfluxDB.ConsistencyLevel.ALL).build(); + batchPoints.point(point); + lineProtocols.add(batchPoints.lineProtocol()); + lastType = dto.getType(); + } + }); + if (!lineProtocols.isEmpty()) { + influxDbUtils.batchInsert(influxDbUtils.getDbName(), "", InfluxDB.ConsistencyLevel.ALL, TimeUnit.MILLISECONDS, lineProtocols); + } + } + public List getPqsCommunicateData(LineCountEvaluateParam lineParam) { InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(PqsCommunicate.class); diff --git a/cs-harmonic/cs-harmonic-boot/src/main/java/com/njcn/csharmonic/service/impl/DataServiceImpl.java b/cs-harmonic/cs-harmonic-boot/src/main/java/com/njcn/csharmonic/service/impl/DataServiceImpl.java index 1fb2faa..768ed3b 100644 --- a/cs-harmonic/cs-harmonic-boot/src/main/java/com/njcn/csharmonic/service/impl/DataServiceImpl.java +++ b/cs-harmonic/cs-harmonic-boot/src/main/java/com/njcn/csharmonic/service/impl/DataServiceImpl.java @@ -1302,7 +1302,6 @@ public class DataServiceImpl implements IDataService { List line2 = linePo.stream().filter(item->item.getClDid()!=0).collect(Collectors.toList()); List lineList2 = line2.stream().map(CsLinePO::getLineId).collect(Collectors.toList()); List> partitions = Lists.partition(lineList2, 95); - //表集合 List dataList = dicDataFeignClient.getDicDataByTypeCode("Data").getData(); Map map = dataList.stream().collect(Collectors.toMap(DictData::getId, v -> v)); @@ -1488,7 +1487,8 @@ public class DataServiceImpl implements IDataService { if (item2.getLineList() != null && item2.getLineList().contains(item.getLineId())) { vo.setProjectName(item2.getProjectName()); vo.setDeviceName(item2.getEquipmentName()); - vo.setRunStatus(item.getRunStatus()); + int runStatus = Objects.equals(item2.getRunStatus(),2) ? 0 : 1; + vo.setRunStatus(runStatus); break; } } @@ -1953,6 +1953,8 @@ public class DataServiceImpl implements IDataService { if (item2.getLineList() != null && item2.getLineList().contains(item.getLineId())) { vo.setProjectName(item2.getProjectName()); vo.setDeviceName(item2.getEquipmentName()); + int runStatus = Objects.equals(item2.getRunStatus(),2) ? 0 : 1; + vo.setRunStatus(runStatus); break; } } @@ -1960,8 +1962,8 @@ public class DataServiceImpl implements IDataService { vo.setLineType(0); vo.setPointId(item.getLineId()); vo.setPointName(lineMap.get(item.getLineId()).getName()); - Double ct = lineMap.get(item.getLineId()).getCtRatio() / (Objects.isNull(lineMap.get(item.getLineId()).getCt2Ratio())?1.0:lineMap.get(item.getLineId()).getCtRatio()); - Double pt = lineMap.get(item.getLineId()).getPtRatio() / (Objects.isNull(lineMap.get(item.getLineId()).getPt2Ratio())?1.0:lineMap.get(item.getLineId()).getPtRatio()); + Double ct = Objects.isNull(lineMap.get(item.getLineId()).getCtRatio())?1.0:lineMap.get(item.getLineId()).getCtRatio() / (Objects.isNull(lineMap.get(item.getLineId()).getCt2Ratio())?1.0:lineMap.get(item.getLineId()).getCtRatio()); + Double pt = Objects.isNull(lineMap.get(item.getLineId()).getPtRatio())?1.0:lineMap.get(item.getLineId()).getPtRatio() / (Objects.isNull(lineMap.get(item.getLineId()).getPt2Ratio())?1.0:lineMap.get(item.getLineId()).getPtRatio()); List children2 = new ArrayList<>(); //指标第二层 List finalChildren = children2; @@ -2050,6 +2052,11 @@ public class DataServiceImpl implements IDataService { vo.setChildren(children2); result.add(vo); }); + if (CollUtil.isNotEmpty(result)) { + result.sort(Comparator.comparing(AppLineDetailVo::getProjectName, Comparator.nullsLast(Comparator.naturalOrder())) + .thenComparing(AppLineDetailVo::getDeviceName, Comparator.nullsLast(Comparator.naturalOrder())) + .thenComparing(AppLineDetailVo::getPointName, Comparator.nullsLast(Comparator.naturalOrder()))); + } return result; }