如何高效地使用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已经阻止并等待所有任务完成。