【问题标题】:Combining multiple CompletableFutures组合多个 CompletableFuture
【发布时间】:2016-10-28 11:09:25
【问题描述】:

我有以下组件:

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;
}

它只是调用一些 ENDPOINTS 并聚合一些数据。现在我正在尝试通过使用 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;

我的问题是当 JobInfo 为 `new JobInfo(jobsResult, jobDetails) 时如何在这里创建 CompletableFuture?!

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

任何想法表示赞赏,谢谢

第一个工作版本:

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();
}

【问题讨论】:

  • 您粘贴的大部分代码似乎与问题无关。你能把它简化为你需要的最小例子吗?
  • 问题是如何“映射”上面的未来数据,以便返回 CompletableFuture

标签: java-8 completable-future


【解决方案1】:

你有:

  • CompletableFuture&lt;JobsResult&gt; jobsResultFuture
  • CompletableFuture&lt;List&lt;CompletableFuture&lt;JobDetails&gt;&gt;&gt; jobDetailsFuture
  • JobInfo(JobsResult a, List&lt;JobDetails&gt; b)

你想要

CompletableFuture&lt;JobInfo&gt;

补充观察:jobDetailsFuture 只有在jobsResultFuture 完成后才能完成。

所以你可以实现以下:

  1. List&lt;CompletableFuture&lt;JobDetails&gt;&gt; -> Void 通过allOfthenCompose
  2. Void + List&lt;CompletableFuture&lt;JobDetails&gt;&gt;(作为捕获的变量)-> List&lt;JobDetails&gt; 通过 thenApply
  3. List&lt;JobDetails&gt; + CompletableFuture&lt;JobsResult&gt;(作为捕获的变量)-> JobInfo 通过thenApply

您可以通过 get() 在这些映射器函数中简单地解包期货,因为由于其祖先期货在该点的依赖关系,期货可以保证在该点完成。

使用thenCombine 和流缩减的其他方法是可能的,但更冗长并创建更多中间期货。

【讨论】:

  • 对不起,我不明白..'awaitAll' 和虚空是什么意思?
  • 表示allOf。它返回一个CompletableFuture&lt;Void&gt;,因此下一步是取回数据
  • 还是不明白。试过 jobDetailsListFuture.thenCompose(jobDetailsFutures -> { return CompletableFuture.allOf(jobDetailsFutures.toArray( new CompletableFuture[jobDetailsFutures.size()])) .thenApply(aVoid -> { ??? }); });但不知道如何处理虚空。
  • 你什么都不做,第二部分很重要
  • 将可完成期货视为两个独立的信息流:完成/计算进度和携带的数据对象。有时您只对完成感兴趣而不对数据感兴趣,因此您可以知道何时可以安全地调用get() 而不会出现阻塞或异常,可能在与完成的不同的期货上。
猜你喜欢
  • 1970-01-01
  • 2021-06-04
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 2018-08-17
  • 1970-01-01
相关资源
最近更新 更多