SupplyAsync 等待所有 CompletableFutures 完成

Question

我正在下面运行一些异步任务，需要等到它们全部完成。我不知道为什么，但join() 没有强制等待所有任务，代码继续执行而无需等待。连接流没有按预期工作是有原因的吗？

CompletableFutures 列表只是一个映射 supplyAsync 的流

List<Integer> items = Arrays.asList(1, 2, 3);

List<CompletableFuture<Integer>> futures = items
                .stream()
                .map(item -> CompletableFuture.supplyAsync(() ->  {

                    System.out.println("processing");
                    // do some processing here
                    return item;

                }))
                .collect(Collectors.toList());

而我等待未来之类的。

CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]))
                .thenApply(ignored -> futures.stream()
                        .map(CompletableFuture::join)
                        .collect(Collectors.toList()));

我可以使用 futures.forEach(CompletableFuture::join); 等待等待，但我想知道为什么我的流方法不起作用。

Answer 1

这段代码：

CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]))
                .thenApply(ignored -> futures.stream()
                        .map(CompletableFuture::join)
                        .collect(Collectors.toList()));

不是否等待futures 中的所有期货完成。它所做的是创建一个新的未来，它将等待futures 中的所有异步执行在它自己完成之前完成（但在所有这些未来完成之前不会阻塞）。当这个 allOf 未来完成时，您的 thenApply 代码就会运行。但是allOf() 会立即返回而不会阻塞。

这意味着您的代码中的futures.stream().map(CompletableFuture::join).collect(Collectors.toList()) 仅在所有异步执行完成后运行，这与您的目的相违背。 join() 调用将立即返回。但这不是更大的问题。您的挑战是 allOf().thenApply() 不会等待异步执行完成。它只会创造另一个不会阻塞的未来。

最简单的解决方案是使用第一个管道并映射到一个整数列表：

List<Integer> results = items.stream()
    .map(item -> CompletableFuture.supplyAsync(() -> {

        System.out.println("processing " + item);
        // do some processing here
        return item;

    }))
    .collect(Collectors.toList()) //force-submit all
    .stream()
    .map(CompletableFuture::join) //wait for each
    .collect(Collectors.toList());

如果您想使用类似于原始代码的内容，则必须将您的第二个 sn-p 更改为：

List<Integer> reuslts = futures.stream()
    .map(CompletableFuture::join)
    .collect(Collectors.toList());

那是因为CompletableFuture.allOf 不等待，它只是将所有期货组合成一个新的期货，当全部完成时完成：

返回一个新的 CompletableFuture，当所有给定的 CompletableFuture 都完成时，它就完成了。

或者，您仍然可以将allOf() 与join() 一起使用，然后运行您当前的thenApply() 代码：

//wrapper future completes when all futures have completed
CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]))
        .join(); 

//join() calls below return immediately
List<Integer> result = futures.stream()
        .map(CompletableFuture::join) 
        .collect(Collectors.toList());

最后一条语句中的 join() 调用立即返回，因为包装器 (allOf()) 上的 join() 调用将等待传递给它的所有期货完成。这就是为什么当您可以使用第一种方法时，我看不到这样做的原因。