了解 Java 8 和 Java 9 中的顺序与并行流拆分器

Question

一个乍一看并不简单的关于拆分器的问题。

在流中，.parallel() 改变了流处理的行为。但是，我期望从顺序流和并行流创建的拆分器是相同的。例如，通常在顺序流中，.trySplit() 永远不会被调用，而在并行流中，它会被调用，以便将拆分拆分器移交给另一个线程。

stream.spliterator() 与 stream.parallel().spliterator() 之间的差异：

1. 它们可能具有不同的特征：

   Stream.of(1L, 2L, 3L).limit(2);            // ORDERED
   Stream.of(1L, 2L, 3L).limit(2).parallel(); // SUBSIZED, SIZED, ORDERED
   

这里讨论的似乎是另一个废话流拆分器特征策略（并行计算似乎更好）：Understanding deeply spliterator characteristics in java 8 and java 9

1. 他们在使用.trySplit()进行拆分方面可能有不同的行为：

   Stream.of(1L, 2L, 3L);                     // NON NULL
   Stream.of(1L, 2L, 3L).limit(2);            // NULL
   Stream.of(1L, 2L, 3L).limit(2).parallel(); // NON NULL
   

为什么最后两个有不同的行为？如果我愿意，为什么我不能拆分顺序流？ （例如，丢弃其中一个拆分以进行快速处理可能很有用）。

1. 将拆分器转换为流时的重大影响：

   spliterator = Stream.of(1L, 2L, 3L).limit(2).spliterator();
   stream = StreamSupport.stream(spliterator, true); // No parallel processing!
   

在这种情况下，拆分器是从禁用拆分功能的顺序流创建的（.trySplit() 返回 null）。稍后，需要转换回流时，该流将无法从并行处理中受益。可惜了。

大问题：作为一种解决方法，在调用.spliterator() 之前总是将流转换为并行的主要影响是什么？

// Supports activation of parallel processing later
public static <T> Stream<T> myOperation(Stream<T> stream) {
    boolean isParallel = stream.isParallel();
    Spliterator<T> spliterator = stream.parallel().spliterator();
    return StreamSupport.stream(new Spliterator<T>() {
        // My implementation of the interface here (omitted for clarity)
    }, isParallel).onClose(stream::close);
}

// Now I have the option to use parallel processing when needed:
myOperation(stream).skip(1).parallel()...

Answer 1

这不是拆分器的一般属性，而只是封装流管道的包装拆分器。

当您在从拆分器生成且没有链接操作的流上调用 spliterator() 时，您将获得可能支持或不支持 trySplit 的源拆分器，无论流 parallel状态。

ArrayList<String> list = new ArrayList<>();
Collections.addAll(list, "foo", "bar", "baz");
Spliterator<String> sp1 = list.spliterator(), sp2=list.stream().spliterator();
// true
System.out.println(sp1.getClass()==sp2.getClass());
// not null
System.out.println(sp2.trySplit());

同样

Spliterator<String> sp = Stream.of("foo", "bar", "baz").spliterator();
// not null
System.out.println(sp.trySplit());

但是，只要在调用spliterator() 之前链接操作，您就会得到一个包装流管道的拆分器。现在，可以实现执行相关操作的专用拆分器，例如 LimitSpliterator 或 MappingSpliterator，但这还没有完成，因为将流转换回拆分器已被视为最后的手段，而另一个终端操作不适合，不是高优先级用例。相反，您将始终获得单个实现类的实例，该实例试图将流管道实现的内部工作转换为拆分器 API。

对于有状态的操作，这可能很复杂，最值得注意的是，sorted、distinct 或 skip&limit 用于非SIZED 流。对于琐碎的无状态操作，例如map 或filter，提供支持会容易得多，甚至remarked in a code comment

在第一次操作时绑定到管道助手的拆分器的抽象包装拆分器。 此拆分器不是后期绑定的，并且在首次操作时将绑定到源拆分器。 如果存在有状态操作，则无法拆分从顺序流生成的包装拆分器。

…

   // @@@ Detect if stateful operations are present or not
   //     If not then can split otherwise cannot

   /**
    * True if this spliterator supports splitting
    */
   final boolean isParallel;

但目前似乎还没有实现这种检测，所有中间操作都被视为有状态操作。

Spliterator<String> sp = Stream.of("foo", "bar", "baz").map(x -> x).spliterator();
// null
System.out.println(sp.trySplit());

---

当您尝试通过始终调用 parallel 来解决此问题时，当流管道仅包含无状态操作时不会产生任何影响。但是当有一个有状态的操作时，它可能会显着改变行为。例如，当您有一个sorted 步骤时，所有元素都必须经过缓冲和排序，然后才能使用第一个元素。对于并行流，它可能会使用parallelSort，即使您从未调用过trySplit。