共计 3775 个字符，预计需要花费 10 分钟才能阅读完成。
流可以并行执行，以增加大量输入元素的运行时性能。并行流 ForkJoinPool 通过静态 ForkJoinPool.commonPool() 方法使用公共可用的流。底层线程池的大小最多使用五个线程 – 具体取决于可用物理 CPU 核心的数量：
ForkJoinPool commonPool = ForkJoinPool.commonPool();
System.out.println(commonPool.getParallelism()); // 3
在我的机器上，公共池初始化为默认值为 3 的并行度。通过设置以下 JVM 参数可以减小或增加此值：
-Djava.util.concurrent.ForkJoinPool.common.parallelism=5
集合支持创建并行元素流的方法 parallelStream()。或者，您可以在给定流上调用中间方法 parallel()，以将顺序流转换为并行流。
为了评估并行流的并行执行行为，下一个示例将有关当前线程的信息打印出来：
Arrays.asList(“a1”, “a2”, “b1”, “c2”, “c1”)
.parallelStream()
.filter(s -> {
System.out.format(“filter: %s [%s]\n”,
s, Thread.currentThread().getName());
return true;
})
.map(s -> {
System.out.format(“map: %s [%s]\n”,
s, Thread.currentThread().getName());
return s.toUpperCase();
})
.forEach(s -> System.out.format(“forEach: %s [%s]\n”,
s, Thread.currentThread().getName()));
通过调查调试输出，我们应该更好地理解哪些线程实际用于执行流操作：
filter: b1 [main]
filter: a2 [ForkJoinPool.commonPool-worker-1]
map: a2 [ForkJoinPool.commonPool-worker-1]
filter: c2 [ForkJoinPool.commonPool-worker-3]
map: c2 [ForkJoinPool.commonPool-worker-3]
filter: c1 [ForkJoinPool.commonPool-worker-2]
map: c1 [ForkJoinPool.commonPool-worker-2]
forEach: C2 [ForkJoinPool.commonPool-worker-3]
forEach: A2 [ForkJoinPool.commonPool-worker-1]
map: b1 [main]
forEach: B1 [main]
filter: a1 [ForkJoinPool.commonPool-worker-3]
map: a1 [ForkJoinPool.commonPool-worker-3]
forEach: A1 [ForkJoinPool.commonPool-worker-3]
forEach: C1 [ForkJoinPool.commonPool-worker-2]
如您所见，并行流利用公共中的所有可用线程 ForkJoinPool 来执行流操作。输出在连续运行中可能不同，因为实际使用的特定线程的行为是非确定性的。
让我们通过一个额外的流操作来扩展该示例：
Arrays.asList(“a1”, “a2”, “b1”, “c2”, “c1”)
.parallelStream()
.filter(s -> {
System.out.format(“filter: %s [%s]\n”,
s, Thread.currentThread().getName());
return true;
})
.map(s -> {
System.out.format(“map: %s [%s]\n”,
s, Thread.currentThread().getName());
return s.toUpperCase();
})
.sorted((s1, s2) -> {
System.out.format(“sort: %s <> %s [%s]\n”,
s1, s2, Thread.currentThread().getName());
return s1.compareTo(s2);
})
.forEach(s -> System.out.format(“forEach: %s [%s]\n”,
s, Thread.currentThread().getName()));
结果可能最初看起来很奇怪：
filter: c2 [ForkJoinPool.commonPool-worker-3]
filter: c1 [ForkJoinPool.commonPool-worker-2]
map: c1 [ForkJoinPool.commonPool-worker-2]
filter: a2 [ForkJoinPool.commonPool-worker-1]
map: a2 [ForkJoinPool.commonPool-worker-1]
filter: b1 [main]
map: b1 [main]
filter: a1 [ForkJoinPool.commonPool-worker-2]
map: a1 [ForkJoinPool.commonPool-worker-2]
map: c2 [ForkJoinPool.commonPool-worker-3]
sort: A2 <> A1 [main]
sort: B1 <> A2 [main]
sort: C2 <> B1 [main]
sort: C1 <> C2 [main]
sort: C1 <> B1 [main]
sort: C1 <> C2 [main]
forEach: A1 [ForkJoinPool.commonPool-worker-1]
forEach: C2 [ForkJoinPool.commonPool-worker-3]
forEach: B1 [main]
forEach: A2 [ForkJoinPool.commonPool-worker-2]
forEach: C1 [ForkJoinPool.commonPool-worker-1]
似乎 sort 只在主线程上顺序执行。实际上，sort 在并行流上使用新的 Java 8 方法 Arrays.parallelSort()。如 Javadoc 中所述，如果排序将按顺序或并行执行，则此方法决定数组的长度：
如果指定数组的长度小于最小粒度，则使用适当的 Arrays.sort 方法对其进行排序。
回到 reduce 一节的例子。我们已经发现组合器函数只是并行调用，而不是顺序流调用。让我们看看实际涉及哪些线程：
List<Person> persons = Arrays.asList(
new Person(“Max”, 18),
new Person(“Peter”, 23),
new Person(“Pamela”, 23),
new Person(“David”, 12));
persons
.parallelStream()
.reduce(0,
(sum, p) -> {
System.out.format(“accumulator: sum=%s; person=%s [%s]\n”,
sum, p, Thread.currentThread().getName());
return sum += p.age;
},
(sum1, sum2) -> {
System.out.format(“combiner: sum1=%s; sum2=%s [%s]\n”,
sum1, sum2, Thread.currentThread().getName());
return sum1 + sum2;
});
控制台输出显示累加器和组合器函数在所有可用线程上并行执行：
accumulator: sum=0; person=Pamela; [main]
accumulator: sum=0; person=Max; [ForkJoinPool.commonPool-worker-3]
accumulator: sum=0; person=David; [ForkJoinPool.commonPool-worker-2]
accumulator: sum=0; person=Peter; [ForkJoinPool.commonPool-worker-1]
combiner: sum1=18; sum2=23; [ForkJoinPool.commonPool-worker-1]
combiner: sum1=23; sum2=12; [ForkJoinPool.commonPool-worker-2]
combiner: sum1=41; sum2=35; [ForkJoinPool.commonPool-worker-2]
总之，并行流可以为具有大量输入元素的流带来良好的性能提升。但请记住，某些并行流操作 reduce,collect 需要额外的计算（组合操作），这在顺序执行时是不需要的。
此外，我们了解到所有并行流操作共享相同的 JVM 范围 ForkJoinPool。因此，您可能希望避免实施慢速阻塞流操作，因为这可能会减慢严重依赖并行流的应用程序的其他部分。
