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

编程问答 2022-05-30

问题描述

我想返回由所有键到值的映射组成的映射，作为对这些键的 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,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,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，所以可以删除 parallelStream。然后我仍然阻止通过以下方式获得结果：

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

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

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

asynchronouscompletable-futureguavajavajavamultithreadingmultithreading