转换列表<完整未来> 为可<列表>

concurrency java java-8 completable-future

2022-08-31 11:52:40

我正在尝试转换为 .这非常有用，因为当您有许多异步任务并且需要获取所有这些任务的结果时。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

在线程 “pool-1-thread-2611” java.lang.StackOverflowError at java.util.concurrent.CompletableFuture.internalComplete（CompletableFuture.java：210） at java.util.concurrent.CompletableFuture$ThenCompose.run（CompletableFuture.java：1487） at java.util.concurrent.CompletableFuture.postComplete（CompletableFuture.java：193） at java.util.concurrent.CompletableFuture.internalComplete（CompletableFuture.java：210） at java.util.concurrent.CompletableFuture$ThenCompose.run（CompletableFuture.java：1487）

我做错了什么？

注意：当任何将来失败时，上述返回的未来将失败。被接受的答案也应该采取这一点。

答案 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...AsyncthenXXXXAsync

此外，不应用于累积到可变容器中。即使它在流是顺序的时可能正常工作，但如果流是并行的，它将失败。要执行可变缩减，请改用。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);execFuture

答案 2

您可以获取Spotify的库并使用allAsList方法。我认为它的灵感来自Guava的Futures.allAsList方法。CompletableFutures

public static <T> CompletableFuture<List<T>> allAsList(
    List<? extends CompletionStage<? extends T>> stages) {

如果您不想使用库，这里有一个简单的实现：

public <T> CompletableFuture<List<T>> allAsList(final List<CompletableFuture<T>> futures) {
    return CompletableFuture.allOf(
        futures.toArray(new CompletableFuture[futures.size()])
    ).thenApply(ignored ->
        futures.stream().map(CompletableFuture::join).collect(Collectors.toList())
    );
}

转换 列表<完整未来> 为可<列表>

转换列表<完整未来> 为可<列表>