如何在链式可压缩未来中扇出?

Question

我想链接一个CompletableFuture，这样它在处理过程中就会扇出。我的意思是我有一个针对列表的开放 CompletableFuture，我想对该列表中的每个项目应用计算。

第一步是调用m_myApi.getResponse(request， executor(来发出异步调用。

该异步调用的结果具有getCandidate方法。我想并行解析所有这些候选。

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

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

如本答案中所述，如果Executor恰好是 Fork/Join 池，则有一个(未记录的(功能，即在其工作线程之一中启动并行流将使用该执行器执行并行操作。

当您想要支持任意Executor实现时，事情会更加复杂。一个解决方案看起来像

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

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

在第二步中，我们可以迭代列表并join所有期货。如果执行器是 Fork/Join 池，并且将来尚未完成，它将检测到这一点并启动补偿线程以重新获得配置的并行度。但是，对于任意执行器，我们不能假设这样的功能。最值得注意的是，如果执行程序是单线程执行程序，则可能导致死锁。

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

Answer 2

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

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

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