如何有效地使用 CompletableFuture 来映射每个输入的异步任务

Question

我想返回包含所有键到值的映射的映射，这些值是 API 对这些键的响应。我正在使用CompletableFutureandGuava为此。以下是我的尝试。是否有其他标准方法可以使用 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()));

    }

Answer 1

这里有一个有趣的行为，我将尽力解释。让我们从简单的开始，让我们暂时忘记这CompletableFuture一点，只需使用 plain 来完成此操作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说：

返回已启动但尚未终止的工作线程数。当创建线程以在其他线程协同阻塞时保持并行性时，此方法返回的结果可能与 getParallelism 不同。

您的情况中的阻塞是通过 发生的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));

join因为工作线程上没有ForJoinPool，可以drop parallelStream。然后我仍然阻止通过以下方式获取结果：

CompletableFuture.allOf(list.toArray(new CompletableFuture[0])).join();

但不会生成补偿线程。因为CompletableFuture.allOf返回 a CompletableFuture<Void>，所以我需要再次流式传输才能获得结果。

不要让.map(CompletableFuture::join)最后一个流操作欺骗您，因为之前的操作CompletableFuture::allOf已经阻塞并等待所有任务完成，所以没有阻塞。