java - 为什么 flatMap() 之后的 filter() 在 Java 流中是 "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.

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