本文介紹了如何高效地使用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的操作在commonForkJoinPool中執行。輸出含義可能也很明顯：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的文檔是這樣寫的：
返回已啟動但尚未終止的工作線程數。當創建線程以在其他線程被協作阻止時保持并行性時，此方法返回的結果可能不同于getParallism。
您的情況是通過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映射每個輸入的異步任務的文章就介紹到這了，希望我們推薦的答案對大家有所幫助，
