CompletableFuture单任务继续执行多个并行任务

Question

CompletableFuture单任务继续执行多个并行任务

javamultithreadingjava-8completable-future

3

我有以下代码：

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个任务完成时，一个线程会被锁定。

- igr

为什么你会有这一行代码 CompletableFuture.allOf(parallel).join();，当你不想等待完成时？没有人要求你等待... - Holger

我当时是盲目的。 - igr

2个回答

0

由于CompletableFuture.allOf已经返回另一个CompletableFuture<Void>，因此您可以在其上执行另一个.thenAccept，并在回调中从parallel中提取CFs的返回值，这样您就避免了调用join

- the8472

或者我可以在外部创建数组，然后在循环中使用thenAcceptAsync创建并行任务。 - igr

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- Holger · Accepted Answer

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

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

这将产生N个并行任务，同时调用work(foo)，在提供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来检查所有作业的完成情况或链接其他操作。