只使用一个线程的Java并行流？

Question

我正在使用最新的Java8 lambdas和并行流处理数据。我的代码如下：

ForkJoinPool forkJoinPool = new ForkJoinPool(10);
List<String> files = Arrays.asList(new String[]{"1.txt"}); 
List<String> result = forkJoinPool.submit(() ->
    files.stream().parallel()
        .flatMap(x -> stage1(x)) //at this stage we add more elements to the stream
        .map(x -> stage2(x))
        .map(x -> stage3(x))
        .collect(Collectors.toList())
).get();

流以一个元素开始，但在第二阶段会添加更多的元素。我的假设是这个流应该并行运行，但在这种情况下只使用一个工作线程。

如果我从2个元素开始（即我在初始列表中添加第二个元素），那么就会产生2个线程来处理流，依此类推...如果我没有显式地将流提交给ForkJoinPool,也会发生这种情况。

问题是：它是记录在案的行为还是可能在实现中改变？有没有什么方法可以控制这种行为，允许更多的线程，而不管初始列表是什么？

Answer 1

您所观察的是特定于实现的行为，而不是指定的行为。

当前的JDK8实现查看最外层流的spliterator，并将其用作分割并行工作负载的基础。由于该示例在原始源流中只有单个元素，因此无法对其进行拆分，流以单线程方式运行。这对于flatmap返回零个、一个或几个元素的常见（但绝不是唯一）情况很好，但对于返回大量元素的情况，它们都是按顺序处理的。实际上，flatmap函数返回的流被强制进入顺序模式。请参阅referencePipeline.java的第270行。

要做的“显而易见”的事情是使这个流平行，或者至少不强迫它是顺序的。这可能会也可能不会改善事情。最有可能的是，它会改善一些事情，但会使其他事情变得更糟。这里当然需要一个更好的政策，但我不确定它会是什么样子。

还要注意，通过向并行流提交运行管道的任务，强制并行流在您选择的fork-join池中运行的技术也是特定于实现的行为。它在JDK8中是这样工作的，但将来可能会改变。