java - 为什么 stream.spliterator() 的 tryAdvance 可能会将项目累积到缓冲区中？

Question

获取 Spliterator来自Stream管道可能会返回 StreamSpliterators.WrappingSpliterator 的一个实例.例如，获取以下 Spliterator :

Spliterator<String> source = new Random()
            .ints(11, 0, 7) // size, origin, bound
            .filter(nr -> nr % 2 != 0)
            .mapToObj(Integer::toString)
            .spliterator();

鉴于上述Spliterator<String> source ，当我们通过 tryAdvance (Consumer<? super P_OUT> consumer) 单独遍历元素时Spliterator的方法| ，在本例中是 StreamSpliterators.WrappingSpliterator 的一个实例，它会首先将项目累积到内部缓冲区中，然后再使用这些项目，正如我们在 StreamSpliterators.java#298 中看到的那样.从简单的角度来看，doAdvance()首先将项目插入 buffer然后它获取下一个项目并将其传递给 consumer.accept (…) .

public boolean tryAdvance(Consumer<? super P_OUT> consumer) {
    boolean hasNext = doAdvance();
    if (hasNext)
        consumer.accept(buffer.get(nextToConsume));
    return hasNext;
}

但是，我没有弄清楚这个 buffer 的必要性.

在这种情况下，为什么 consumer tryAdvance 的参数不是简单的当终端使用 Sink 管道的？

Answer 1

请记住，这是 Spliterator由 public 返回方法 Stream.spliterator() ，因此无法对调用者做出任何假设(只要它在契约(Contract)内)。

tryAdvance方法可能会为流的每个元素调用一次，并再次检测流的末尾，好吧，实际上，即使到达末尾，它也可能被调用任意次数。并且不能保证调用者将始终传递同一个消费者。

要将消费者直接传递给源拆分器而无需缓冲，您必须编写一个将执行所有管道阶段的消费者，即调用映射函数并使用其结果或测试谓词，如果为负则不调用下游消费者等等。传递给源拆分器的消费者也将负责通知 WrappingSpliterator不知何故，一个值被过滤器拒绝为源拆分器的 tryAdvance方法仍然返回 true在这种情况下，必须重复该操作。

作为Eugene correctly mentioned ，这是一刀切的实现，不考虑有多少或什么样的流水线阶段。构成这样一个消费者的成本可能很重，而且可能必须为每个 tryAdvance 重新应用调用，读取每个流元素，例如当不同的消费者被传递给tryAdvance或者当平等检查不起作用时。请记住，消费者通常作为 lambda 表达式实现，而 lambda 表达式生成的实例的身份或相等性是未指定的。

所以 tryAdvance实现通过在第一次调用时仅组合一个消费者实例来避免这些成本，该实例将始终将元素存储到相同的缓冲区中，如果没有被过滤器拒绝，也会在第一次调用时分配。请注意，在正常情况下，缓冲区只会容纳一个元素。阿法克 flatMap is the only operation that may push more elements to the buffer .但请注意 flatMap 的这种非惰性行为的存在这也是为什么需要这种缓冲策略的原因，至少在 flatMap 时是这样。参与其中，以确保Spliterator public 分发的实现方法将履行在 tryAdvance 的一次调用期间最多将一个元素传递给消费者的契约。 .

相比之下，当您调用 forEachRemaining 时，这些问题都不存在。只有一个 Consumer整个操作过程中的实例和 flatMap 的非惰性也没关系，因为所有元素都会被消耗掉。因此，只要之前没有tryAdvance，就会尝试非缓冲传输。进行的调用可能导致某些元素的缓冲:

     public void forEachRemaining(Consumer<? super P_OUT> consumer) {
         if (buffer == null && !finished) {
             Objects.requireNonNull(consumer);
             init();

             ph.wrapAndCopyInto((Sink<P_OUT>) consumer::accept, spliterator);
             finished = true;
         }
         else {
             do { } while (tryAdvance(consumer));
         }
     }

如你所见，只要buffer尚未初始化，即没有先前的 tryAdvance已调用电话，consumer::accept 被绑定(bind)为Sink并进行了完整的直接传输。