Warum ist filter () nach flatMap () in Java-Streams "nicht vollständig" faul?

Ich habe den folgenden Beispielcode:

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

Die Ausgabe ist wie folgt:

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

Von hier aus sehe ich, dass sich der erste Fall streamwirklich träge verhält - wir verwenden ihn findFirst(), sobald wir das erste Element haben, wird unser Filter-Lambda nicht aufgerufen. Im zweiten Fall, in dem flatMaps verwendet wird, sehen wir jedoch, dass trotz des ersten Elements, das die Filterbedingung erfüllt (es ist nur jedes erste Element, da Lambda immer true zurückgibt), weitere Inhalte des Streams weiterhin durch die Filterfunktion eingespeist werden.

Ich versuche zu verstehen, warum es sich so verhält, anstatt aufzugeben, nachdem das erste Element wie im ersten Fall berechnet wurde. Alle hilfreichen Informationen wäre dankbar.

TL; DR, dies wurde in JDK-8075939 behoben und in Java 10 behoben (und in JDK-8225328 auf Java 8 zurückportiert ).

Wenn ReferencePipeline.javawir uns die Implementierung ( ) ansehen, sehen wir die Methode [ Link ]

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

welches für den findFirstBetrieb aufgerufen wird. Das Besondere ist, dass Sie sink.cancellationRequested()die Schleife beim ersten Spiel beenden können. Vergleiche mit [ 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);
                    }
                }
            };
        }
    };
}

Die Methode zum Vorrücken eines Elements ruft forEachden Sub-Stream auf, ohne dass eine frühere Beendigung möglich ist, und der Kommentar am Anfang der flatMapMethode gibt sogar Auskunft über diese fehlende Funktion.

Da dies mehr als nur eine Optimierungssache ist, da dies impliziert, dass der Code einfach bricht, wenn der Substream unendlich ist, hoffe ich, dass die Entwickler bald beweisen, dass sie „besser als das können“…

Um die Auswirkungen zu veranschaulichen Stream.iterate(0, i->i+1).findFirst(), Stream.of("").flatMap(x->Stream.iterate(0, i->i+1)).findFirst()wird es , obwohl es wie erwartet funktioniert, in einer Endlosschleife enden.

In Bezug auf die Spezifikation finden Sie das meiste davon in der

Kapitel „Stream-Operationen und Pipelines“ der Paketspezifikation :

…

Zwischenoperationen geben einen neuen Stream zurück. Sie sind immer faul ;

…

… Durch Faulheit kann auch vermieden werden, dass alle Daten überprüft werden, wenn dies nicht erforderlich ist. Bei Operationen wie "Finden Sie die erste Zeichenfolge mit mehr als 1000 Zeichen" müssen Sie nur so viele Zeichenfolgen untersuchen, dass eine Zeichenfolge mit den gewünschten Eigenschaften gefunden wird, ohne alle von der Quelle verfügbaren Zeichenfolgen zu untersuchen. (Dieses Verhalten wird noch wichtiger, wenn der Eingabestream unendlich und nicht nur groß ist.)

…

Ferner werden einige Operationen als Kurzschlussoperationen angesehen . Eine Zwischenoperation ist kurzgeschlossen, wenn sie bei unendlicher Eingabe einen endlichen Strom erzeugen kann. Eine Terminaloperation ist kurzgeschlossen, wenn sie bei unendlicher Eingabe in endlicher Zeit beendet werden kann. Ein Kurzschlussbetrieb in der Pipeline ist eine notwendige, aber nicht ausreichende Bedingung, damit die Verarbeitung eines unendlichen Stroms normal in endlicher Zeit endet.

Es ist klar, dass eine Kurzschlussoperation keine endliche Zeitbeendigung garantiert, z. B. wenn ein Filter keinem Element entspricht, das die Verarbeitung nicht abschließen kann, sondern eine Implementierung, die keine Beendigung in endlicher Zeit durch einfaches Ignorieren unterstützt Der Kurzschlusscharakter einer Operation liegt weit außerhalb der Spezifikation.

Die Elemente des Eingabestreams werden nacheinander träge verbraucht. Das erste Element 1wird von den beiden flatMaps in den Stream umgewandelt -1, 0, 1, 0, 1, 2, 1, 2, 3, so dass der gesamte Stream nur dem ersten Eingabeelement entspricht. Die verschachtelten Ströme werden von der Pipeline eifrig materialisiert, dann abgeflacht und dann der filterBühne zugeführt . Dies erklärt Ihre Ausgabe.

Das Obige beruht nicht auf einer grundlegenden Einschränkung, aber es würde die Dinge wahrscheinlich viel komplizierter machen, wenn verschachtelte Streams vollständig faul sind. Ich vermute, es wäre eine noch größere Herausforderung, es performant zu machen.

Zum Vergleich erhalten Clojures faule Sequenzen für jede dieser Verschachtelungsebenen eine weitere Umhüllungsschicht. Aufgrund dieser Konstruktion können die Operationen sogar fehlschlagen, StackOverflowErrorwenn die Verschachtelung extrem ausgeführt wird.

In Bezug auf das Brechen mit unendlichen Teilströmen wird das Verhalten von flatMap noch überraschender, wenn man einen Zwischenschluss (im Gegensatz zum Terminal) kurzschließt.

Während das Folgende wie erwartet funktioniert, drucken Sie die unendliche Folge von ganzen Zahlen aus

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

Der folgende Code gibt nur die "1" aus, wird jedoch immer noch nicht beendet:

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

Ich kann mir keine Lektüre der Spezifikation vorstellen, in der das kein Fehler war.

In meiner kostenlosen StreamEx- Bibliothek habe ich die Kurzschlusssammler vorgestellt. Beim Sammeln eines sequentiellen Stroms mit einem kurzgeschlossenen Kollektor (wie MoreCollectors.first()) wird genau ein Element von der Quelle verbraucht. Intern ist es ziemlich schmutzig implementiert: Verwenden einer benutzerdefinierten Ausnahme, um den Kontrollfluss zu unterbrechen. Mit meiner Bibliothek könnte Ihr Beispiel folgendermaßen umgeschrieben werden:

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

Das Ergebnis ist folgendes:

-1
Result: -1

Ist leider .flatMap()nicht faul. Hier finden Sie jedoch eine benutzerdefinierte ProblemumgehungflatMap : Warum .flatMap () in Java 8 und Java 9 so ineffizient (nicht faul) ist

Ich stimme anderen Leuten zu, dass dies ein Fehler ist, der bei JDK-8075939 geöffnet wurde . Und da es noch nicht mehr als ein Jahr später behoben ist. Ich möchte Ihnen empfehlen: 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

Offenlegung: Ich bin der Entwickler von AbacusUtil.

Heute bin ich auch auf diesen Fehler gestoßen. Das Verhalten ist nicht so direkt, da ein einfacher Fall wie unten gut funktioniert, aber ein ähnlicher Produktionscode nicht funktioniert.

 stream(spliterator).map(o -> o).flatMap(Stream::of).flatMap(Stream::of).findAny()

Für Leute, die nicht noch ein paar Jahre auf die Migration zu JDK-10 warten können, gibt es einen alternativen echten Lazy Stream. Parallel wird nicht unterstützt. Es war für die JavaScript-Übersetzung vorgesehen, hat aber für mich funktioniert, da die Benutzeroberfläche dieselbe ist.

StreamHelper basiert auf Sammlungen, aber es ist einfach, Spliterator anzupassen.

https://github.com/yaitskov/j4ts/blob/stream/src/main/java/javaemul/internal/stream/StreamHelper.java