状估数据分批调用

services/load-transfer-si/src/main/java/com/hdkj/lt/si/config/SeConfig.java

@@ -1,5 +1,6 @@
 package com.hdkj.lt.si.config;
 
+import com.ruoyi.common.core.config.ThreadPoolProperties;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -11,8 +12,10 @@ import java.util.concurrent.atomic.AtomicInteger;
 
 /**
  * 状估模块 Bean 配置: RestTemplate (带超时) + 有界线程池.
+ * @author Annie
  */
 @Configuration
+@EnableConfigurationProperties(ThreadPoolProperties.class)
 public class SeConfig {
 
     /**
@@ -20,20 +23,22 @@ public class SeConfig {
      * 即并发调用远程接口的全局上限. 守护线程, 应用关闭时 shutdown.
      */
     @Bean(name = "seQueryExecutor", destroyMethod = "shutdown")
-    public ExecutorService seQueryExecutor() {
+    public ExecutorService seQueryExecutor(ThreadPoolProperties config) {
         ThreadFactory factory = new ThreadFactory() {
             private final AtomicInteger counter = new AtomicInteger(0);
 
             @Override
             public Thread newThread(Runnable r) {
-                Thread t = new Thread(r, "se-query-" + counter.incrementAndGet());
-//                t.setDaemon(true);
-                return t;
+                return new Thread(r, "se-query-" + counter.incrementAndGet());
             }
         };
         return new ThreadPoolExecutor(
-                8,8,0L, TimeUnit.MILLISECONDS,
-                new ArrayBlockingQueue<>(200),factory,new ThreadPoolExecutor.AbortPolicy()
+                config.getCoreSize(),
+                config.getMaxSize(),
+                config.getKeepAlive(),
+                TimeUnit.SECONDS,
+                new ArrayBlockingQueue<>(config.getQueueSize()),factory,
+                new ThreadPoolExecutor.CallerRunsPolicy()
         );
     }
 }

services/load-transfer-si/src/main/java/com/hdkj/lt/si/service/estimation/impl/StateEstimationServiceImpl.java

@@ -70,7 +70,7 @@ public class StateEstimationServiceImpl implements StateEstimationService {
     /**
      * 单批 psrIds 上限 (远程接口限制 100)
      * */
-    private static final int batchSize = 100;
+    private static final int batchSize = 50;
     @Autowired
     private FhzgFeederAutoSwitchMappingMapper fhzgFeederAutoSwitchMappingMapper;
 
@@ -172,51 +172,56 @@ public class StateEstimationServiceImpl implements StateEstimationService {
         List<String> feederIdsList = list.stream().map(DwdShbDsFeederBase::getPsrId).collect(Collectors.toList());
 
         // 分批 (≤ batchSize, 远程限制 100)
-//        List<List<String>> batches = partition(feederIdsList, batchSize);
+        List<List<String>> batches = partition(feederIdsList, batchSize);
 //        log.info("[SE] feederIds 总量={},  批次数={}, 批大小={}", feederIdsList.size(), batches.size(), batchSize);
-
-        // 并行调用 (有界线程池), 每批带重试, 任一批最终失败则整体失败
-        StateEstimationRequest req = buildStateEstimationReq();
-        List<CompletableFuture<StateEstimation>> futures = feederIdsList.stream()
-                .map(id -> CompletableFuture.supplyAsync(() -> callWithRetry(id, req), executor))
-                .collect(Collectors.toList());
         //合并所有线路到集合中
         List<StateEstimation> mergedList = Lists.newArrayList();
         List<String> pointTime = Lists.newArrayList();
         //存放失败的线路id
         List<String> failFeederIds = Lists.newArrayList();
-        for (int i = 0; i < futures.size(); i++) {
-            CompletableFuture<StateEstimation> future = futures.get(i);
-            String feederId = feederIdsList.get(i);
-            try {
-                StateEstimation part = future.get(20L,TimeUnit.SECONDS);
-                if (pointTime.isEmpty()) {
-                    pointTime = part.getPointTime();
+        // 并行调用 (有界线程池), 每批带重试, 任一批最终失败则整体失败
+        StateEstimationRequest req = buildStateEstimationReq();
+        for (List<String> batchFeederIds : batches) {
+            List<CompletableFuture<StateEstimation>> futures = batchFeederIds.stream()
+                    .map(id -> CompletableFuture.supplyAsync(() -> callWithRetry(id, req), executor))
+                    .collect(Collectors.toList());
+
+            for (int i = 0; i < futures.size(); i++) {
+                CompletableFuture<StateEstimation> future = futures.get(i);
+                String feederId = feederIdsList.get(i);
+                try {
+                    StateEstimation part = future.get(20L,TimeUnit.SECONDS);
+                    if (pointTime.isEmpty()) {
+                        pointTime = part.getPointTime();
+                    }
+                    List<String> feederList = part.getPeriodFeederSeResult().stream()
+                            .map(StateEstimation.StateEstimationCommonResult::getPsrId)
+                            .collect(Collectors.toList());
+                    String feeder = feederList.isEmpty() ? null : feederList.get(0);
+                    mergeToList(mergedList, part, feeder);
+                } catch (CompletionException e) {
+                    // 取消尚未完成的批次, 减少无效远程调用
+                    futures.forEach(f -> f.cancel(true));
+                    log.error("[SE] 查询状估失败: {}", e.getMessage(), e);
+                    failFeederIds.add(feederId);
+                    ExceptionCast.cast("调用状估接口失败");
+                } catch (ExecutionException | InterruptedException | TimeoutException e) {
+                    futures.forEach(f -> f.cancel(true));
+                    failFeederIds.add(feederId);
+                    ExceptionCast.cast("调用状估接口失败");
+                }catch (Exception e) {
+                    //未知错误
+                    futures.forEach(f -> f.cancel(true));
+                    failFeederIds.add(feederId);
+                    ExceptionCast.cast("调用状估接口失败");
                 }
-                List<String> feederList = part.getPeriodFeederSeResult().stream()
-                        .map(StateEstimation.StateEstimationCommonResult::getPsrId)
-                        .collect(Collectors.toList());
-                String feeder = feederList.isEmpty() ? null : feederList.get(0);
-                mergeToList(mergedList, part, feeder);
-            } catch (CompletionException e) {
-                // 取消尚未完成的批次, 减少无效远程调用
-                futures.forEach(f -> f.cancel(true));
-                log.error("[SE] 查询状估失败: {}", e.getMessage(), e);
-                failFeederIds.add(feederId);
-                ExceptionCast.cast("调用状估接口失败");
-            } catch (ExecutionException | InterruptedException | TimeoutException e) {
-                futures.forEach(f -> f.cancel(true));
-                failFeederIds.add(feederId);
-                ExceptionCast.cast("调用状估接口失败");
-            }catch (Exception e) {
-                //未知错误
-                futures.forEach(f -> f.cancel(true));
-                failFeederIds.add(feederId);
-                ExceptionCast.cast("调用状估接口失败");
             }
+            log.info("当前批次处理完成，本批馈线数量：{},成功数量：{}，失败数量{}，失败线路ID：{}",
+                    batchFeederIds.size(), mergedList.size(),failFeederIds.size(),failFeederIds);
         }
         log.info("调用状估接口执行完毕，总馈线数量：{},成功数量：{}，失败数量{}，失败线路ID：{}",
-                failFeederIds.size(), mergedList.size(),failFeederIds.size(),failFeederIds);
+                feederIdsList.size(), mergedList.size(),failFeederIds.size(),failFeederIds);
+
         return mergedList;
     }