java - 为什么在 Java 流中 flatMap() 之后的 filter() 是 "not completely"惰性的？

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

从这里我看到，在第一种情况下，stream 的行为确实很懒——我们使用 findFirst()，所以一旦我们有了第一个元素，我们的过滤 lambda 就不会被调用。然而，在使用 flatMap 的第二种情况下，我们看到尽管找到了满足过滤条件的第一个元素(它只是任何第一个元素，因为 lambda 总是返回 true)，但流的其他内容仍在被馈送通过过滤功能。

我试图理解为什么它的行为是这样的，而不是像第一种情况那样在计算第一个元素后放弃。任何有用的信息将不胜感激。

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);
                    }
                }
            };
        }
    };
}

前进一项的方法最终会在子流上调用 forEach ，而没有任何提前终止的可能性，并且 flatMap 方法开头的注释甚至告诉我们关于这个缺失的功能。

由于这不仅仅是一个优化，因为它意味着当子流无限时代码会简单地中断，我希望开发人员很快证明他们“可以做得更好”......

<小时/>

为了说明其含义，虽然 Stream.iterate(0, i->i+1).findFirst() 按预期工作，但 Stream.of("").flatMap( x->Stream.iterate(0, i->i+1)).findFirst() 最终将陷入无限循环。

关于规范，大部分可以在中找到

chapter “Stream operations and pipelines” of the package specification :

…

Intermediate operations return a new stream. They are always lazy;

…

… Laziness also allows avoiding examining all the data when it is not necessary; for operations such as "find the first string longer than 1000 characters", it is only necessary to examine just enough strings to find one that has the desired characteristics without examining all of the strings available from the source. (This behavior becomes even more important when the input stream is infinite and not merely large.)

…

Further, some operations are deemed short-circuiting operations. An intermediate operation is short-circuiting if, when presented with infinite input, it may produce a finite stream as a result. A terminal operation is short-circuiting if, when presented with infinite input, it may terminate in finite time. Having a short-circuiting operation in the pipeline is a necessary, but not sufficient, condition for the processing of an infinite stream to terminate normally in finite time.

很明显，短路操作不能保证有限时间终止，例如当过滤器不匹配任何项目时，处理无法完成，但是通过简单地忽略操作的短路性质来不支持在有限时间内终止的实现远远超出了规范。