并行运算符如何更改后续流元素上的流？

Question

在代码下面执行

List.of(1, 2, 3, 4).stream()
    .map(
        integer -> {
          System.out.println(
              "Before parallel operator : " + Thread.currentThread().getName() + " : " + integer);
          return integer * 2;
        })
    .parallel()
    .map(
        integer -> {
          System.out.println(
              " After parallel operator : " + Thread.currentThread().getName() + " : " + integer);
          return integer * 2;
        })
    .forEach(
            integer -> {
          System.out.println(" For Each : " + Thread.currentThread().getName() + " : " + integer);
        });

输出：

    Before parallel operator : main : 3
Before parallel operator : ForkJoinPool.commonPool-worker-19 : 2
Before parallel operator : ForkJoinPool.commonPool-worker-23 : 1
Before parallel operator : ForkJoinPool.commonPool-worker-5 : 4
 After parallel operator : main : 6
 After parallel operator : ForkJoinPool.commonPool-worker-23 : 2
 After parallel operator : ForkJoinPool.commonPool-worker-19 : 4
 After parallel operator : ForkJoinPool.commonPool-worker-5 : 8
 For Each : ForkJoinPool.commonPool-worker-19 : 8
 For Each : main : 12
 For Each : ForkJoinPool.commonPool-worker-23 : 4
 For Each : ForkJoinPool.commonPool-worker-5 : 16

除了3号元素，其他都是平行的吗？想了解并行运算符在后续调用中行为吗？

并行运算符在哪里开始，并行如何继续？

Answer 1

在调用终端操作（例如 forEach 或 collect）之前，不会处理流，稍后会对此进行更多介绍。因此，回答您的问题，“并行运算符在哪里启动，并行性如何继续？

文档怎么说？

文件中明确说明了这一点：

可以使用基流修改流的模式。sequential（）和基流。parallel（）操作。最新的顺序或并行模式设置适用于整个流管道的执行

一个小演示

现在考虑以下代码（原谅我的System.out，这是出于演示目的）。如果我们在并行和顺序之间切换，整个管道都会发生变化，而不仅仅是后续运算符。

System.out.println("=== Creating stream s1 as 1,2,3,4");
var s1 = List.of(1, 2, 3, 4).stream();
System.out.println("s1 is parallel? " + s1.isParallel());

System.out.println("=== s2 results of applying map to s1");
var s2 = s1.map(integer -> integer * 2);
System.out.println("s1 is parallel? " + s1.isParallel());
System.out.println("s2 is parallel? " + s2.isParallel());

System.out.println("=== s3 results of applying parallel to s2");
var s3 = s2.parallel();
System.out.println("s1 is parallel? " + s1.isParallel());
System.out.println("s2 is parallel? " + s2.isParallel());
System.out.println("s3 is parallel? " + s3.isParallel());

System.out.println("=== s4 results of applying map to s3");
var s4 = s3.map(integer -> integer * 2);
System.out.println("s1 is parallel? " + s1.isParallel());
System.out.println("s2 is parallel? " + s2.isParallel());
System.out.println("s3 is parallel? " + s3.isParallel());
System.out.println("s4 is parallel? " + s4.isParallel());

System.out.println("=== s5 results of applying sequential to s4");
var s5 = s4.sequential();
System.out.println("s1 is parallel? " + s1.isParallel());
System.out.println("s2 is parallel? " + s2.isParallel());
System.out.println("s3 is parallel? " + s3.isParallel());
System.out.println("s4 is parallel? " + s4.isParallel());
System.out.println("s5 is parallel? " + s5.isParallel());

这将输出以下内容：

=== Creating stream s1 as 1,2,3,4
s1 is parallel? false
=== s2 results of applying map to s1
s1 is parallel? false
s2 is parallel? false
=== s3 results of applying parallel to s2
s1 is parallel? true
s2 is parallel? true
s3 is parallel? true
=== s4 results of applying map to s3
s1 is parallel? true
s2 is parallel? true
s3 is parallel? true
s4 is parallel? true
=== s5 results of applying sequential to s4
s1 is parallel? false
s2 is parallel? false
s3 is parallel? false
s4 is parallel? false
s5 is parallel? false

现在，当您调用诸如forEach或collect之类的终端操作符时，它将在处理过程中只考虑顺序流，即使在中间在中间。如文档所述，最新应用的模式用于整个管道。

这有什么用？

你可能会问。可以通过与终端操作员“打破”管道来更改管道中间的行为。例如，以您的示例为例，如果我们在第一个映射之后立即应用 collect，则第一个映射将按顺序执行，然后并行将仅应用于后续运算符，但实际上，这现在是一个不同的管道，因为所有内容都是在列表的中间收集的。

java prettyprint-override">List.of(1, 2, 3, 4).stream()
    .map(integer -> {
        System.out.println("Before stream : " + Thread.currentThread().getName() + " : " + integer);
        return integer * 2;
    })
    .collect(Collectors.toList())
    .stream()
    .parallel()
    .map(integer -> {
        System.out.println("After parallel stream : " + Thread.currentThread().getName() + " : " + integer);
        return integer * 2;
    })
    .forEach(integer -> System.out.println("For Each : " + Thread.currentThread().getName() + " : " + integer));

这将输出如下内容:

Before stream : main : 1
Before stream : main : 2
Before stream : main : 3
Before stream : main : 4
After parallel stream : main : 6
After parallel stream : ForkJoinPool.commonPool-worker-23 : 2
After parallel stream : ForkJoinPool.commonPool-worker-5 : 4
After parallel stream : ForkJoinPool.commonPool-worker-19 : 8
For Each : ForkJoinPool.commonPool-worker-19 : 16
For Each : ForkJoinPool.commonPool-worker-5 : 8
For Each : ForkJoinPool.commonPool-worker-23 : 4
For Each : main : 12

注意第一个< code>map是如何顺序执行的，而其余的操作符是并行执行的。

诸如RxJava之类的可观察流实现与观测运算符有不同的用法，但它们也是一种完全不同的做事方式。