java - 从 List 转换为 CompletableFuture

Question

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

如果其中任何一个失败，那么最终的 future 将失败。这就是我的实现方式:

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

如果其中任何一个失败，那么它就会失败。即使有一百万个 future ，它也会按预期提供输出。我遇到的问题是:假设有超过 5000 个 future ，如果其中任何一个失败，我会得到 StackOverflowError :

Exception in thread "pool-1-thread-2611" java.lang.StackOverflowErroratjava.util.concurrent.CompletableFuture.internalComplete(CompletableFuture.java:210)atjava.util.concurrent.CompletableFuture$ThenCompose.run(CompletableFuture.java:1487)atjava.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:193)atjava.util.concurrent.CompletableFuture.internalComplete(CompletableFuture.java:210)atjava.util.concurrent.CompletableFuture$ThenCompose.run(CompletableFuture.java:1487)

我做错了什么？

注意:当任何 future 失败时，上面返回的 future 也会失败。接受的答案也应该采取这一点。

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;

另外，如果您想在第一次失败时取消剩余操作，请在 result.completeExceptionally(ex); 之后添加 exec.shutdownNow();。当然，这假设 exec 仅存在于这一计算中。如果没有，您将不得不循环并单独取消每个剩余的 Future。