Java parallelstream 在使用 newCachedThreadPool() 时未使用最佳线程数

Question

我做了两个独立的数据库并行读取实现。第一个实现是将 ExecutorService 与 newCachedThreadPool() 构造函数和 Futures 一起使用:我只需调用一个为每个读取案例返回 future 的调用，然后在完成所有调用后调用get() 在他们身上。此实现工作正常并且速度足够快。

第二种实现是使用并行流。当我将并行流调用放入同一个 ExecutorService 池时，它的工作速度几乎慢 5 倍，而且它似乎没有使用我希望的那么多线程。当我将它放入 ForkJoinPool pool = new ForkJoinPool(50) 时，它的运行速度与之前的实现一样快。

我的问题是:

为什么并行流在 newCachedThreadPool 版本中未充分利用线程？

这是第二个实现的代码(我没有发布第一个实现，因为无论如何它都能正常工作):

private static final ExecutorService pool = Executors.newCachedThreadPool();

final List<AbstractMap.SimpleImmutableEntry<String, String>> simpleImmutableEntryStream =
                personIdList.stream().flatMap(
                        personId -> movieIdList.stream().map(
                                movieId -> new AbstractMap.SimpleImmutableEntry<>(personId, movieId))).collect(Collectors.toList());

final Future<Map<String, List<Summary>>> futureMovieSummaryForPerson = pool.submit(() -> {
      final Stream<Summary> summaryStream = simpleImmutableEntryStream.parallelStream().map(
            inputPair -> {
                   return FeedbackDao.find(inputPair.getKey(), inputPair.getValue());
            }).filter(Objects::nonNull);
return summaryStream.collect(Collectors.groupingBy(Summary::getPersonId));
});

Answer 1

这与如何ForkJoinTask.fork有关已实现，如果当前线程来自 ForkJoinPool，它将使用相同的池来提交新任务，但如果不是，它将使用本地计算机和此处的处理器总数的公共(public)池当您使用 Executors.newCachedThreadPool() 创建池时，该池创建的线程不会被识别为来自 ForkJoinPool，因此它使用公共(public)池。

下面是它的实现方式，应该能帮助你更好地理解:

public final ForkJoinTask<V> fork() {
    Thread t;
    if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
        ((ForkJoinWorkerThread)t).workQueue.push(this);
    else
        ForkJoinPool.common.externalPush(this);
    return this;
}

池 Executors.newCachedThreadPool() 创建的线程将不是 ForkJoinWorkerThread 类型，因此它将使用池大小未优化的公共(public)池来提交新任务。