Почему filter() после того, как flatMap() "не полностью" ленив в потоках Java?

Ответ 1

TL; DR, это было решено в JDK-8075939 и исправлено в Java 10 (и перенесено в Java 8 в JDK-8225328).

При изучении реализации (ReferencePipeline.java) мы видим метод [ ссылка ]

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

который будет вызываться для операции findFirst. Особая вещь, о которой следует позаботиться, - это sink.cancellationRequested() который позволяет завершить цикл при первом совпадении. Сравнить с [ ссылка ]

@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() окажется в бесконечном цикле.

Что касается спецификации, большинство из них можно найти в

глава "Потоковые операции и конвейеры" спецификации пакета:

...

Промежуточные операции возвращают новый поток. Они всегда ленивы;

...

… Лень также позволяет избежать проверки всех данных, когда в этом нет необходимости; для таких операций, как "найти первую строку длиной более 1000 символов", необходимо только изучить достаточно строк, чтобы найти ту, которая обладает желаемыми характеристиками, без проверки всех строк, доступных из источника. (Такое поведение становится еще более важным, когда входной поток бесконечен, а не просто велик.)

...

Кроме того, некоторые операции считаются операциями с коротким замыканием. Промежуточная операция - это короткое замыкание, если при наличии бесконечного ввода она может привести к конечному потоку в результате. Работа терминала является коротким замыканием, если при наличии бесконечного входа она может завершиться за конечное время. Наличие операции короткого замыкания в конвейере является необходимым, но не достаточным условием для прекращения обработки бесконечного потока в обычном режиме за конечное время.

Понятно, что операция с коротким замыканием не гарантирует конечное завершение времени, например, когда фильтр не соответствует ни одному элементу, обработка не может быть завершена, но реализация, которая не поддерживает любое завершение за конечное время, просто игнорируя характер короткого замыкания операции далеко от спецификации.

Ответ 2

Элементы входного потока потребляются лениво один за другим. Первый элемент 1 преобразуется двумя flatMap в поток -1, 0, 1, 0, 1, 2, 1, 2, 3, так что весь поток соответствует только первому входному элементу. Вложенные потоки с готовностью материализуются по трубопроводу, затем сплющиваются, а затем подаются на стадию filter. Это объясняет ваш результат.

Вышеизложенное не связано с фундаментальным ограничением, но, вероятно, это значительно усложнит задачу получения полной лени для вложенных потоков. Я подозреваю, что это будет еще более сложной задачей, чтобы сделать ее работоспособной. Для сравнения, Clojure lazy seqs получают еще один слой обертывания для каждого такого уровня вложенности. Из-за этой конструкции операции могут даже терпеть неудачу с StackOverflowError, когда вложение выполняется до крайности.

Ответ 3

Что касается поломки с бесконечными подпотоками, то поведение flatMap становится еще более удивительным, когда вы выполняете короткую операцию промежуточного (в отличие от терминала).

В то время как следующее работает, как ожидалось, распечатывает бесконечную последовательность целых чисел

Stream.of("x").flatMap(_x -> Stream.iterate(1, i -> i + 1)).forEach(System.out::println);

следующий код выводит только "1", но все же не завершает:

Stream.of("x").flatMap(_x -> Stream.iterate(1, i -> i + 1)).limit(1).forEach(System.out::println);

Я не могу себе представить, что такое спецификация, в которой это не было ошибкой.

Ответ 4

В моей бесплатной библиотеке StreamEx я представил коллекционеры короткого замыкания. При сборе последовательного потока с короткозамкнутым коллектором (например, MoreCollectors.first()) ровно один элемент потребляется из источника. Внутренне он реализован довольно грязно: с помощью настраиваемого исключения для прерывания потока управления. Используя мою библиотеку, ваш образец можно переписать следующим образом:

System.out.println(
        "Result: " +
                StreamEx.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;
                })
                .collect(MoreCollectors.first())
                .get()
        );

В результате получается следующее:

-1
Result: -1

Ответ 5

Я согласен с другими людьми, это ошибка, открытая в JDK-8075939. И поскольку это все еще не исправлено больше чем через год. Я хотел бы порекомендовать вам: AbacusUtil

N.println("Result: " + Stream.of(1, 2, 3).peek(N::println).first().get());

N.println("-----------");

N.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))
                        .peek(N::println).first().get());

// output:
// 1
// Result: 1
// -----------
// -1
// Result: -1

Раскрытие информации: я разработчик AbacusUtil.

Ответ 6

К сожалению, .flatMap() не ленив. Тем не менее, обходное решение flatMap доступно здесь: Почему .flatMap() настолько неэффективен (не ленив) в java 8 и java 9