如何在连锁的未来展开？

Question

我想链一个完整的未来，使它在处理过程中扇出。我的意思是，我对一个列表有一个开放的、可完成的未来，我想对列表中的每一项进行计算。

第一步是调用m_myApi.get响应（请求，执行器）发出异步调用。

该异步调用的结果有一个getCandidates方法。我想并行分析所有这些候选者。

目前，我的代码以串行方式解析它们

public CompletableFuture<List<DOMAIN_OBJECT>> parseAllCandidates(@Nonnull final REQUEST request, @Nonnull final Executor executor)
{
        CompletableFuture<RESPONSE> candidates = m_myApi.getResponse(request, executor);
        return candidates.thenApplyAsync(response -> response.getCandidates()
                                                   .stream()
                                                   .map(MyParser::ParseCandidates)
                                                   .collect(Collectors.toList()));
}

我想要这样的东西：

public CompletableFuture<List<DOMAIN_OBJECT>> parseAllCandidates(@Nonnull final REQUEST request, @Nonnull final Executor executor)
{
        CompletableFuture<RESPONSE> candidates = m_myApi.getResponse(request, executor);
        return candidates.thenApplyAsync(response -> response.getCandidates()
                                                   .stream()
                                                   .PARSE_IN_PARALLEL_USING_EXECUTOR
}

Answer 1

已有一个版本的thenApply将Executor作为附加参数。

<U> CompletionStage<U>  thenApplyAsync​(Function<? super T,​? extends U> fn, Executor executor)

如果在那里传递forkjoin执行器，那么lambda中的并行流将使用传递的执行器而不是公共池。

Answer 2

正如在这个答案中所说，如果执行器碰巧是一个Fork/Join池，则有一个（未记录的）特性，即在一个工作线程中启动并行流将使用该执行器执行并行操作。

当您想要支持任意执行器实现时，事情就更复杂了。一个解决方案看起来像

public CompletableFuture<List<DOMAIN_OBJECT>> parseAllCandidates(
       @Nonnull final REQUEST request, @Nonnull final Executor executor)
{
    CompletableFuture<RESPONSE> candidates = m_myApi.getResponse(request, executor);
    return candidates.thenComposeAsync(
        response -> {
            List<CompletableFuture<DOMAIN_OBJECT>> list = response.getCandidates()
                .stream()
                .map(CompletableFuture::completedFuture)
                .map(f -> f.thenApplyAsync(MyParser::ParseCandidates, executor))
                .collect(Collectors.toList());
            return CompletableFuture.allOf(list.toArray(new CompletableFuture<?>[0]))
                .thenApplyAsync(x ->
                    list.stream().map(CompletableFuture::join).collect(Collectors.toList()),
                    executor);
        },
        executor);
}

第一个关键点是，我们必须在开始等待任何作业之前提交所有可能的异步作业，以实现执行器可能支持的最大并行性。因此，我们必须在第一步收集列表中的所有期货。

在第二步中，我们可以迭代列表并加入所有期货。如果执行器是一个Fork/Join池，并且未来还没有完成，它将检测到这一点并启动一个补偿线程以恢复配置的并行性。然而，对于任意的执行者，我们不能假设这样的特性。最值得注意的是，如果执行器是单线程执行器，这可能会导致死锁。

因此，该解决方案使用CompletableFuture.allOf执行迭代和加入所有期货的操作，只有当所有期货都已经完成时。因此，此解决方案永远不会阻塞执行程序的线程，使其与任何Execator实现兼容。