为什么 parallelStream 不使用整个可用的并行度？答案

【问题标题】：Why does parallelStream not use the entire available parallelism?为什么 parallelStream 不使用整个可用的并行度？
【发布时间】：2015-06-12 22:15:15
【问题描述】：

我创建了一个自定义 ForkJoinPool，并行度为 25。

customForkJoinPool = new ForkJoinPool(25);

我有一个包含 700 个文件名的列表，我使用这样的代码从 S3 并行下载文件并将它们转换为 Java 对象：

customForkJoinPool.submit(() -> {
   return fileNames
     .parallelStream()
     .map((fileName) -> {
        Logger log = Logger.getLogger("ForkJoinTest");
        long startTime = System.currentTimeMillis();
        log.info("Starting job at Thread:" + Thread.currentThread().getName());
        MyObject obj = readObjectFromS3(fileName);
        long endTime = System.currentTimeMillis();
        log.info("completed a job with Latency:" + (endTime - startTime));
        return obj;
     })
     .collect(Collectors.toList);
   });
});

当我查看日志时，我看到只使用了 5 个线程。并行度为 25，我预计这将使用 25 个线程。下载文件并将其转换为对象的平均延迟约为 200 毫秒。我错过了什么？

可能更好的问题是并行流如何计算在为其创建线程之前将原始列表拆分多少？在这种情况下，它似乎决定将其拆分 5 次并停止。

【问题讨论】：

这个问题不适合 fork-join。对于初学者，您需要使用ManagedBlocker，因为您的任务是阻塞的。但是，由于它是 I/O，并且不能递归整除，因此仍然不太合适。

标签： java multithreading java-8 java-stream fork-join

【解决方案1】：

您为什么要使用ForkJoinPool？它适用于 CPU 密集型任务，其子任务太快而无法保证单独调度。您的工作负载受 IO 限制，延迟为 200 毫秒，单个调度开销可以忽略不计。

使用Executor：

import static java.util.stream.Collectors.toList;
import static java.util.concurrent.CompletableFuture.supplyAsync;

ExecutorService threads = Executors.newFixedThreadPool(25);

List<MyObject> result = fileNames.stream()
        .map(fn -> supplyAsync(() -> readObjectFromS3(fn), threads))
        .collect(toList()).stream()
        .map(CompletableFuture::join)
        .collect(toList());

【讨论】：

谢谢。这回答了我的实际问题。虽然我仍然想知道 parallelStream 如何计算出要拆分多少以及何时停止 :)
@Aishwar - 在这种情况下，它不会......这就是问题所在。看我的回答。

【解决方案2】：

我认为答案就在这个...来自ForkJoinPool javadoc。

“池尝试通过动态添加、挂起或恢复内部工作线程来维持足够的活动（或可用）线程，即使某些任务暂停等待加入其他任务。但是，没有这样的调整在遇到阻塞的 I/O 或其他非托管同步时得到保证。”

在您的情况下，下载将执行阻塞 I/O 操作。

【讨论】：