为什么filter（）flatMap（）之后的filter（）在Java流中“不完全”懒散？

Question

我有以下示例代码：

System.out.println(
       "Result: " +
        Stream.of(1, 2, 3)
                .filter(i -> {
                    System.out.println(i);
                    return true;
                })
                .findFirst()
                .get()
);
System.out.println("-----------");
System.out.println(
       "Result: " +
        Stream.of(1, 2, 3)
                .flatMap(i -> Stream.of(i - 1, i, i + 1))
                .flatMap(i -> Stream.of(i - 1, i, i + 1))
                .filter(i -> {
                    System.out.println(i);
                    return true;
                })
                .findFirst()
                .get()
);

1
Result: 1
-----------
-1
0
1
0
1
2
1
2
3
Result: -1

Answer 1

Tl；Dr，这已经在JDK-8075939中解决，并在Java ；10中修复（并在JDK-8225328中回传到Java 8）。

在研究实现(referencePipeline.java)时，我们看到方法[link]

@Override
final void forEachWithCancel(Spliterator<P_OUT> spliterator, Sink<P_OUT> sink) {
    do { } while (!sink.cancellationRequested() && spliterator.tryAdvance(sink));
}

将为findfirst操作调用。需要特别注意的是sink.cancellationRequested()，它允许在第一次匹配时结束循环。比较[link]

@Override
public final <R> Stream<R> flatMap(Function<? super P_OUT, ? extends Stream<? extends R>> mapper) {
    Objects.requireNonNull(mapper);
    // We can do better than this, by polling cancellationRequested when stream is infinite
    return new StatelessOp<P_OUT, R>(this, StreamShape.REFERENCE,
                                 StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT | StreamOpFlag.NOT_SIZED) {
        @Override
        Sink<P_OUT> opWrapSink(int flags, Sink<R> sink) {
            return new Sink.ChainedReference<P_OUT, R>(sink) {
                @Override
                public void begin(long size) {
                    downstream.begin(-1);
                }

                @Override
                public void accept(P_OUT u) {
                    try (Stream<? extends R> result = mapper.apply(u)) {
                        // We can do better that this too; optimize for depth=0 case and just grab spliterator and forEach it
                        if (result != null)
                            result.sequential().forEach(downstream);
                    }
                }
            };
        }
    };
}

因为这不仅仅是一个优化的事情，因为它意味着当子流是无限的时候代码就会中断，我希望开发人员很快证明他们“可以做得更好”…

为了说明其中的含义，虽然stream.iterate(0,I->i+1).findfirst()按预期工作，但stream.of("“）.FlatMap(X->stream.iterate(0,I->i+1)).findfirst()将以无限循环结束。

关于该规范，可以在

…懒惰也允许在没有必要的时候避免检查所有的数据；对于“查找第一个长度超过1000个字符的字符串”这样的操作，只需要检查足够的字符串，以找到具有所需特征的字符串，而不需要检查源中所有可用的字符串。（当输入流是无限大的而不仅仅是大的时，这种行为变得更加重要。）

…

此外，一些操作被认为是短路操作。如果一个中间操作在提供无限的输入时可能产生有限的流，那么它就是短路的。如果一个终端操作在无限输入的情况下可能在有限时间内终止，那么它就是短路操作。在管道中具有短路操作是无限流处理在有限时间内正常终止的必要条件，但不是充分条件。

很明显，短路操作并不能保证有限时间的终止，例如，当一个滤波器不匹配任何处理不能完成的项目时，但是一个通过简单地忽略一个操作的短路性质而不支持有限时间内的任何终止的实现是远远不符合规范的。