有 3 个服务都是 REST 调用 - serviceOne、serviceTwo 和 serviceThree。serviceTwo 处理依赖于 serviceOne 的输出,serviceThree 处理依赖于 serviceTwo 的输出。但是,serviceThree 需要循环运行。
我想要一种在 serviceThreeRespCF 可完成的未来完成后获得最终响应的方法。目前,plansResponse 在 service3 完成处理之前返回。
我尝试将 join() 用于 serviceThreeRespCF 但没有帮助。有没有更好的方法来做到这一点,即不使用 get() 或 join()?
public CompletionStage<PlansResponse> process() {
final PlansResponse plansResponse = new PlansResponse();
final CompletableFuture<PlansResponse> plansResp = new CompletableFuture<>();
final CompletionStage<ServiceOneResp> serviceOneRespCF = getResp1(); <A REST CALL>
return serviceOneRespCF.thenCompose(serviceOneResp -> {
// function to validate service one resp
if (condition based on validation)
{
List<Plans> plans = serviceOneResp.getPlans();
if (condition) {
plansResponse.setSysMsgs());
}
else {
final CompletionStage<ServiceTwoResp> serviceTwoRespCF = getResp2(); <A REST CALL>
serviceTwoRespCF.thenCompose(serviceTwoResp -> {
plans.parallelStream().forEach(plan -> {
if (Optional.ofNullable(plan.getPlanId()).isPresent()) {
final CompletionStage<ServiceThreeResp> serviceThreeRespCF = getResp3(); <A REST CALL>
serviceThreeRespCF.thenAcceptAsync(serviceThreeResp -> {
Set<PlanDetail> planDetails = new HashSet<>();
planDetails = populatePlanDetails(plan, serviceTwoResp, serviceThreeResp, planDetails);
planDetailsList.addAll(planDetails.stream().distinct().collect(Collectors.toList()));
plansResponse.setPlanDetails(planDetailsList);
planDetails.clear();
});
}
});
return serviceThreeRespCF;
});
}
}
plansResp.complete(plansResponse);
return plansResp;
});
}
编辑 1:尝试加入(return 语句是上面代码的延续)
直接在 serviceThreeRespCF 上
return serviceThreeRespCF; }).toCompletableFuture().join();
使用 thenRunAsync
return serviceThreeRespCF; }).thenRunAsync(() -> { serviceThreeRespCF.toCompletableFuture().join(); });
编辑 2:我尝试使用 allOf 在可完成的未来获取 service3 完成阶段的值。(这件作品是在服务2 thenCompose之后)。可以看出,service2 响应不可用于这个可完成的未来并且它不起作用。除此之外,这还涉及必须遍历计划列表 2 次。还有另一种方法可以做到这一点吗?
CompletableFuture<List<Service3Response>> service3ResponseList = allAsList(futures);
service3ResponseList.thenAccept(service3Responses -> {
AtomicInteger i = new AtomicInteger(0);
plans.parallelStream().forEach(plan -> {
Set<PlanDetail> planDetails = new HashSet<>();
planDetails = populatePlanDetails(acctSetFromRequest, plan, service3Responses.get(i.get()), service2Response <NOT AVAILABLE>, planDetails, acctMapFromResponse.get(plan.getPlanId()));
planDetailsList.addAll(planDetails.stream().distinct().collect(Collectors.toList()));
plansResponse.setDistributionPlanDetails(planDetailsList);
planDetails.clear();
i.addAndGet(1);
});
});
public <T> CompletableFuture<List<T>> allAsList(final List<CompletableFuture<T>> futures) {
return CompletableFuture.allOf(
futures.toArray(new CompletableFuture[futures.size()])
).thenApply(ignored ->
futures.stream().map(CompletableFuture::join).collect(Collectors.toList())
);
}