Eigene Collector und Spliterator

1. Eigene Collector: wann und wie man sie schreibt

In der Java Stream-API wird das Interface Collector verwendet, um einen Stream in eine Sammlung oder ein Aggregat zu überführen. Üblicherweise nutzen Sie die fertigen Collector aus der Klasse Collectors (toList(), toMap(), groupingBy() usw.), aber manchmal brauchen Sie etwas Besonderes – dann können Sie Ihren eigenen Collector schreiben.

Collector ist ein Objekt, das beschreibt, wie aus den Elementen eines Streams ein Endergebnis gebildet wird. Er definiert vier (eigentlich fünf) zentrale Komponenten:

• supplier – erzeugt einen neuen Container zum Sammeln der Elemente (z. B. eine neue Liste oder Map).
• accumulator – fügt das nächste Element in den Container ein.
• combiner – führt zwei Container zusammen (wichtig für parallele Streams!).
• finisher – wandelt den Container in das Endergebnis um (z. B. macht ihn unveränderlich oder konvertiert in einen anderen Typ).
• characteristics – eine Menge von Flags, die die Eigenschaften des Collectors beschreiben (z. B. ob Parallelität unterstützt wird, ob der Ergebnis-Typ verändert wird usw.).

Signatur:

Collector<T, A, R>

• T – Typ der Stream-Elemente,
• A – Typ des Zwischen-Akkumulators,
• R – Typ des Ergebnisses.

2. Beispiel: Collector für eine MultiMap (Map<K, List<V>>)

Angenommen, Sie möchten einen Stream von Paaren Pair<K, V> in eine Map<K, List<V>> (Multi-Map) sammeln, in der jedem Schlüssel eine Liste von Werten zugeordnet ist.

Beispielimplementierung:

public static <K, V> Collector<Pair<K, V>, ?, Map<K, List<V>>> toMultiMap() {
    return Collector.of(
        HashMap::new, // supplier
        (map, pair) -> map.computeIfAbsent(pair.key(), k -> new ArrayList<>()).add(pair.value()), // accumulator
        (map1, map2) -> { // combiner
            map2.forEach((k, vList) -> map1.merge(k, vList, (l1, l2) -> { l1.addAll(l2); return l1; }));
            return map1;
        },
        Function.identity(), // finisher
        Collector.Characteristics.UNORDERED
    );
}

Verwendung:

List<Pair<String, Integer>> pairs = List.of(
    new Pair<>("a", 1), new Pair<>("b", 2), new Pair<>("a", 3)
);

Map<String, List<Integer>> multiMap = pairs.stream().collect(toMultiMap());
// multiMap: {a=[1, 3], b=[2]}

3. Beispiel: Collector für die Top‑N‑Elemente

Angenommen, Sie möchten einen Stream in eine Liste der N größten Elemente sammeln (z. B. Top‑5 absteigend).

Implementierung:

public static <T> Collector<T, ?, List<T>> topN(int n, Comparator<? super T> comparator) {
    return Collector.of(
        () -> new PriorityQueue<>(n, comparator), // supplier
        (pq, t) -> {
            pq.offer(t);
            if (pq.size() > n) pq.poll(); // kleinstes Element entfernen
        },
        (pq1, pq2) -> {
            pq2.forEach(t -> {
                pq1.offer(t);
                if (pq1.size() > n) pq1.poll();
            });
            return pq1;
        },
        pq -> {
            List<T> result = new ArrayList<>(pq);
            result.sort(comparator.reversed()); // absteigend
            return result;
        },
        Collector.Characteristics.UNORDERED
    );
}

Verwendung:

List<Integer> top3 = Stream.of(5, 1, 9, 3, 7, 2).collect(topN(3, Comparator.naturalOrder()));
// top3: [9, 7, 5]

4. Wann Sie keinen eigenen Collector schreiben sollten

• Wenn sich die Aufgabe durch eine Kombination aus Standard-Collector und Downstream-Operationen (groupingBy, mapping, flatMapping, collectingAndThen usw.) ausdrücken lässt, sollten Sie diese bevorzugen.
• Ein eigener Collector ist nur für wirklich ungewöhnliche Szenarien nötig (besondere Datenstruktur, komplexe Aggregation, Top‑N, Multi-Maps usw.).
• Schreiben Sie keinen Collector nur um des Collectors willen – das erschwert Wartung und Tests.

Beispiel:

// Statt eines eigenen Collectors für Map<K, Set<V>>:
.collect(Collectors.groupingBy(
    Pair::key,
    Collectors.mapping(Pair::value, Collectors.toSet())
))

5. Eigener Spliterator: wozu und wie

Spliterator ist ein spezielles Interface für das effiziente Durchlaufen und Aufteilen von Collections (oder anderen Datenquellen) in Teile, insbesondere für die parallele Verarbeitung. Im Gegensatz zu einem normalen Iterator kann ein Spliterator eine Collection „aufteilen“ (split), um unabhängige Stücke für die parallele Verarbeitung zu erzeugen.

Schlüsselmethoden:

• tryAdvance(Consumer<? super T> action) – das nächste Element verarbeiten.
• trySplit() – versuchen, die Collection in zwei Teile aufzuteilen (gibt einen neuen Spliterator für einen der Teile zurück).
• estimateSize() – Schätzung der verbleibenden Elementanzahl.
• characteristics() – Bitmaske der Eigenschaften (ORDERED, SIZED, SUBSIZED usw.).

trySplit: Teilungsstrategien

Ausgewogene Teilung ist wichtig für parallele Streams: trySplit sollte ungefähr gleich große Teile zurückgeben, damit die Threads gleichmäßig ausgelastet sind.

Wenn es nichts aufzuteilen gibt (z. B. zu wenige Elemente) – geben wir null zurück.

Beispiel: Spliterator für das portionierte Lesen einer Datei
Angenommen, Sie haben eine große Datei und möchten sie in Portionen von jeweils 1000 Zeilen verarbeiten, um nicht alles im Speicher zu halten.

public class ChunkedLineSpliterator implements Spliterator<List<String>> {
    private final BufferedReader reader;
    private final int chunkSize;

    public ChunkedLineSpliterator(BufferedReader reader, int chunkSize) {
        this.reader = reader;
        this.chunkSize = chunkSize;
    }

    @Override
    public boolean tryAdvance(Consumer<? super List<String>> action) {
        List<String> chunk = new ArrayList<>(chunkSize);
        try {
            String line;
            for (int i = 0; i < chunkSize && (line = reader.readLine()) != null; i++) {
                chunk.add(line);
            }
            if (chunk.isEmpty()) return false;
            action.accept(chunk);
            return true;
        } catch (IOException e) {
            throw new UncheckedIOException(e);
        }
    }

    @Override
    public Spliterator<List<String>> trySplit() {
        // Für das Streaming-Lesen aus einer Datei ergibt das Aufteilen keinen Sinn – wir geben null zurück
        return null;
    }

    @Override
    public long estimateSize() {
        return Long.MAX_VALUE; // im Voraus unbekannt
    }

    @Override
    public int characteristics() {
        return ORDERED | NONNULL;
    }
}

Verwendung:

try (BufferedReader reader = Files.newBufferedReader(Path.of("big.txt"))) {
    StreamSupport.stream(new ChunkedLineSpliterator(reader, 1000), false)
        .forEach(chunk -> processChunk(chunk));
}

Spliterator-Eigenschaften

• ORDERED – Elemente kommen in einer definierten Reihenfolge (z. B. Liste).
• SIZED – die genaue Elementanzahl ist bekannt.
• SUBSIZED – alle durch trySplit erzeugten Spliterator sind ebenfalls SIZED.
• IMMUTABLE – die Quelle ändert sich während des Traversierens nicht.
• CONCURRENT – die Quelle unterstützt sichere parallele Modifikation.
• DISTINCT, SORTED, NONNULL – zusätzliche Eigenschaften.

Wichtig: Eigenschaften korrekt anzugeben beeinflusst die Optimierung von Streams.

6. Beispiele

• Datei stückweise (in Chunks) lesen – ermöglicht es, große Dateien abschnittsweise zu verarbeiten, ohne alles in den Speicher zu laden.
• Parsen ohne unnötige Allokationen – wenn Sie einen Byte-/Zeichenstrom parsen und die Erstellung temporärer Objekte minimieren möchten, können Sie einen Spliterator implementieren, der „Fenster“ oder „Slices“ des Ausgangsarrays liefert.

Beispiel: Spliterator für das zeilenweise Parsen von CSV

public class CsvLineSpliterator implements Spliterator<String[]> {
    private final BufferedReader reader;

    public CsvLineSpliterator(BufferedReader reader) {
        this.reader = reader;
    }

    @Override
    public boolean tryAdvance(Consumer<? super String[]> action) {
        try {
            String line = reader.readLine();
            if (line == null) return false;
            action.accept(line.split(","));
            return true;
        } catch (IOException e) {
            throw new UncheckedIOException(e);
        }
    }

    @Override
    public Spliterator<String[]> trySplit() {
        return null; // sequentielles Parsen
    }

    @Override
    public long estimateSize() {
        return Long.MAX_VALUE;
    }

    @Override
    public int characteristics() {
        return ORDERED | NONNULL;
    }
}

7. Integration mit parallel() – wie man es sicher macht

• Wenn Ihr Spliterator parallele Aufteilung unterstützt (trySplit gibt nicht null zurück) und die Eigenschaften SIZED/SUBSIZED einschließen, kann die Stream-API die Verarbeitung effektiv parallelisieren.
• Für Streaming-Quellen (Dateien, Sockets) wird eine Aufteilung normalerweise nicht unterstützt – verwenden Sie sequentielle Streams.
• Für Collections und Arrays – implementieren Sie eine ausgewogene Teilung (z. B. Array halbieren).

Beispiel: Spliterator für ein Array

public class ArraySpliterator<T> implements Spliterator<T> {
    private final T[] array;
    private int start, end;

    public ArraySpliterator(T[] array, int start, int end) {
        this.array = array;
        this.start = start;
        this.end = end;
    }

    @Override
    public boolean tryAdvance(Consumer<? super T> action) {
        if (start < end) {
            action.accept(array[start++]);
            return true;
        }
        return false;
    }

    @Override
    public Spliterator<T> trySplit() {
        int mid = (start + end) >>> 1;
        if (mid == start) return null;
        ArraySpliterator<T> split = new ArraySpliterator<>(array, start, mid);
        start = mid;
        return split;
    }

    @Override
    public long estimateSize() {
        return end - start;
    }

    @Override
    public int characteristics() {
        return ORDERED | SIZED | SUBSIZED | IMMUTABLE;
    }
}

Verwendung:

String[] arr = {"a", "b", "c", "d"};
StreamSupport.stream(new ArraySpliterator<>(arr, 0, arr.length), true)
    .forEach(System.out::println);