java - 了解 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)。稍后，需要转换回流时，该流将无法从并行处理中获益。真可惜。

大问题:作为解决方法，总是在调用之前将流转换为并行的主要影响是什么。分离器()?

// 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，但这尚未完成，因为将流转换回当其他终端操作不适合时，spliterator 被认为是最后的手段，不是高优先级的用例。相反，您将始终获得单个实现类的实例，该实例试图将流管道实现的内部工作转换为拆分器 API。

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

Abstract wrapping spliterator that binds to the spliterator of a pipeline helper on first operation. This spliterator is not late-binding and will bind to the source spliterator when first operated on. A wrapping spliterator produced from a sequential stream cannot be split if there are stateful operations present.

…

   // @@@ 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。