修复bug。算法内部使用新线程导致LiteFlow层面的process()方法瞬间返回，但实际计算还在进行，会影响有上下层关系的算法计算

algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/ExecutionCenter.java

@@ -123,9 +123,9 @@ public class ExecutionCenter extends BaseController {
                     startDate = DateUtil.offsetDay(startDate, 1);
                 }
                 calculatedParam.setDataDate(DateUtil.format(startDate, DatePattern.NORM_DATE_PATTERN));
-               // CalculatedParam repairParam = BeanUtil.copyProperties(calculatedParam, CalculatedParam.class);
+                CalculatedParam repairParam = BeanUtil.copyProperties(calculatedParam, CalculatedParam.class);
-                calculatedParam.setType(0);
+                repairParam.setType(0);
-                flowService.execute2Resp(methodDescribe, calculatedParam);
+                flowService.execute2Resp(methodDescribe, repairParam);
             }
         } else {
             //非补招

algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/service/line/IDataLimitRateAsync.java

@@ -4,11 +4,13 @@ import com.njcn.device.biz.pojo.po.Overlimit;
 
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.CompletableFuture;
 
 public interface IDataLimitRateAsync {
 
     /**
      * limit_rate多线程算法
+     *
      * @param dataDate
      * @param list
      * @param phase
@@ -16,10 +18,11 @@ public interface IDataLimitRateAsync {
      * @param size
      * @param i
      * @param type         系统类型
+     * @return
      * @Author: wr
      * @Date: 2025/12/17 12:16
      */
-    void lineDataRate(String dataDate,
+    CompletableFuture<Void> lineDataRate(String dataDate,
                                          List<String> list,
                                          List<String> phase,
                                          Map<String, Overlimit> overLimitMap,

algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/serviceimpl/line/DataComAssServiceImpl.java

@@ -58,7 +58,7 @@ public class DataComAssServiceImpl implements IDataComAssService {
         lineParam.setStartTime(TimeUtils.getBeginOfDay(calculatedParam.getDataDate()));
         lineParam.setEndTime(TimeUtils.getEndOfDay(calculatedParam.getDataDate()));
         List<String> lineIdList = calculatedParam.getIdList();
-        getAbnormalData(lineParam);
+        //getAbnormalData(lineParam);
         for (String lineId : lineIdList) {
             DataComassesDPO rStatComassesDpo = new DataComassesDPO();
             rStatComassesDpo.setTime(calculatedParam.getDataDate());

algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/serviceimpl/line/IDataCrossingServiceImpl.java

@@ -46,6 +46,7 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
@@ -108,11 +109,26 @@ public class IDataCrossingServiceImpl implements IDataCrossingService {
         List<List<String>> pendingIds = ListUtils.partition(lineIds, 1);
         ArrayList<String> phase = ListUtil.toList(PhaseType.PHASE_A, PhaseType.PHASE_B, PhaseType.PHASE_C);
         MemorySizeUtil.getNowMemory();
+
+        List<CompletableFuture<Void>> futures = new ArrayList<>();
         for (int i = 0; i < pendingIds.size(); i++) {
             logger.info(calculatedParam.getDataDate()+" 总分区数据：" + pendingIds.size() + "=====》当前第"+(i + 1)+"小分区");
             List<String> list = pendingIds.get(i);
-            dataLimitRateAsync.lineDataRate(calculatedParam.getDataDate(), list, phase, overLimitMap, pendingIds.size(), (i + 1), lineParam.getType());
+            // 获取Future
+            CompletableFuture<Void> future = dataLimitRateAsync.lineDataRate(
+                    calculatedParam.getDataDate(),
+                    list,
+                    phase,
+                    overLimitMap,
+                    pendingIds.size(),
+                    (i + 1),
+                    lineParam.getType()
+            );
+
+            futures.add(future);
         }
+        // 等待所有任务完成
+        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
         System.gc();
     }
 

algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/serviceimpl/line/IDataLimitRateAsyncImpl.java

@@ -25,6 +25,7 @@ import org.springframework.util.CollectionUtils;
 import javax.annotation.Resource;
 import java.lang.reflect.Method;
 import java.util.*;
+import java.util.concurrent.CompletableFuture;
 import java.util.stream.Collectors;
 
 /**
@@ -56,7 +57,7 @@ public class IDataLimitRateAsyncImpl implements IDataLimitRateAsync {
 
     @Override
     @Async("asyncExecutor")
-    public void lineDataRate(String dataDate,
+    public CompletableFuture<Void> lineDataRate(String dataDate,
                                                 List<String> list,
                                                 List<String> phase,
                                                 Map<String, Overlimit> overLimitMap,
@@ -123,7 +124,7 @@ public class IDataLimitRateAsyncImpl implements IDataLimitRateAsync {
          */
         Map<String, List<DataHarmDto>> harmRateV = dataVHarmList.stream()
                 .filter(x -> phase.contains(x.getPhasicType()))
-                .filter(x -> InfluxDBTableConstant.CP95.equals(x.getValueType()))
+                .filter(x -> InfluxDBTableConstant.CP95.equalsIgnoreCase(x.getValueType()))
                 .collect(Collectors.groupingBy(DataHarmDto::getLineId));
 
         /**
@@ -131,7 +132,7 @@ public class IDataLimitRateAsyncImpl implements IDataLimitRateAsync {
          */
         Map<String, List<DataIDto>> dataI = dataIList.stream()
                 .filter(x -> phase.contains(x.getPhasicType()))
-                .filter(x -> InfluxDBTableConstant.CP95.equals(x.getValueType()))
+                .filter(x -> InfluxDBTableConstant.CP95.equalsIgnoreCase(x.getValueType()))
                 .collect(Collectors.groupingBy(DataIDto::getLineId));
 
 
@@ -140,7 +141,7 @@ public class IDataLimitRateAsyncImpl implements IDataLimitRateAsync {
          */
         Map<String, List<DataHarmDto>> inHarmV = dataVInHarmList.stream()
                 .filter(x -> phase.contains(x.getPhasicType()))
-                .filter(x -> InfluxDBTableConstant.CP95.equals(x.getValueType()))
+                .filter(x -> InfluxDBTableConstant.CP95.equalsIgnoreCase(x.getValueType()))
                 .collect(Collectors.groupingBy(DataHarmDto::getLineId));
 
         /**
@@ -148,49 +149,57 @@ public class IDataLimitRateAsyncImpl implements IDataLimitRateAsync {
          */
         Map<String, List<DataVDto>> dataVThd = dataVAllTime.stream()
                 .filter(x -> phase.contains(x.getPhasicType()))
-                .filter(x -> InfluxDBTableConstant.CP95.equals(x.getValueType()))
+                .filter(x -> InfluxDBTableConstant.CP95.equalsIgnoreCase(x.getValueType()))
                 .collect(Collectors.groupingBy(DataVDto::getLineId));
 
         /**
          * 功能描述:获取influxDB -> data_v -> 负序电压不平衡度 -> 最大值 && 日95%概率值
          */
         Map<String, List<DataVDto>> dataVUnbalance = dataVAllTime.stream()
-                .filter(x -> InfluxDBTableConstant.PHASE_TYPE_T.equals(x.getPhasicType()))
+                .filter(x -> InfluxDBTableConstant.PHASE_TYPE_T.equalsIgnoreCase(x.getPhasicType()))
-                .filter(x -> InfluxDBTableConstant.CP95.equals(x.getValueType()) ||
+                .filter(x -> InfluxDBTableConstant.CP95.equalsIgnoreCase(x.getValueType()) ||
-                        InfluxDBTableConstant.MAX.equals(x.getValueType()))
+                        InfluxDBTableConstant.MAX.equalsIgnoreCase(x.getValueType()))
                 .collect(Collectors.groupingBy(DataVDto::getLineId));
 
         /**
          * 功能描述:获取influxDB -> data_i -> 负序电流 -> 最大值 && 日95%概率值
          */
         Map<String, List<DataIDto>> dataINeg = dataIList.stream()
-                .filter(x -> InfluxDBTableConstant.PHASE_TYPE_T.equals(x.getPhasicType()))
+                .filter(x -> InfluxDBTableConstant.PHASE_TYPE_T.equalsIgnoreCase(x.getPhasicType()))
-                .filter(x -> InfluxDBTableConstant.CP95.equals(x.getValueType()) ||
+                .filter(x -> InfluxDBTableConstant.CP95.equalsIgnoreCase(x.getValueType()) ||
-                        InfluxDBTableConstant.MAX.equals(x.getValueType()))
+                        InfluxDBTableConstant.MAX.equalsIgnoreCase(x.getValueType()))
                 .collect(Collectors.groupingBy(DataIDto::getLineId));
 
         /**
          * 功能描述:获取influxDB -> data_v -> 频率偏差 -> 最大值 && 最小值
          */
         Map<String, List<DataVDto>> dataVFreq = dataVAllTime.stream()
-                .filter(x -> InfluxDBTableConstant.PHASE_TYPE_T.equals(x.getPhasicType()))
+                .filter(x -> InfluxDBTableConstant.PHASE_TYPE_T.equalsIgnoreCase(x.getPhasicType()))
-                .filter(x -> InfluxDBTableConstant.MIN.equals(x.getValueType()) ||
+                .filter(x -> InfluxDBTableConstant.MIN.equalsIgnoreCase(x.getValueType()) ||
-                        InfluxDBTableConstant.MAX.equals(x.getValueType()))
+                        InfluxDBTableConstant.MAX.equalsIgnoreCase(x.getValueType()))
                 .collect(Collectors.groupingBy(DataVDto::getLineId));
         /**
          * 功能描述:获取influxDB -> data_v -> 电压偏差 -> 最大值
          */
         Map<String, List<DataVDto>> dataVDev = dataVAllTime.stream()
                 .filter(x -> phase.contains(x.getPhasicType()))
-                .filter(x -> InfluxDBTableConstant.MAX.equals(x.getValueType()))
+                .filter(x -> InfluxDBTableConstant.MAX.equalsIgnoreCase(x.getValueType()))
                 .collect(Collectors.groupingBy(DataVDto::getLineId));
 
         /**
          * 功能描述:获取influxDB -> data_plt -> 长时间闪变 -> 注(取最大值原始算法去掉了，现没有根据最大值比较)
          */
-        Map<String, List<DataPltDto>> dataPlt = dataPltAllTime.stream()
+        Map<String, List<DataPltDto>> dataPlt;
+        if (type == 2) {
+            dataPlt = dataPltAllTime.stream()
                     .filter(x -> phase.contains(x.getPhasicType()))
+                    .filter(x -> InfluxDBTableConstant.MAX.equalsIgnoreCase(x.getValueType()))
                     .collect(Collectors.groupingBy(DataPltDto::getLineId));
+        } else {
+            dataPlt = dataPltAllTime.stream()
+                    .filter(x -> PhaseType.PHASE_A.equals(x.getPhasicType()))
+                    .collect(Collectors.groupingBy(DataPltDto::getLineId));
+        }
 
         for (String item : list) {
             if (ObjectUtil.isNotNull(overLimitMap.get(item))) {
@@ -229,6 +238,8 @@ public class IDataLimitRateAsyncImpl implements IDataLimitRateAsync {
             MemorySizeUtil.getNowMemory();
         }
         System.gc();
+
+        return CompletableFuture.completedFuture(null);
     }
 
     /**

algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/serviceimpl/line/PollutionCalcImpl.java

@@ -80,19 +80,21 @@ public class PollutionCalcImpl implements IPollutionCalc {
                     LineDevGetDTO line = lineDetailMap.get(id);
                     if (limitMap.containsKey(id)) {
                         Overlimit overlimit = limitMap.get(id);
+                        lineParam.setValueType(Arrays.asList(line.getTimeInterval() + ""));
+                        lineParam.setLineId(Arrays.asList(id));
+
+                        if(!dataVFeignClient.excludeZeroData(lineParam).getData()){
                             dataPollutionD = new DataPollutionD();
                             dataPollutionD.setLineId(id);
                             dataPollutionD.setDataDate(LocalDateTimeUtil.parseDate(calculatedParam.getDataDate()));
                             dataPollutionD.setPollutionType(vHarmonicLimit);
-                        lineParam.setValueType(Arrays.asList(line.getTimeInterval() + ""));
-                        lineParam.setLineId(Arrays.asList(id));
-
                             List<DataVDto> dataVDtoList = dataVFeignClient.getGroupByTimeDataV(lineParam).getData();
                             List<DataHarmDto> dataHarmDtoList = dataHarmRateVFeignClient.getGroupByTimeHarmRateV(lineParam).getData();
                             //计算谐波电压污染值
                             dataPollutionD.setValue(PubUtils.doubleRound(2, calcVAllPollutionValue(dataVDtoList, dataHarmDtoList, overlimit) * line.getTimeInterval()));
-
                             list.add(dataPollutionD);
+                        }
+
                         dataPollutionD = new DataPollutionD();
                         dataPollutionD.setLineId(id);
                         dataPollutionD.setDataDate(LocalDateTimeUtil.parseDate(calculatedParam.getDataDate()));

data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/api/DataVFeignClient.java

@@ -68,4 +68,8 @@ public interface DataVFeignClient {
     //按时间分组获取原始数据
     @PostMapping("/getGroupByTimeDataV")
     HttpResult<List<DataVDto>> getGroupByTimeDataV(@RequestBody LineCountEvaluateParam lineParam);
+
+
+    @PostMapping("/excludeZeroData")
+    HttpResult<Boolean> excludeZeroData(@RequestBody LineCountEvaluateParam lineParam);
 }

data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/api/fallback/DataVFeignClientFallbackFactory.java

@@ -116,6 +116,12 @@ public class DataVFeignClientFallbackFactory implements FallbackFactory<DataVFei
                 log.error("{}异常，降级处理，异常为：{}","DataV按时间分组获取原始数据",cause.toString());
                 throw new BusinessException(finalExceptionEnum);
             }
+
+            @Override
+            public HttpResult<Boolean> excludeZeroData(LineCountEvaluateParam lineParam) {
+                log.error("{}异常，降级处理，异常为：{}","判断是否存在是零飘数据",cause.toString());
+                throw new BusinessException(finalExceptionEnum);
+            }
         };
     }
 }

data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/pojo/dto/PqsCommunicateDto.java

@@ -18,6 +18,8 @@ public class PqsCommunicateDto {
     private String description;
 
     private Integer type;
+    //是否更新updateTime标志；数据上送更新1，状态翻转不更新0
+    private Integer flag=0;
 
 
 }

data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/controller/DataVController.java

@@ -166,6 +166,15 @@ public class DataVController extends BaseController {
         return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, dataV, methodDescribe);
     }
 
+    @OperateInfo(info = LogEnum.BUSINESS_COMMON)
+    @PostMapping("/excludeZeroData")
+    @ApiOperation("判断是否存在是零飘数据")
+    public HttpResult<Boolean> excludeZeroData(@RequestBody LineCountEvaluateParam lineParam) {
+        String methodDescribe = getMethodDescribe("excludeZeroData");
+        Boolean b = dataVQuery.excludeZeroData(lineParam);
+        return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, b, methodDescribe);
+    }
+
 
 
 

data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/IDataV.java

@@ -81,4 +81,17 @@ public interface IDataV extends IMppService<RStatDataVD> {
     List<DataVDto> getDataV(LineCountEvaluateParam lineParam);
 
     List<DataVDto> getGroupByTimeDataV(LineCountEvaluateParam lineParam);
+
+
+    /**
+     * 查看监测点的电压，判断是否存在是零飘数据
+     * 如果零飘数据占比一半多，则抛弃该监测点数据
+     * @param lineParam
+     * @return: java.lang.Boolean
+     * @Author: wr
+     * @Date: 2026/1/15 9:26
+     */
+    Boolean excludeZeroData(LineCountEvaluateParam lineParam);
+
+
 }

data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/LnDataDealServiceImpl.java

@@ -97,7 +97,7 @@ public class LnDataDealServiceImpl implements LnDataDealService {
                 pqsCommunicateDto.setTime(LocalDateTimeUtil.format(dataVDTO.getTimeid(), DatePattern.NORM_DATETIME_PATTERN));
                 pqsCommunicateDto.setDevId(temp);
                 pqsCommunicateDto.setType(1);
-
+                pqsCommunicateDto.setFlag(1);
 
                 iPqsCommunicate.insertion(pqsCommunicateDto);
             });

data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/influxdb/InfluxdbDataVImpl.java

@@ -425,6 +425,27 @@ public class InfluxdbDataVImpl extends MppServiceImpl<RStatDataVRelationMapper,
         return result;
     }
 
+
+    @Override
+    public Boolean excludeZeroData(LineCountEvaluateParam lineParam) {
+        InfluxQueryWrapper dataVQueryWrapper = new InfluxQueryWrapper(DataV.class);
+        dataVQueryWrapper.eq(DataV::getLineId, lineParam.getLineId().get(0))
+                .eq(DataV::getValueType, InfluxDbSqlConstant.AVG_WEB)
+                .ne(DataV::getPhasicType, InfluxDBTableConstant.PHASE_TYPE_T)
+                .select(DataV::getRms)
+                .between(DataV::getTime, lineParam.getStartTime(), lineParam.getEndTime());
+        List<DataV> rmsResult = dataVMapper.selectByQueryWrapper(dataVQueryWrapper);
+        if (CollUtil.isNotEmpty(rmsResult)) {
+            List<DataV> exceptionData = rmsResult.stream().filter(dataV -> dataV.getRms() < 1.0).collect(Collectors.toList());
+            if (CollUtil.isNotEmpty(exceptionData)) {
+                if (exceptionData.size() * 2 >= rmsResult.size()) {
+                    return true;
+                }
+            }
+        }
+        return false;
+    }
+
     /**
      * 按监测点集合、时间条件获取dataV分钟数据
      * timeMap参数来判断是否进行数据处理 timeMap为空则不进行数据处理

data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/influxdb/InfluxdbPqsCommunicateImpl.java

@@ -129,7 +129,9 @@ public class InfluxdbPqsCommunicateImpl implements IPqsCommunicate {
         //更新mysql数据
         DevComFlagDTO devComFlagDTO = new DevComFlagDTO();
         devComFlagDTO.setId(pqsCommunicateDto.getDevId());
+        if(Objects.equals(pqsCommunicateDto.getFlag(),1)){
             devComFlagDTO.setDate(LocalDateTime.parse(pqsCommunicateDto.getTime(), DATE_TIME_FORMATTER));
+        }
         devComFlagDTO.setStatus(pqsCommunicateDto.getType());
 
         deviceFeignClient.updateDevComFlag(devComFlagDTO);

data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/relation/RelationDataVImpl.java

@@ -167,6 +167,11 @@ public class RelationDataVImpl extends MppServiceImpl<RStatDataVRelationMapper,
         return Collections.emptyList();
     }
 
+    @Override
+    public Boolean excludeZeroData(LineCountEvaluateParam lineParam) {
+        return null;
+    }
+
     private List<DataVDto> quality(List<DataVDto> list, LineCountEvaluateParam lineParam) {
         List<DataVDto> result = new ArrayList<>();
         Map<String, List<DataVDto>> lineMap = list.stream().collect(Collectors.groupingBy(DataVDto::getLineId));