本文介绍了如何高效地使用CompletableFuture映射每个输入的异步任务的处理方法,对大家解决问题具有一定的参考价值,需要的朋友们下面随着小编来一起学习吧!
问题描述
我希望返回包含所有键到值的映射的映射,该值是对这些键的API响应。为此,我使用CompletableFuture
和Guava
。以下是我的尝试。有没有其他标准的方法来实现与Java 8和线程API相同的功能?
映射为id -> apiResponse(id)
。
public static List<String> returnAPIResponse(Integer key) {
return Lists.newArrayList(key.toString() + " Test");
}
public static void main(String[] args) {
List<Integer> keys = Lists.newArrayList(1, 2, 3, 4);
List<CompletableFuture<SimpleEntry<Integer, List<String>>>> futures = keys
.stream()
.map(key -> CompletableFuture.supplyAsync(
() -> new AbstractMap.SimpleEntry<>(key, returnAPIResponse(key))))
.collect(Collectors.toList());
System.out.println(
futures.parallelStream()
.map(CompletableFuture::join)
.collect(Collectors.toList()));
}
推荐答案
这里有一个有趣的行为,我会尽力解释一下。让我们从简单的开始,让我们暂时忘记CompletableFuture
,简单地使用一个简单的parallelStream
,并添加一个较小的调试步骤:
List<Integer> keys = Lists.newArrayList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16);
Map<Integer, List<String>> result =
keys.parallelStream()
.map(x -> new AbstractMap.SimpleEntry<>(x, returnAPIResponse(x)))
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
System.out.println("parallelism : " + pool.getParallelism() + " current : " + pool.getPoolSize());
在我的机器上,打印:
parallelism : 11 current : 11
我假设您已经知道parallelStream
的操作在common
ForkJoinPool
中执行。输出含义可能也很明显:11 threads
可用且全部已使用。我现在稍微修改一下您的示例:
List<Integer> keys = Lists.newArrayList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16);
ForkJoinPool pool = ForkJoinPool.commonPool();
ExecutorService supplyPool = Executors.newFixedThreadPool(2);
Map<Integer, List<String>> result =
keys.parallelStream()
.map(x -> CompletableFuture.supplyAsync(
() -> new AbstractMap.SimpleEntry<>(x, returnAPIResponse(x)),
supplyPool
))
.map(CompletableFuture::join)
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
System.out.println("parallelism : " + pool.getParallelism() + " current : " + pool.getPoolSize());
这实际上只是一个重要的更改,我将让您的supplyAsync
在它自己的线程池中运行;其余的都是一样的。运行此命令后,会发现:parallelism : 11 current : 16
惊喜。创建了比我们想要的更多的线程吗?那么getPoolSize
的文档是这样写的:
您的情况是通过map(CompletableFuture::join)
阻止的。您已经有效地阻止了ForkJoinPool
中的一个工作线程,它通过旋转另一个线程来补偿这一点。
如果您不想受到这样的惊喜:
List<CompletableFuture<AbstractMap.SimpleEntry<Integer, List<String>>>> list =
keys.stream()
.map(x -> CompletableFuture.supplyAsync(
() -> new AbstractMap.SimpleEntry<>(x, returnAPIResponse(x)),
supplyPool
))
.collect(Collectors.toList());
CompletableFuture.allOf(list.toArray(new CompletableFuture[0])).join();
Map<Integer, List<String>> result =
list.stream()
.map(CompletableFuture::join)
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
因为ForJoinPool
的工作线程上没有join
,所以可以删除parallelStream
。然后我仍然屏蔽通过:
CompletableFuture.allOf(list.toArray(new CompletableFuture[0])).join();
但不会生成补偿线程。因为CompletableFuture.allOf
返回CompletableFuture<Void>
,所以我需要再次串流才能获得结果。
不要让上一个流操作中.map(CompletableFuture::join)
欺骗您,因为前一个CompletableFuture::allOf
已经阻止并等待所有任务完成。
这篇关于如何高效地使用CompletableFuture映射每个输入的异步任务的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持!