Java 8的forEachOrdered()和sequential()方法的区别？

Question

我正在研究 java 8 并行流，并且希望以某种顺序打印并行流中的元素（例如插入顺序、反向顺序或顺序）。

为此我尝试了以下代码：

        System.out.println("With forEachOrdered:");
        listOfIntegers
            .parallelStream()
            .forEachOrdered(e -> System.out.print(e + " "));
        System.out.println("");

        System.out.println("With Sequential:");
        listOfIntegers.parallelStream()
                    .sequential()
                    .forEach(e -> System.out.print(e + " "));

对于这两个，我得到了相同的输出如下：

With forEachOrdered:
1 2 3 4 5 6 7 8 
With Sequential:
1 2 3 4 5 6 7 8 

从 api 文档中，我可以看到：

forEachOrdered -> 这是一个终端操作。

和

sequential -> 这是一个中间操作。

所以我的问题是哪个更好用？ 在哪些情况下，应该优先选择一个？

Answer 1

listOfIntegers.parallelStream().sequential().forEach() 创建一个并行的Stream，然后将其转换为一个顺序的Stream，因此您不妨改用listOfIntegers.stream().forEach()，并首先获得一个顺序的Stream。

listOfIntegers.parallelStream().forEachOrdered(e -> System.out.print(e + " ")) 对并行的Stream 执行操作，但保证元素将按照Stream 的遇到顺序被消耗（如果Stream 具有定义的遇到顺序）。但是，它可以在多个线程上执行。

我没有看到使用listOfIntegers.parallelStream().sequential() 的理由。如果你想要一个连续的Stream，为什么要先创建一个并行的Stream？

Answer 2

你问的是一个误导性的问题，首先你问的是：

 .parallelStream()
 .forEachOrdered(...)

这将创建一个并行流，但元素将被按顺序使用。如果你像这样添加map 操作：

.map(...)
.parallelStream()
.forEachOrdered(...)

这将使map 的操作非常有限（从并行处理的角度来看），因为线程必须等待遇到的所有其他元素才能被处理（由forEachOrdered 消耗）。这涉及无状态操作。

另一方面，如果您有 有状态操作，例如：

.parallelStream()
.map()
.sorted()
.// other operations

由于sorted 是有状态的，无状态操作在它之前从并行处理中的好处会更大。发生这种情况是因为sorted 必须从 Stream 中收集所有元素，并且线程不必“等待”（在forEachOrdered）以遇到顺序的元素。

第二个例子：

listOfIntegers.parallelStream()
                .sequential()
                .forEach(e -> System.out.print(e + " "))

您基本上是说打开并行然后将其关闭。流是由终端操作驱动的，所以即使你这样做：

 .map...
 .filter...
 .parallel()
 .map...
 .sequential

这意味着整个管道将按顺序执行，而不是某些部分是并行的，而另一个是顺序的。您还依赖于 forEach 保留顺序这一事实，并且可能目前确实如此，但可能在以后的版本中，因为您说您不关心顺序（首先使用 forEach )，会有一个内部的元素洗牌。

Answer 3

流管道可以按顺序或并行执行。这种执行模式是流的属性。流是通过初始选择顺序或并行执行来创建的。例如，Collection.stream() 创建一个顺序流，Collection.parallelStream() 创建一个并行流。这种执行模式的选择可以通过BaseStream.sequential() 或BaseStream.parallel() 方法进行修改。

所以没有必要使用：

listOfIntegers.parallelStream().sequential()

你只能使用：

listOfIntegers.stream()

如果您正在创建parallel stream，则流的元素可能由不同的线程处理。 forEach 和 forEachOrdered 之间的区别在于 forEach 将允许以任何顺序处理并行流的任何元素，而 forEachOrdered 将始终按照它们在原始流中出现的顺序处理并行流的元素。当使用parallelStream() 和forEachOrdered 时，这是一个很好的例子，说明如何利用多个内核并仍然保持输出的顺序。请注意，forEachOrdered 强制以有序方式迭代流的元素。但是，在 forEachOrdered 之前链接的任何操作仍将并行发生，因为流是并行流。

Oracle 没有准确记录当您在管道中多次更改流执行模式时会发生什么。不清楚是最后一次更改重要还是调用parallel() 后调用的操作可以并行执行，调用sequential() 后调用的操作将按顺序执行。