合并多个CompletableFutures

Question

我有以下组成部分：

private JobInfo aggregateJobInfo() { 
    final JobsResult jobsResult = restClient().getJobs(); 
    final List<String> jobIds = extractJobIds(jobsResult); 

    //fetch details, exceptions and config for each job 
    final List<JobDetails> jobDetails = jobIds.stream().map(jobId -> { 
     final JobDetailResult jobDetailResult = restClient().getJobDetails(jobId); 
     final JobExceptionsResult jobExceptionsResult = restClient().getJobExceptions(jobId); 
     final JobConfigResult jobConfigResult = restClient().getJobConfig(jobId); 
     return new JobDetails(jobDetailResult, jobExceptionsResult, jobConfigResult); 
    }).collect(Collectors.toList()); 
    return new JobInfo(jobsResult, jobDetails); 
} 

private static List<String> extractJobIds(final JobsResult jobsResult) { 
    final ArrayList<String> jobIds = new ArrayList<>(); 
    jobIds.addAll(jobsResult.getRunning()); 
    jobIds.addAll(jobsResult.getFinished()); 
    jobIds.addAll(jobsResult.getCanceled()); 
    jobIds.addAll(jobsResult.getFailed()); 
    return jobIds; 
} 

它只是调用一些终端和aggergates一些数据。现在，我尝试使用CompletableFutures，我并没有真正使用过，以使该无阻塞..

private CompletableFuture<JobInfo> aggregateJobInfo() { 
    final CompletableFuture<JobsResult> jobsResultFuture = restClient().getJobs(); 
    final CompletableFuture<List<String>> jobIdsFuture = jobsResultFuture.thenApply(JobInfoCollector::extractJobIds); 

    //fetch details, exceptions and config for each job 
    final CompletableFuture<List<CompletableFuture<JobDetails>>> jobDetailsFuture = jobIdsFuture.thenApply(jobIds -> { 
     return jobIds.stream().map(jobId -> { 
      final CompletableFuture<JobDetailResult> jobDetailsResultFuture = restClient().getJobDetails(jobId); 
      final CompletableFuture<JobExceptionsResult> jobExceptionsFuture = restClient().getJobExceptions(jobId); 
      final CompletableFuture<JobConfigResult> jobConfigFuture = restClient().getJobConfig(jobId); 
      return jobDetailsResultFuture.thenCompose(jobDetailResult -> { 
       return jobExceptionsFuture.thenCombine(jobConfigFuture, (jobExceptionsResult, jobConfigResult) -> { 
        return new JobDetails(jobDetailResult, jobExceptionsResult, jobConfigResult); 
       }); 
      }); 

     }).collect(Collectors.toList()); 
    }); 
    return null; 

我的问题是如何在这里创建CompletableFuture时JOBINFO是'新JOBINFO（jobsResult，jobDetails ）？

正如我所说，我是新手，也许我的方法不好，有更好的解决方案？

赞赏任何想法，感谢

第一工作版本：

private CompletableFuture<JobInfo> aggregateJobInfo() { 

    final CompletableFuture<JobsResult> jobsResultFuture = restClient().getJobs(); 
    final CompletableFuture<List<String>> jobIdsFuture = jobsResultFuture.thenApply(JobInfoCollector::extractJobIds); 

    final CompletableFuture<List<CompletableFuture<JobDetails>>> jobDetailsFutureListFuture = 
      jobIdsFuture.thenApply(jobIds -> jobIds.stream().map(jobId -> { 
       final CompletableFuture<JobDetailResult> jobDetailsResultFuture = restClient().getJobDetails(jobId); 
       final CompletableFuture<JobExceptionsResult> jobExceptionsFuture = restClient().getJobExceptions(jobId); 
       final CompletableFuture<JobConfigResult> jobConfigFuture = restClient().getJobConfig(jobId); 
       return jobDetailsResultFuture.thenCompose(jobDetailResult -> 
         jobExceptionsFuture.thenCombine(jobConfigFuture, (jobExceptionsResult, jobConfigResult) -> 
           new JobDetails(jobDetailResult, jobExceptionsResult, jobConfigResult))); 
      }).collect(Collectors.toList())); 

    return jobDetailsFutureListFuture.thenCompose(jobDetailsFutures -> 
      CompletableFuture.allOf(jobDetailsFutures.toArray(
        new CompletableFuture[jobDetailsFutures.size()])).thenApply(aVoid -> 
        jobDetailsFutures.stream() 
          .map(CompletableFuture::join) 
          .collect(Collectors.toList()))) 
      .thenApply(jobDetails -> jobsResultFuture.thenApply(jobsResult -> 
        new JobInfo(jobsResult, jobDetails))) 
      .join(); 
} 

Answer 1

您有：

• CompletableFuture<JobsResult> jobsResultFuture
• CompletableFuture<List<CompletableFuture<JobDetails>>> jobDetailsFuture
• JobInfo(JobsResult a, List<JobDetails> b)

你想

CompletableFuture<JobInfo>

额外观察：当jobsResultFuture完成jobDetailsFuture才能完成。通过thenApply

List<JobDetails> + CompletableFuture<JobsResult>>List<JobDetails>（如捕获 -

1. List<CompletableFuture<JobDetails>> - >通过allOfVoid在thenCompose
2. Void + List<CompletableFuture<JobDetails>>（如捕获VAR）：

   这样你就可以实现以下var）→JobInfo via thenApply

您可以简单地通过get()在这些映射函数内展开期货，因为由于当时的祖先期货依赖关系，期货被保证在该点完成。

其他方法使用thenCombine和流量减少将是可能的，但更详细，并创造更多的中间期货。