本文介绍了从列表<;CompletableFuture>;转换为CompletableFuture<;列表>;的处理方法,对大家解决问题具有一定的参考价值,需要的朋友们下面随着小编来一起学习吧!
问题描述
我正在尝试将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);
如果其中任何一个失败,那么它就失败了。即使有100万个期货,它也会给出预期的产量。我的问题是:假设有5000多个期货,如果其中任何一个失败了,我会得到一个StackOverflowError
:
我做错了什么?
注意:当任何未来失败时,上述返回的未来也会失败。接受的答案也应考虑这一点。
推荐答案
使用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) {
result.completeExceptionally(ex);
}
}));
return result;
如果您还想在第一次失败时取消剩余的操作,请在result.completeExceptionally(ex);
之后添加exec.shutdownNow();
。当然,这假设exec
只存在于这一次计算中。如果没有,您将不得不循环并逐个取消剩余的每个Future
。 这篇关于从列表<;CompletableFuture>;转换为CompletableFuture<;列表>;的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持!