从列表&lt；CompletableFuture&gt；转换为CompletableFuture&lt；列表&gt；

本文介绍了从列表&lt；CompletableFuture&gt；转换为CompletableFuture&lt；列表&gt；的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！ 问题描述

我正在尝试将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 = CompletableFuturepletedFuture(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);

如果其中任何一个失败，那么它就失败了。即使有100万个期货，它也会给出预期的产量。我的问题是：假设有5000多个期货，如果其中任何一个失败了，我会得到一个StackOverflowError：

线程"；pool-1-thread-2611"；java.lang.StackOverflowError中出现异常 在… java.util.concurrent.CompletableFuture.internalComplete(CompletableFuture.java:210) 在… 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)

我做错了什么？

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

推荐答案

使用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方法在单独的线程中运行提供给它们的函数。因此，在您的示例中，您将导致在提供的Executor中运行将新项添加到列表中。不需要将轻量级操作填充到缓存的线程执行器中。请不要在没有充分理由的情况下使用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) { resultpleteExceptionally(ex); } })); return result; 如果您还想在第一次失败时取消剩余的操作，请在resultpleteExceptionally(ex);之后添加exec.shutdownNow();。当然，这假设exec只存在于这一次计算中。如果没有，您将不得不循环并逐个取消剩余的每个Future。