|
@@ -180,30 +180,43 @@ public class StateEstimationServiceImpl implements StateEstimationService {
|
|
|
List<CompletableFuture<StateEstimation>> futures = feederIdsList.stream()
|
|
List<CompletableFuture<StateEstimation>> futures = feederIdsList.stream()
|
|
|
.map(id -> CompletableFuture.supplyAsync(() -> callWithRetry(id, req), executor))
|
|
.map(id -> CompletableFuture.supplyAsync(() -> callWithRetry(id, req), executor))
|
|
|
.collect(Collectors.toList());
|
|
.collect(Collectors.toList());
|
|
|
|
|
+ //合并所有线路到集合中
|
|
|
List<StateEstimation> mergedList = Lists.newArrayList();
|
|
List<StateEstimation> mergedList = Lists.newArrayList();
|
|
|
- try {
|
|
|
|
|
-
|
|
|
|
|
- List<String> pointTime = Lists.newArrayList();
|
|
|
|
|
- for (CompletableFuture<StateEstimation> f : futures) {
|
|
|
|
|
- // 任一批失败 -> 抛 CompletionException -> 整体失败
|
|
|
|
|
- StateEstimation part = f.get(15,TimeUnit.SECONDS);
|
|
|
|
|
|
|
+ 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()) {
|
|
if (pointTime.isEmpty()) {
|
|
|
pointTime = part.getPointTime();
|
|
pointTime = part.getPointTime();
|
|
|
}
|
|
}
|
|
|
- String feeder = part.getPeriodFeederSeResult().stream()
|
|
|
|
|
- .map(StateEstimation.StateEstimationCommonResult::getPsrId).collect(Collectors.toList()).get(0);
|
|
|
|
|
|
|
+ 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);
|
|
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("调用状估接口失败");
|
|
|
}
|
|
}
|
|
|
- return mergedList;
|
|
|
|
|
- } catch (CompletionException e) {
|
|
|
|
|
- // 取消尚未完成的批次, 减少无效远程调用
|
|
|
|
|
- futures.forEach(f -> f.cancel(true));
|
|
|
|
|
- log.error("[SE] 查询状估失败: {}", e.getMessage(), e);
|
|
|
|
|
- ExceptionCast.cast("调用状估接口失败");
|
|
|
|
|
- } catch (ExecutionException | InterruptedException | TimeoutException e) {
|
|
|
|
|
- futures.forEach(f -> f.cancel(true));
|
|
|
|
|
- ExceptionCast.cast("调用状估接口失败");
|
|
|
|
|
}
|
|
}
|
|
|
|
|
+ log.info("调用状估接口执行完毕,总馈线数量:{},成功数量:{},失败数量{},失败线路ID:{}",
|
|
|
|
|
+ failFeederIds.size(), mergedList.size(),failFeederIds.size(),failFeederIds);
|
|
|
return mergedList;
|
|
return mergedList;
|
|
|
}
|
|
}
|
|
|
|
|
|