Java Stream Gatherers – Schreibe deine eigenen Stream-Operationen!

Die Java-Stream-API wurde mit Java 8 im März 2014 veröffentlicht und hat uns ein grundlegend neues Werkzeug an die Hand gegeben, um Datenströme zu verarbeiten.

Allerdings führt der begrenzte Satz an intermediären Operationen – filter, map, flatMap, mapMulti, distinct, sorted, peek, limit, skip, takeWhile und dropWhile – dazu, dass komplexere Datentransformationen durch die Stream-API nicht ausgedrückt werden können.

Es fehlen z. B. verbreitete intermediäre Operationen wie window und fold und zahlreiche mehr, wenn man sich die Feature-Requests der Java-Community ansieht.

Anstatt nun all diese Operationen im Stream-Interface zu implementieren, entschieden die JDK-Developer, eine API zu entwickeln, die zum einen im JDK selbst genutzt werden kann, um heiß begehrte intermediäre Operationen zu implementieren, und die zum anderen von Entwickler:innen dazu eingesetzt werden kann, um eigene Operationen zu entwickeln.

Diese neue API heißt „Stream Gatherers“ und wurde in Java 22 im März 2024 – also genau zehn Jahre nach der Einführung der Stream-API – als Preview-Feature veröffentlicht und in Java 24 finalisiert.

In diesem Artikel erfährst du,

• was ein Gatherer ist,
• wie die neue Stream-Gatherers-API funktioniert,
• wie du mit ihr beliebige intermediäre Stream-Operationen implementierst,
• welche Gatherer das JDK-Team bereits implementiert hat und wie du diese Gatherer erzeugst.

Beginnen wir mit einer kurzen Zusammenfassung, wie die Stream-API überhaupt funktioniert.

Stufen der Stream-API

Java-Streams bestehen aus drei Stufen:

1. Stream-Quelle – diese erzeugt einen Stream, z. B. durch IntStream.of() oder Collection.stream().
2. Intermediäre Operationen – diese transformieren die im Stream enthaltenen Elemente, z. B. die Stream-Methoden map(), filter() und limit().
3. Terminale Operationen – diese sammeln z. B. durch toList() die Elemente in einer Liste, durch collect(Collectors.toMap()) in einer Map oder zählen die Elemente mit count().

Hier ist ein einfaches Beispiel – eine Methode, die zählt, wie viele Wörter einer bestimmten Länge in einer Liste von Wörtern enthalten sind:

public long countLongWords(List<String> words, int minLength) {
  return words.stream()                       // ⟵ Source
      .map(String::length)                    // ⟵ Intermediate operation
      .filter(length -> length >= minLength)  // ⟵ Intermediate operation
      .count();                               // ⟵ Terminal operation
}Code-Sprache: Java (java)

Terminale Operation: Stream Collector

Und hier ein Beispiel, das die Wörter in Großbuchstaben umwandelt und nach Länge gruppiert in einer Map speichert:

public Map<Integer, List<String>> groupByLength(List<String> words) {
  return words.stream()                                 // ⟵ Source
      .map(String::toUpperCase)                         // ⟵ Intermediate operation
      .collect(Collectors.groupingBy(String::length));  // ⟵ Terminal operation
}Code-Sprache: Java (java)

In diesem zweiten Beispiel wird der terminalen Operation collect() ein sogenannter „Collector“ übergeben. Ein Collector ist ein Objekt einer Klasse, die das Collector-Interface implementiert und definiert, was am Ende des Streams mit den Elementen des Streams passieren soll. In diesem Fall sollen sie nach Länge gruppiert in eine Map gespeichert werden.

Intermediäre Operation: Stream Gatherer

Analog dazu definiert die Stream-Gatherers-API die Methode Stream.gather() sowie ein Gatherer-Interface. Das folgende Code-Beispiel verwendet die intermediäre Operation „Fixed Window“, welche die Wörter in Listen zu jeweils drei Wörtern gruppiert:

public List<List<String>> groupsOfThree(List<String> words) {
  return words.stream()                  // ⟵ Source
      .gather(Gatherers.windowFixed(3))  // ⟵ Intermediate operation
      .toList();                         // ⟵ Terminal operation
}Code-Sprache: Java (java)

Wenn wir diese Methode z. B. wie folgt aufrufen:

List<String> words = List.of("the", "be", "two", "of", "and", "a", "in", "that");
List<List<String>> groups = groupsOfThree(words);
System.out.println(groups);Code-Sprache: Java (java)

Dann ist die Ausgabe:

[[the, be, two], [of, and, a], [in, that]]Code-Sprache: Klartext (plaintext)

(Da die Stream-Quelle eine Anzahl an Elementen geliefert hat, die nicht ohne Rest durch drei teilbar ist, enthält die letzte Gruppe nur zwei Wörter.)

Welche Gatherer das JDK bereits mitbringt und wie du sie einsetzt, erfährst du im nächsten Kapitel. Wie ein Stream-Gatherer im Inneren aufgebaut ist und wie du eigene schreibst, sehen wir uns danach an.

Die fünf im JDK enthaltenen Stream Gatherers

Bevor wir uns ansehen, wie man eigene Gatherer schreibt, lohnt sich ein Blick auf die Gatherer, die das JDK-Team bereits für uns implementiert hat. Denn in der Praxis wirst du eigene Gatherer eher selten brauchen – die fünf vordefinierten Gatherer aus der Klasse java.util.stream.Gatherers decken die häufigsten Fälle ab:

• windowFixed(int size) – zerlegt den Stream in nicht überlappende Blöcke fester Größe
• windowSliding(int size) – erzeugt überlappende Fenster fester Größe
• fold(Supplier initial, BiFunction folder) – reduziert den Stream zu einem einzigen Element, ähnlich wie reduce, aber als intermediäre Operation
• scan(Supplier initial, BiFunction scanner) – wie fold, emittiert aber jeden Zwischenstand
• mapConcurrent(int maxConcurrency, Function mapper) – führt die Mapping-Funktion mit begrenzter Parallelität über virtuelle Threads nebenläufig aus

Schauen wir sie uns der Reihe nach an.

windowFixed – Blöcke fester Größe

Den windowFixed-Gatherer hast du bereits im Einführungsbeispiel kennengelernt: Er gruppiert die Stream-Elemente in Listen fester Größe. Ein typischer Praxisfall ist die Batch-Verarbeitung – etwa, wenn du viele Datensätze nicht einzeln, sondern in Blöcken in eine Datenbank schreiben willst:

customers.stream()
    .gather(Gatherers.windowFixed(100))
    .forEach(database::insertBatch);Code-Sprache: Java (java)

Statt mit subList() und Index-Arithmetik zu hantieren, übernimmt der Gatherer das Aufteilen. Geht die Anzahl der Elemente nicht ohne Rest auf, enthält das letzte Fenster entsprechend weniger Elemente.

windowSliding – überlappende Fenster

windowSliding funktioniert wie windowFixed, doch die Fenster überlappen sich und verschieben sich um jeweils ein Element:

Stream.of(1, 2, 3, 4, 5)
    .gather(Gatherers.windowSliding(3))
    .forEach(System.out::println);Code-Sprache: Java (java)

Ausgabe:

[1, 2, 3]
[2, 3, 4]
[3, 4, 5]Code-Sprache: Klartext (plaintext)

Der klassische Anwendungsfall ist der gleitende Durchschnitt. Aus einem Stream<Double> mit Messwerten – zum Beispiel stündlichen Temperaturen – berechnest du damit den Mittelwert über jeweils drei aufeinanderfolgende Werte, ohne den Stream vorher in einer Liste zu materialisieren:

List<Double> rollingAvg = readings
    .gather(Gatherers.windowSliding(3))
    .map(window -> window.stream().mapToDouble(Double::doubleValue).average().orElseThrow())
    .toList();Code-Sprache: Java (java)

Bei N Werten und Fenstergröße K erhältst du so N − K + 1 Fenster. Ein Sonderfall: Ist der Stream kürzer als die Fenstergröße, liefert windowSliding ein einziges Fenster mit allen vorhandenen Elementen. Bei einem vollständig leeren Stream entsteht gar kein Fenster.

fold – wie reduce, aber als Gatherer

fold reduziert – wie das aus Java 8 bekannte reduce – den gesamten Stream zu einem einzigen Ergebnis. Der Unterschied: fold ist eine intermediäre Operation und liefert das Ergebnis als ein-elementigen Stream, nicht als fertigen Wert:

List<Integer> result = Stream.of(10, 20, 30, 40)
    .gather(Gatherers.fold(() -> 0, Integer::sum))
    .toList();
// result = [100]Code-Sprache: Java (java)

Wofür ist das gut, wenn reduce doch dasselbe Ergebnis liefert? Für zwei Situationen:

• Die Pipeline soll nach der Aggregation weitergehen. Da fold einen Stream zurückgibt, kannst du direkt weitere Operationen anhängen, wie z. B. map.
• Die Operation ist nicht assoziativ. Bei reduce auf einem parallelen Stream werden zunächst die parallelen Teilstreams aggregiert und dann die Ergebnisse der Teilstreams. Bei einer nicht assoziativen Operation liefert das u. U. nicht das gewünschte Ergebnis. Der fold-Gatherer hingegen arbeitet immer sequenziell – selbst wenn der Stream parallel ist. Deshalb kannst du bei fold auch nicht-assoziative Verknüpfungen verwenden, etwa eine String-Konkatenation in fester Reihenfolge.

Der Startwert wird übrigens als Supplier übergeben und nicht als fertiger Wert – so lässt sich die Erzeugung eines teuren Startobjekts (etwa einer leeren Liste oder einer HashMap) bei Bedarf hinauszögern.

scan – jeden Zwischenstand emittieren

scan verwendet dieselbe Signatur wie fold, emittiert aber nicht nur das Endergebnis, sondern nach jedem Element den aktuellen Zwischenstand:

Stream.of(10, 20, 30, 40)
    .gather(Gatherers.scan(() -> 0, Integer::sum))
    .forEach(System.out::println);Code-Sprache: Java (java)

Ausgabe:

10
30
60
100Code-Sprache: Klartext (plaintext)

Jedes Eingabe-Element erzeugt genau ein Ausgabe-Element – nämlich die Summe bis zu dieser Stelle. Das ist immer dann nützlich, wenn dich nicht das Endergebnis, sondern der Verlauf interessiert: ein laufender Kontostand, eine kumulierte Summe, der Stand zu jedem Zeitpunkt.

reduce, fold und scan im Vergleich

Da sich die drei leicht verwechseln lassen, hier die Unterschiede auf einen Blick:

Faustregel: reduce für einfache Einzelwert-Aggregationen wie bisher, fold, wenn nach der Aggregation noch Stream-Schritte folgen oder die Operation nicht assoziativ ist, und scan, wenn du den Verlauf statt nur des Endergebnisses brauchst.

mapConcurrent – Mapping mit begrenzter Parallelität

mapConcurrent ist der spannendste der fünf Gatherer, weil hier die Stream-API auf die virtuellen Threads aus Java 21 trifft. Er wendet – wie map – eine Funktion auf jedes Element an, tut das aber nebenläufig in virtuellen Threads, mit einer von dir festgelegten maximalen Parallelität:

List<UserData> users = urls.stream()
    .gather(Gatherers.mapConcurrent(50, this::fetchUser))
    .toList();Code-Sprache: Java (java)

Der erste Parameter begrenzt die Anzahl gleichzeitiger Aufrufe: Bei 50 laufen höchstens 50 fetchUser-Aufrufe parallel; der 51. wartet, bis einer der ersten 50 fertig ist. Genau dafür sind virtuelle Threads gemacht – tausende blockierende I/O-Aufrufe auf einer Handvoll Carrier-Threads.

Damit stellt sich die Frage: Wann nimmst du mapConcurrent und wann das altbekannte parallel()?

Der entscheidende Unterschied liegt im Thread-Modell. parallel() verteilt die Arbeit auf Platform-Threads aus dem ForkJoinPool.commonPool, dessen Größe sich an der Anzahl der CPU-Kerne orientiert – ideal für CPU-lastige Aufgaben wie Berechnungen oder Sortieren, aber denkbar schlecht für blockierende I/O: Schon ein paar Dutzend wartende HTTP-Aufrufe legen den von der gesamten JVM genutzten Pool lahm. mapConcurrent startet hingegen pro Element einen virtuellen Thread und ist damit genau auf I/O-lastige Schritte wie HTTP-, Datenbank- oder Dateizugriffe zugeschnitten.

Hinzu kommen zwei praktische Unterschiede: Die maximale Parallelität legst du bei mapConcurrent pro Aufruf über den Parameter fest, während sie bei parallel() global durch die Pool-Größe vorgegeben ist. Und mapConcurrent behält die Reihenfolge der Elemente bei – bei parallel() ist die Ausgabe-Reihenfolge nicht garantiert, es sei denn, du erzwingst sie mit forEachOrdered, was die Parallelisierung wieder ausbremst.

Kurz: parallel() bleibt das Werkzeug für CPU-lastige Arbeit, mapConcurrent ist die Antwort der Stream-API auf I/O-lastige Schritte.

Diese fünf Gatherer decken einen Großteil der typischen Anforderungen ab. Doch was, wenn du eine Operation brauchst, die nicht dabei ist? Dann implementierst du sie selbst – und genau das sehen wir uns jetzt an.

Aufbau eines Stream Gatherers

Bevor wir uns den Aufbau eines Stream-Gatherers ansehen, ist es wichtig, zwei Eigenschaften von Gatherern zu kennen:

• Sie können einen Status haben, so dass Elemente unterschiedlich transformiert werden können, je nachdem, was vorher passiert ist (wofür das relevant ist, zeige ich dir gleich an einem Beispiel).
• Sie können den Stream vorzeitig terminieren, wie es z. B. limit() und takeWhile() tun.

Gatherer werden aus bis zu vier Komponenten aufgebaut:

• Einem optionalen „Initializer“, der den eben erwähnten Status initialisiert.
• Einem „Integrator“, der jedes Element des Streams verarbeitet (ggf. unter Berücksichtigung des aktuellen Status), ggf. den Status aktualisiert, Elemente an die nächste Stufe der Stream-Pipeline weitergibt und ggf. den Stream vorzeitig beendet.
• Einem optionalen „Finisher“, der nach der Verarbeitung des letzten Elements aufgerufen wird, um ggf. anhand des Status weitere Elemente an die nächste Stufe der Stream-Pipeline zu emittieren.
• Und einem optionalen „Combiner“, der bei der parallelen Verarbeitung eines Streams eingesetzt wird, um die Status von parallel ausgeführten Transformationen zu kombinieren.

In den folgenden Abschnitten schauen wir uns die Komponenten nach und nach und mit vielen Beispielen an.

Integrator

Der Integrator ist die einzige Komponente, die zwingend erforderlich ist. Mit ausschließlich einem Integrator können wir bereits einen einfachen, statuslosen Gatherer entwickeln.

Im Folgenden zeige ich dir, wie sich die Stream.map()-Funktion mit einem Gatherer implementieren lässt.

Die gezeigten Interfaces haben zusätzliche statische oder Default-Methoden, die für ein grundlegendes Verständnis nicht wichtig sind. Ich lasse sie daher weg und schreibe anstelle der ausgelassenen Methoden drei Punkte.

Integrator ist ein funktionales Interface mit einer integrate()-Methode:

@FunctionalInterface
public interface Integrator<A, T, R> {
  boolean integrate(A state, T element, Downstream<? super R> downstream);
  . . .
}Code-Sprache: Java (java)

Die integrate()-Methode bekommt drei Parameter übergeben: 

• den (optionalen) Zustand vom Typ A,
• das Element vom Typ T – das ist das Element, das die vorherige Stufe der Stream-Pipeline an diese Stufe schickt,
• einen Downstream, über den die integrate()-Methode ggf. Elemente vom Typ R an die nächste Stufe der Stream-Pipeline schickt.

Downstream ist ein funktionales Interface mit einer push()-Methode:

@FunctionalInterface
public interface Downstream<T> {
  boolean push(T element);
  . . .
}Code-Sprache: Java (java)

push() gibt ein boolean zurück, das anzeigt, ob der Downstream danach weitere Elemente geschickt haben möchte oder nicht. Wenn im Downstream z. B. irgendwo ein limit() vorkommt, würde hier false zurückkommen, wenn das Limit erreicht ist.

Auch die Integrator.integrate() gibt ein boolean zurück. Damit zeigt diese Stufe der Stream-Pipeline der vorherigen Stufe an, ob diese Stufe weitere Elemente verarbeiten möchte. Wenn nicht, gibt die integrate()-Methode false zurück und schickt damit sozusagen ein Stop-Signal an die Stream-Quelle.

Einen Integrator, der eine Mapping-Funktion aufruft und das Ergebnis der Mapping-Funktion an den Downstream emittiert (also an die nächste Verarbeitungsstufe des Streams weiterleitet), können wir als Lambda-Funktion wie folgt schreiben. Da wir hier keinen Zustand benötigen, verwenden wir Void als Typ für die state-Variable:

Function<T, R> mapper = . . .

Integrator<Void, T, R> integrator = 
    (state, element, downstream) -> {
      R mappedElement = mapper.apply(element);
      return downstream.push(mappedElement);
    };Code-Sprache: Java (java)

In der ersten Zeile des Integrators wird die Mapping-Funktion auf das vom Upstream eingehende Element angewendet. In der zweiten Zeile wird das Ergebnis-Element der Mapping-Funktion an den Downstream emittiert und die Antwort des Downstreams zurückgegeben.

Da der oben gezeigte Integrator niemals von sich aus false zurückliefert, sondern nur dann, wenn dies vom Downstream ausgeht, wird der Integrator als „gierig“ (englisch: „greedy“) bezeichnet. Um der Stream-Pipeline dies anzuzeigen und ihr damit Optimierungen zu ermöglichen, sollten wir den Integrator daher einmal mit Integrator.ofGreedy() wrappen:

Integrator<Void, T, R> integrator =
    Integrator.ofGreedy(
        (state, element, downstream) -> {
          R mappedElement = mapper.apply(element);
          return downstream.push(mappedElement);
        });
Code-Sprache: Java (java)

Um aus dem Integrator schließlich einen Gatherer zu machen, verwenden wir die statische Gatherer.of()-Methode:

Gatherer<T, Void, R> gatherer = Gatherer.of(integrator);Code-Sprache: Java (java)

Hier ist ein vollständiges Beispiel mit einer Methode, die einen Gatherer für eine bestimmte Mapping-Funktion erzeugt und einer Methode, die solch einen Gatherer nutzt, um eine Liste von Strings auf deren Längen zu mappen:

public <T, R> Gatherer<T, Void, R> mapping(Function<T, R> mapper) {
  return Gatherer.of(
      Integrator.ofGreedy(
          (state, element, downstream) -> {
            R mappedElement = mapper.apply(element);
            return downstream.push(mappedElement);
          }));
}

public List<Integer> toLengths(List<String> words) {
  return words.stream()
      .gather(mapping(String::length))
      .toList();
}Code-Sprache: Java (java)

Damit ist das grundlegende Konzept eines Gatherers erklärt.

Initializer

Der Gatherer des vorherigen Abschnitts war statuslos, d. h. die Transformation eines Elements war unabhängig von allem, was zuvor passiert ist.

In diesem Abschnitt zeige ich dir, wie sich die Stream.limit()-Funktion mit einem Gatherer implementieren lässt. Dazu muss der Gatherer die verarbeiteten Elemente zählen und den Stream nach Erreichen der gewünschten Anzahl von Elementen vorzeitig terminieren.

Zum Zählen brauchen wir einen Status, einen Zähler. Der Initializer ist vom Typ Supplier und liefert den initialen Status. Zum Zählen eignet sich als Status z. B. ein AtomicInteger:

Supplier<AtomicInteger> initializer = AtomicInteger::new;Code-Sprache: Java (java)

Den limitierenden Integrator implementieren wir wie folgt:

int maxSize = . . .

Integrator<AtomicInteger, T, T> integrator = 
    (state, element, downstream) -> {
      if (state.get() < maxSize) {
        boolean result = downstream.push(element);
        state.incrementAndGet();
        return result;
      } else {
        return false;
      }
    };Code-Sprache: Java (java)

Solange unser Status, der Element-Zähler, kleiner ist als maxSize, emittieren wir die Stream-Elemente in den Downstream, erhöhen den Zähler um eins und geben das Antwort-Boolean des Downstreams zurück. Sobald die gewünschte Anzahl Elemente erreicht ist, geben wir false zurück und zeigen damit an, dass der Stream beendet werden soll.

Wir könnten übrigens auch in der if-Anweisung state.getAndIncrement() anstelle von state.get() schreiben und das state.incrementAndGet() weglassen. Doch das wäre etwas komplizierter zu erklären gewesen.

Beachte, dass wir diesen Integrator nicht mit Integrator.ofGreedy() wrappen, da dieser Integrator auch von sich aus false zurückliefern kann (also nicht nur, wenn dieser Wert vom Downstream kam).

Um aus dem Initializer und dem Integrator einen Gatherer zu machen, verwenden wir die Gatherer.ofSequential()-Methode. Der Name dieser Methode weist darauf hin, dass der zurückgelieferte Gatherer nicht parallel arbeiten kann. Das liegt daran, dass er einen Status hat, aber keinen Combiner (dazu später mehr).

Gatherer<T, AtomicInteger, T> gatherer = Gatherer.ofSequential(initializer, integrator);Code-Sprache: Java (java)

Das folgende Listing zeigt eine Methode, die einen limitierenden Gatherer aus den zuvor gezeigten Bausteinen erzeugt – dieses Mal mit state.getAndIncrement(), und eine Methode, die diesen Gatherer nutzt, um die ersten drei Wörter der Wort-Liste zurückzugeben:

public <T> Gatherer<T, AtomicInteger, T> limiting(int maxSize) {
  return Gatherer.ofSequential(
      // Initializer
      AtomicInteger::new,
 
      // Integrator
      (state, element, downstream) -> {
        if (state.getAndIncrement() < maxSize) {
          return downstream.push(element);
        } else {
          return false;
        }
      });
}

public List<String> firstThreeWords(List<String> words) {
  return words.stream()
      .gather(limiting(3))
      .toList();
}Code-Sprache: Java (java)

Stream Gatherers sind übrigens so mächtig, dass nicht nur map() und limit(), sondern jede existierende intermediäre Operation der Stream-API auch als Gatherer implementiert werden kann.

Finisher

Um zu erklären, wofür wir einen Finisher benötigen, zeige ich dir in diesem Abschnitt, wie sich der im Einführungskapitel gezeigte „Fixed Window“-Gatherer implementieren lässt.

Zur Erinnerung: Dieser soll aus einem Stream von Elementen einen Stream von Listen machen, die jeweils eine bestimmte Anzahl der Elemente enthalten.

Wir beginnen einfach mal mit der Implementierung. An einer Stelle werden wir feststellen, dass wir nicht weiterkommen – und genau dann werde ich den Finisher einführen.

Zunächst einmal benötigt unser Gatherer einen Status. Da wir Elemente in Listen gruppieren wollen, liegt es nahe, solch eine Liste als Status zu verwenden. Entsprechend implementieren wir den Initializer:

Supplier<List<T>> initializer = ArrayList::new;Code-Sprache: Java (java)

Den Integrator implementieren wir wie folgt:

Integrator<List<T>, T, List<T>> integrator =
    (state, element, downstream) -> {
      state.add(element);
      if (state.size() == windowSize) {
        boolean result = downstream.push(List.copyOf(state));
        state.clear();
        return result;
      } else {
        return true;
      }
    };Code-Sprache: Java (java)

Wir hängen das ankommende Element an die Status-Liste an. Sobald die Liste die gewünschte Größe erreicht hat, senden wir eine Kopie der Liste an den Downstream und leeren die Liste.

Das funktioniert allerdings nur, wenn die Anzahl der Elemente ein Vielfaches der Fenstergröße ist. Wenn wir beispielsweise acht Elemente haben und eine Fenstergröße von drei, dann würden für die ersten sechs Elemente zwei Listen zu je drei Elementen an den Downstream emittiert werden. Das siebte und achte Element würden ebenfalls in einer Liste liegen, doch da diese Liste die gewünschte Größe noch nicht erreicht hat, wird sie vom Integrator nicht in den Downstream emittiert.

Genau hier kommt der Finisher ins Spiel. Der Finisher bekommt als Input den Status nach der Verarbeitung aller Stream-Elemente sowie den Downstream und kann dann in Abhängigkeit vom Status weitere Elemente in den Downstream emittieren.

Für die Fixed-Window-Operation würde der Finisher wie folgt aussehen:

BiConsumer<List<T>, Downstream<List<T>>> finisher =
    (state, downstream) -> {
      if (!state.isEmpty()) {
        downstream.push(List.copyOf(state));
      }
    };Code-Sprache: Java (java)

Falls die Liste Elemente enthält, wird sie in den Downstream emittiert.

Initializer, Integrator und Finisher kombinieren wir wie folgt zu einem Gatherer:

Gatherer<T, List<T>, List<T>> gatherer =
    Gatherer.ofSequential(initializer, integrator, finisher);Code-Sprache: Java (java)

Das folgende Listing zeigt eine Methode, die aus den zuvor gezeigten Komponenten einen Window-Gatherer erzeugt, sowie eine Methode, die diesen Gatherer nutzt:

public <T> Gatherer<T, List<T>, List<T>> windowing(int windowSize) {
  return Gatherer.ofSequential(
      // Initializer
      ArrayList::new,

      // Gatherer
      (state, element, downstream) -> {
        state.add(element);
        if (state.size() == windowSize) {
          boolean result = downstream.push(List.copyOf(state));
          state.clear();
          return result;
        } else {
          return true;
        }
      },

      // Finisher
      (state, downstream) -> {
        if (!state.isEmpty()) {
          downstream.push(List.copyOf(state));
        }
      });
}

public List<List<String>> groupWords(List<String> words, int groupSize) {
  return words.stream()
      .gather(windowing(groupSize))
      .toList();
}Code-Sprache: Java (java)

Der Einfachheit halber habe ich im vorangegangenen Beispiel als Statusobjekt eine ArrayList verwendet. Das Erzeugen von Kopien und das Leeren der Liste stellt jedoch einen nicht unerheblichen Overhead dar.

Die folgende Lösung verwendet als Status ein Wrapper-Objekt, das eine Liste enthält, welche direkt in den Downstream gepusht und dann neu erzeugt wird. Diese Variante ist ca. 20 % schneller:

public <T> Gatherer<T, ?, List<T>> windowing(int windowSize) {
  return Gatherer.ofSequential(
      // Initializer
      () -> new Object() { List<T> list = new ArrayList<>(); },

      // Gatherer
      (state, element, downstream) -> {
        state.list.add(element);
        if (state.list.size() == windowSize) {
          boolean result = downstream.push(state.list);
          state.list = new ArrayList<>();
          return result;
        } else {
          return true;
        }
      },

      // Finisher
      (state, downstream) -> {
        if (!state.list.isEmpty()) {
          downstream.push(state.list);
        }
      });
}Code-Sprache: Java (java)

Im nächsten Abschnitt kommen wir zum letzten Stream-Gatherer-Baustein, dem Combiner.

Combiner

Um einen statusbehafteten Gatherer parallel auszuführen, benötigt er eine Combiner-Funktion. Ein Combiner kombiniert in der Join-Phase der parallelen Stream-Verarbeitung jeweils zwei Status zu einem:

In diesem Abschnitt wollen wir einen Gatherer implementieren, der entsprechend eines Comparators das größte aller eingehenden Elemente in den Downstream emittiert.

Als Status verwenden wir eine AtomicReference, die entweder kein Element oder das aktuell größte gefundene Element enthält:

Supplier<AtomicReference<T>> initializer = AtomicReference::new;
Code-Sprache: Java (java)

Der Integrator speichert das eingehende Element im Status, wenn der Status leer ist oder wenn das eingehende Element größer ist als das im Status gespeicherte Element. Da der Integrator immer true zurückliefert, kennzeichnen wir ihn als „greedy“:

Integrator<AtomicReference<T>, T, T> integrator =
    Integrator.ofGreedy(
        (state, element, downstream) -> {
          T bestElement = state.get();
          if (bestElement == null || comparator.compare(element, bestElement) > 0) {
            state.set(element);
          }
          return true;
        });
Code-Sprache: Java (java)

Der Finisher sendet das Element, sofern der Status eines enthält, an den Downstream:

BiConsumer<AtomicReference<T>, Downstream<T>> finisher =
    (state, downstream) -> {
      T bestElement = state.get();
      if (bestElement != null) {
        downstream.push(bestElement);
      }
    };
Code-Sprache: Java (java)

Und der Combiner kombiniert zwei Status zu einem:

• Ist ein Status leer, gibt er den jeweils anderen Status zurück.
• Enthalten beide Status ein Element, gibt er den Status mit dem größeren Element zurück.

Wenn der Eingangsstream leer ist, wird der Combiner nie aufgerufen, d. h. der Fall, dass beide Status leer sind, kann nicht eintreten.

BinaryOperator<AtomicReference<T>> combiner =
    (state1, state2) -> {
      T bestElement1 = state1.get();
      T bestElement2 = state2.get();

      if (bestElement1 == null) {
        return state2;
      } else if (bestElement2 == null) {
        return state1;
      } else if (comparator.compare(bestElement1, bestElement2) > 0) {
        return state1;
      } else {
        return state2;
      }
    };Code-Sprache: Java (java)

Initializer, Integrator, Combiner und Finisher kombinieren wir wieder mit Gatherer.of() zu einem Gatherer:

Gatherer<T, AtomicReference<T>, T> gatherer =
    Gatherer.of(initializer, integrator, combiner, finisher);
Code-Sprache: Java (java)

Und auch für den Combiner folgt ein Listing mit einer Methode, die einen Maximum-Gatherer aus den zuvor gezeigten Bausteinen erzeugt, sowie einer Methode, die diesen in einem parallelen Stream nutzt, um das längste Wort einer Liste zu finden:

public <T> Gatherer<T, AtomicReference<T>, T> maximumBy(Comparator<T> comparator) {
  return Gatherer.of(
      // Initializer
      AtomicReference::new,

      // Integrator
      Integrator.ofGreedy(
          (state, element, downstream) -> {
            T bestElement = state.get();
            if (bestElement == null || comparator.compare(element, bestElement) > 0) {
              state.set(element);
            }
            return true;
          }),

      // Combiner
      (state1, state2) -> {
        T bestElement1 = state1.get();
        T bestElement2 = state2.get();

        if (bestElement1 == null) {
          return state2;
        } else if (bestElement2 == null) {
          return state1;
        } else if (comparator.compare(bestElement1, bestElement2) > 0) {
          return state1;
        } else {
          return state2;
        }
      },

      // Finisher
      (state, downstream) -> {
        T bestElement = state.get();
        if (bestElement != null) {
          downstream.push(bestElement);
        }
      });
}

public Optional<String> getLongest(List<String> words) {
  return words.parallelStream()
      .gather(maximumBy(Comparator.comparing(String::length)))
      .findFirst();
}Code-Sprache: Java (java)

Damit haben wir alle vier Komponenten eines Stream Gatherers beisammen – und du kennst jetzt sowohl die fünf vom JDK mitgelieferten Gatherer als auch den Weg, dir bei Bedarf einen eigenen zu bauen.

Im nächsten Kapitel sehen wir uns an, wie du mehrere Gatherer zu einer einzigen Transformationssequenz kombinierst.

Gatherers kombinieren

So wie du beispielsweise mehrere filter()- und map()-Operationen hintereinander schalten kannst, kannst du auch mehrere Gatherer hintereinander aufrufen, z. B. so:

var result = source
    .gather(a)
    .gather(b)
    .gather(c)
    .collect();Code-Sprache: Java (java)

Wenn du eine bestimmte Abfolge von Gatherern regelmäßig benötigst, kannst du diese – ganz im Sinne von DRY (don't repeat yourself) – auch zu einem einzigen Gatherer kombinieren, z. B. so:

Gatherer abc = a.andThen(b).andThen(c);

var result = source
    .gather(abc)
    .collect();Code-Sprache: Java (java)

So kannst du eine beliebig lange Transformationssequenz redundanzfrei auf verschiedene Streams anwenden.

Limitierungen von Stream Gatherern

Stream Gatherers sind leistungsstarke Werkzeuge, haben jedoch zwei wesentliche Einschränkungen:

• Es gibt sie (genau wie Collectoren) nicht für die primitiven Streams IntStream, LongStream und DoubleStream.
• Sie haben (genau wie Collectoren) keinen Zugriff auf die (im Spliterator-Interface definierten) Charakteristika des Streams. Das bedeutet, dass sie nicht auf der Grundlage dieser Eigenschaften optimiert werden können (z. B. der Tatsache, dass die Größe bekannt ist oder dass der Stream nur unterschiedliche Elemente enthält).

Historie

Stream Gatherers wurden in folgenden JDK Enhancement Proposals definiert:

• Java 22: JEP 461: Stream Gatherers (Preview)
• Java 23: JEP 473: Stream Gatherers (Second Preview)
• Java 24: JEP 485: Stream Gatherers

Fazit

Mit Stream Gatherers kannst du beliebige intermediäre Stream-Operationen implementieren – so, wie du mit Collectoren schon immer beliebige terminale Operationen schreiben konntest. Das erlaubt dir, deutlich aussagekräftigere Stream-Pipelines zu bauen als bisher.

Für die häufigsten Fälle musst du allerdings gar nicht selbst Hand anlegen: Die Klasse java.util.stream.Gatherers liefert fünf vordefinierte Gatherer mit. Mit windowFixed und windowSliding zerlegst du einen Stream in nicht überlappende bzw. überlappende Fenster, mit fold reduzierst du ihn – wie reduce, aber als intermediäre Operation – zu einem einzigen Element, mit scan emittierst du jeden Zwischenstand der Akkumulation, und mit mapConcurrent verarbeitest du I/O-lastige Mapping-Schritte nebenläufig über virtuelle Threads.

Brauchst du eine Operation, die nicht dabei ist, kombinierst du entweder vorhandene Gatherer – oder du baust dir aus Initializer, Integrator, Combiner und Finisher deinen eigenen.

Welchen der vordefinierten Gatherer wirst du als Erstes einsetzen? Und welche Funktionalität planst du, selbst als Gatherer zu implementieren? Schreibe mir einen Kommentar!