java - CompletableFuture 单个任务继续执行许多并行任务

Question

我有以下代码:

return CompletableFuture.supplyAsync(() -> {
    return foo; // some custom object
})
.thenAccept(foo -> {
     // ??? need to spawn N async parallel jobs that works on 'foo'
});

英文:第一个任务异步创建foo对象；然后我需要在其上运行 N 个并行进程。

那么有没有更好的方法来做到这一点:

...
CompletableFuture[] parallel = new CompletableFuture[N];
for (int i = 0; i < N; i++) {
    parallel[i] = CompletableFuture.runAsync(() -> {
        work(foo);
    });
}
CompletableFuture.allOf(parallel).join();
...

我不喜欢这样，因为一个线程在等待 N 个作业完成时被锁定。

Answer 1

您可以在特定的先决条件作业上链接任意数量的独立作业，例如

CompletableFuture<Foo> base=CompletableFuture.supplyAsync(() -> new Foo());
Collections.nCopies(N, base).forEach(f -> f.thenAcceptAsync(foo -> work(foo)));

在提供 Foo 实例的初始作业完成后，

将产生 N 个并行作业，同时调用 work(foo)。

但请记住，底层框架将考虑可用 CPU 内核的数量来确定实际执行并行作业的线程池的大小，因此如果 N > #cores，其中一些作业可能一个接一个地跑。

如果工作受 I/O 限制，因此，您希望拥有更多的并行线程，则必须指定自己的执行器。

---

nCopies/forEach 不是必需的，for 循环也可以，但它提供了如何处理后续操作的提示作业，这取决于所有这些并行作业的完成:

CompletableFuture<Foo> base=CompletableFuture.supplyAsync(() -> new Foo());
CompletableFuture<Void> all = CompletableFuture.allOf(
    Collections.nCopies(N, base).stream()
        .map(f -> f.thenAcceptAsync(foo -> work(foo)))
        .toArray(CompletableFuture<?>[]::new));

现在您可以使用 all 检查所有作业的完成情况或链式附加操作。