清单 走向未来 序列

Question

我正在尝试转换List<CompletableFuture<X>>为CompletableFuture<List<T>>。当您有许多异步任务并且需要获得所有异步任务的结果时，这非常有用。

如果它们中的任何一个失败，那么最终的未来将失败。这就是我实现的方式：

  public static <T> CompletableFuture<List<T>> sequence2(List<CompletableFuture<T>> com, ExecutorService exec) {
        if(com.isEmpty()){
            throw new IllegalArgumentException();
        }
        Stream<? extends CompletableFuture<T>> stream = com.stream();
        CompletableFuture<List<T>> init = CompletableFuture.completedFuture(new ArrayList<T>());
        return stream.reduce(init, (ls, fut) -> ls.thenComposeAsync(x -> fut.thenApplyAsync(y -> {
            x.add(y);
            return x;
        },exec),exec), (a, b) -> a.thenCombineAsync(b,(ls1,ls2)-> {
            ls1.addAll(ls2);
            return ls1;
        },exec));
    }

要运行它：

ExecutorService executorService = Executors.newCachedThreadPool();
        Stream<CompletableFuture<Integer>> que = IntStream.range(0,100000).boxed().map(x -> CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep((long) (Math.random() * 10));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return x;
        }, executorService));
CompletableFuture<List<Integer>> sequence = sequence2(que.collect(Collectors.toList()), executorService);

如果其中任何一个失败，则失败。即使有一百万个期货，它也能提供预期的输出。我的问题是：假设如果有超过5000个期货，并且其中任何一个失败，我都会得到StackOverflowError：

java.util.concurrent.CompletableFuture.internalComplete（CompletableFuture.java:210）处的线程“
pool-1-thread-2611”中的java.lang.StackOverflowError，java.util.concurrent.CompletableFuture
$ ThenCompose.run（CompletableFuture.java）
：1487），位于java.util.concurrent.CompletableFuture.postComplete（CompletableFuture.java:193），位于java.util.concurrent.CompletableFuture.internalComplete（CompletableFuture.java:210），位于java.util.concurrent.CompletableFuture
$ ThenCompose.run（ CompletableFuture.java:1487）

我做错了什么？

注意：当任何将来失败时，上述返回的将来都会失败。接受的答案也应考虑这一点。

Answer 1

用途CompletableFuture.allOf(...)：

static<T> CompletableFuture<List<T>> sequence(List<CompletableFuture<T>> com) {
    return CompletableFuture.allOf(com.toArray(new CompletableFuture<?>[0]))
            .thenApply(v -> com.stream()
                .map(CompletableFuture::join)
                .collect(Collectors.toList())
            );
}

关于您的实现的一些评论：

您使用的.thenComposeAsync，.thenApplyAsync并且.thenCombineAsync很可能没有做你的期望。这些...Async方法在单独的线程中运行提供给它们的函数。因此，在您的情况下，您导致将新项添加到列表中以在提供的执行程序中运行。无需将轻量级操作填充到缓存的线程执行器中。请勿在thenXXXXAsync无充分理由的情况下使用方法。

另外，reduce不应用于堆积到易变容器中。即使在流是顺序的流时它可能正确工作，但是如果将流设为并行流，它将失败。要执行可变减少，请.collect改用。

如果要在第一次失败后立即异常完成整个计算，请在您的sequence方法中执行以下操作：

CompletableFuture<List<T>> result = CompletableFuture.allOf(com.toArray(new CompletableFuture<?>[0]))
        .thenApply(v -> com.stream()
                .map(CompletableFuture::join)
                .collect(Collectors.toList())
        );

com.forEach(f -> f.whenComplete((t, ex) -> {
    if (ex != null) {
        result.completeExceptionally(ex);
    }
}));

return result;

此外，如果您想在第一次失败时取消其余操作，请在exec.shutdownNow();之后添加result.completeExceptionally(ex);。当然，这假定exec仅针对这一计算存在。如果没有，则必须循环遍历并分别取消剩余的Future每个。