如何立即将任务从一个 ThreadPool 传送到另一个？答案

【问题标题】：How can I immediately pipe tasks from one ThreadPool to another?如何立即将任务从一个 ThreadPool 传送到另一个？
【发布时间】：2019-08-30 17:28:32
【问题描述】：

我有一个输入元素列表，我想将其排入多个线程池。假设这是我的输入：

final List<Integer> ints = Stream.iterate(1, i -> i + 1).limit(100).collect(Collectors.toList());

这是我希望元素一个接一个地运行的三个函数：

final Function<Integer, Integer> step1 =
        value -> { // input from the ints list
            return value * 2;
        };

final Function<Integer, Double> step2 =
        value -> { // input from the previous step1
            return (double) (value * 2); //
        };

final Function<Double, String> step3 =
        value -> { // input from the previous step2
            return "Result: " + value * 2;
        };

这些将是每个步骤的池：

final ExecutorService step1Pool = Executors.newFixedThreadPool(4);
final ExecutorService step2Pool = Executors.newFixedThreadPool(3);
final ExecutorService step3Pool = Executors.newFixedThreadPool(1);

我希望每个元素都贯穿step1Pool 并应用step1。一旦一个元素完成，它的结果应该以step2pool 结尾，这样step2 就可以在这里应用。只要step2Pool 中的某事完成，就应该应该应用在step3Pool 和step3 中排队。在我的主线程上，我想等到我得到来自step3 的所有结果。每个元素的处理顺序没关系。只是它们都在正确的线程池上运行step1 -> step2 -> step3。

基本上我想并行化Stream.map，立即将每个结果推送到下一个队列并等到我完成从我的最后一个线程池中得到了ints.size() 结果。

有没有简单的Java实现方法？

【问题讨论】：

标签： java threadpool executorservice

【解决方案1】：

我相信 CompletableFuture 会在这里为您提供帮助！

List<CompletableFuture<String>> futures = ints.stream()
            .map(i -> CompletableFuture.supplyAsync(() -> step1.apply(i), step1Pool)
                    .thenApplyAsync(step2, step2Pool)
                    .thenApplyAsync(step3, step3Pool))
            .collect(Collectors.toList());
List<String> result = futures.stream()
            .map(CompletableFuture::join)
            .collect(Collectors.toList());

【讨论】：

【解决方案2】：

为此更好地使用流：

List<String> stringList = Stream.iterate(1, i -> i + 1)
                .limit(100)
                .parallel()
                .map(step1)
                .map(step2)
                .map(step3)
                .collect(Collectors.toList());

【讨论】：