本文介绍了如何高效地使用CompletableFuture映射每个输入的异步任务的处理方法,对大家解决问题具有一定的参考价值,需要的朋友们下面随着小编来一起学习吧!

问题描述

我希望返回包含所有键到值的映射的映射,该值是对这些键的API响应。为此,我使用CompletableFutureGuava。以下是我的尝试。有没有其他标准的方法来实现与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的文档是这样写的:

您的情况是通过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映射每个输入的异步任务的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持!

10-11 13:34