Manchmal reicht es nicht aus, dass Code funktioniert – er muss auch unter Last funktionieren. In modernen Anwendungen, die große Datenmengen verarbeiten, steht Entwicklerinnen und Entwicklern mit der Streams-API in Java ein elegantes, deklaratives Werkzeug zur Verfügung, um Daten in Pipelines zu transformieren, zu filtern und schließlich zu aggregieren. Die Vorstellung, mit wenigen Zeilen komplexe Datenoperationen zu beschreiben, ist nicht nur verführerisch, sondern tatsächlich realistisch. Doch was passiert, wenn diese Operationen auf Millionen von Einträgen treffen? Wenn die Ausführung in mehreren Threads parallel erfolgen soll, um Zeit zu sparen und Mehrkernsysteme effektiv zu nutzen?

Genau an dieser Stelle rückt ein Konzept in den Vordergrund, das oft zu wenig Beachtung findet: der Collector. Er ist das Element am Ende einer Stream-Pipeline, das bestimmt, was mit den verarbeiteten Daten geschehen soll. Und obwohl die API einfach erscheint – collect(Collectors.toList()) – verbirgt sich dahinter eine Architektur, die in paralleler Ausführung ganz eigene Herausforderungen mit sich bringt.

Im Folgenden geht es daher nicht nur um die Syntax oder die Mechanik von Collectoren, sondern um ein tiefes Verständnis für die Bedingungen, unter denen sie korrekt und effizient zum Einsatz kommen. Wir schauen auf Standardlösungen des JDK (Java Development Kit), diskutieren individuelle Implementierungen, zeigen typische Fehler – und kommen letztlich zu der Frage: Wie viel Parallelisierung verträgt ein Collector, ohne dass es gefährlich wird?

Grundlagen: Collector und Parallelität

Die Streams-API von Java vermittelt auf den ersten Blick den Eindruck, dass sich das Sammeln von Ergebnissen – das sogenannte terminale Aggregieren – problemlos parallelisieren lässt. Doch hinter der Methode collect(...) verbirgt sich mehr als nur syntaktische Bequemlichkeit. Sie ist eine koordinierte Zusammenarbeit zwischen einem Datenstrom und einem Collector – einem Objekt, das aus Einzelteilen ein Ganzes formt.

Ein Collector besteht im Kern aus vier funktionalen Komponenten: dem supplier, der für jeden Teilprozess einen neuen Zwischenspeicher bereitstellt; dem accumulator, der Elemente in diesen Zwischenspeicher einspeist; dem combiner, der mehrere Zwischenspeicher zu einem zusammenführt; und schließlich dem finisher, der das Endergebnis produziert. Während supplier und accumulator auch in sequenziellen Streams essenziell sind, tritt der combiner erst dann in Aktion, wenn mehrere Threads unabhängig voneinander gesammelt haben – also bei einem parallelStream().

Hier liegt der erste fundamentale Unterschied zwischen sequenzieller und paralleler Verarbeitung: In einem sequenziellen Stream genügt es, schrittweise in einen einzigen Speicher zu akkumulieren. In der parallelen Variante hingegen entstehen mehrere voneinander isolierte Zwischenspeicher, deren Inhalte später konfliktfrei zu einem Endergebnis verschmolzen werden müssen. Dieses Verschmelzen geschieht durch den combiner – und genau an dieser Stelle entscheidet sich, ob ein Collector für parallele Verarbeitung tauglich ist oder nicht.

Die Tauglichkeit hängt von mehreren Eigenschaften ab: Die Operationen müssen assoziativ sein, also unabhängig von der Kombination der Zwischenergebnisse dasselbe Resultat liefern. Zudem darf kein geteilter Zustand ohne Synchronisierung vorliegen. Und nicht zuletzt müssen die einzelnen Schritte deterministisch und frei von Seiteneffekten bleiben – andernfalls wird aus einer Parallelisierung schnell eine Quelle subtiler Fehler.

Das Wissen um diese strukturellen Anforderungen ist der erste Schritt zu einem bewussten Einsatz paralleler Verarbeitung. Denn nur wer verstanden hat, wie Collector und Stream im Zusammenspiel funktionieren, kann abschätzen, wann ein Performancegewinn möglich ist – und wann man sich stattdessen instabile oder schlicht falsche Ergebnisse einhandelt.

Kriterien für parallelisierbare Collectoren

Stellen wir uns vor, ein Stream wird parallel ausgeführt – etwa über ein großes Dataset, das in mehrere Segmente aufgeteilt ist. Jedes dieser Segmente wird nun unabhängig verarbeitet. Was trivial klingt, hat tiefgreifende Implikationen: Sobald mehrere Threads gleichzeitig sammeln, dürfen sich deren Zwischenergebnisse nicht in die Quere kommen. Die Verantwortung für die Korrektheit liegt beim Collector – genauer: bei seiner strukturellen und funktionalen Ausgestaltung.

Die erste grundlegende Eigenschaft ist Assoziativität. Ein combiner-Aufruf muss unabhängig von der Reihenfolge konsistente Ergebnisse liefern. combine(a, b) und combine(b, a) müssen äquivalente Resultate erzeugen. Das ist notwendig, weil die Reihenfolge der Kombination in einem parallelen Kontext vom Scheduler abhängt – und somit unvorhersagbar ist.

Der zweite Punkt betrifft den Zugriff auf Speicherstrukturen. Sobald ein Collector während der Akkumulation einen gemeinsamen, veränderbaren Zustand nutzt – etwa eine nicht synchronisierte Liste oder Map – entsteht ein potenzieller Hotspot für Race Conditions. Der Collector muss entweder ausschließlich mit lokalen, thread-isolierten Zwischenspeichern arbeiten oder sich auf nebenläufige Datenstrukturen stützen, wie etwa ConcurrentHashMap, LongAdder oder explizit synchronisierte Wrapper.

Darüber hinaus ist auch Determinismus ein wesentliches Kriterium: Eine parallele Ausführung darf nicht zu unterschiedlichen Ergebnissen führen – weder inhaltlich noch strukturell. Insbesondere bei ungeordneten Strukturen wie HashSet oder HashMap ist Vorsicht geboten, da die Iterationsreihenfolge variieren kann – was bei Collectors.joining() oder Collectors.toMap() problematisch wird, wenn die Anwendung auf Ordnung angewiesen ist.

Die drei Anforderungen Assoziativität, isolierter Zustand und Determinismus bilden den technischen Prüfstein für parallele Collectoren. Sie sind nicht optional, sondern grundlegend. Wer sie ignoriert, riskiert schwer zu reproduzierende Fehler, unvollständige Ergebnisse oder performante, aber semantisch falsche Ausgaben.

Parallelität in Java Streams ist mächtig, aber nicht trivial umzusetzen

Beispiele aus der Java-Standardbibliothek: Ein naheliegender Weg, um das abstrakte Konzept paralleler Collectoren greifbar zu machen, führt über die bereits in der Java-Standardbibliothek enthaltenen Collectors. Viele Entwickler nutzen Collectors.toList(), toSet() oder joining() nahezu täglich – selten jedoch im Wissen darum, ob und wie sich diese Collectoren in einem parallelen Kontext verhalten.

Ein einfaches Beispiel: Der Collector Collectors.toList() nutzt intern eine ArrayList. Diese ist nicht thread-sicher. Folglich ist das Ergebnis bei paralleler Verwendung potenziell inkonsistent, sofern nicht intern für Isolation der Zwischenspeicher gesorgt ist.

public static 
Collector> toList() {
   return new CollectorImpl<>(ArrayList::new, List::add,
                              (left, right) -> { left.addAll(right); return left; },
                              CH_ID);
}

Tatsächlich funktioniert dieser Collector in parallelen Streams dennoch korrekt, weil die Streams-API jedem Thread seinen eigenen Akkumulationsbereich zuteilt und erst am Ende über einen kombinierten Merge-Prozess zusammenführt. Der entscheidende Punkt liegt also nicht in der Datenstruktur selbst, sondern in ihrer kontrollierten Isolierung.

Weniger robust zeigt sich Collectors.groupingBy(...). Diese Variante basiert auf einer HashMap, die nicht für gleichzeitigen Zugriff ausgelegt ist. Wird dieser Collector ohne Schutzmaßnahmen in einem parallelStream() eingesetzt, drohen Race Conditions. Die Standardlösung dafür lautet Collectors.groupingByConcurrent(...), die intern auf ConcurrentHashMap setzt und somit für gleichzeitigen Zugriff konzipiert ist.

public static 
Collector>>
groupingByConcurrent(Function super T, ? extends K> classifier) {
   return groupingByConcurrent(classifier, ConcurrentHashMap::new, toList());
}

Ein Blick auf die Signatur dieser Methode zeigt bereits die Intention:

Map> result = namen.parallelStream()
    .collect(Collectors.groupingByConcurrent(String::length));

In diesem Beispiel werden Strings nach ihrer Länge gruppiert – in einer parallel verarbeitbaren Weise. Entscheidend ist, dass sowohl die Map-Implementierung als auch der Akkumulationsprozess thread-safe sind.

Ebenso interessant ist Collectors.toConcurrentMap(...), der explizit dafür vorgesehen ist, große Mengen von Key-Value-Paaren parallel zu aggregieren. Hier ist die Kombination von Schlüsselkonflikten und der richtige Umgang mit Merge-Funktionen von besonderem Interesse.

Die Erkenntnis aus diesen Beispielen lautet: Nicht jeder Standard-Collector ist per se für Parallelität geeignet. Nur weil eine Methode aus dem Collectors-Baukasten stammt, bedeutet das nicht, dass sie in jeder Ausführungskonfiguration korrekt funktioniert. Der Kontext entscheidet – und mit ihm die verwendete Datenstruktur, das Verhalten des combiner und die Art der Akkumulation.

Wer also aus einem Stream nicht nur ein beliebiges Ergebnis, sondern ein korrektes und performantes Ergebnis ziehen will, sollte die Wahl seines Collectors ebenso sorgfältig treffen wie das Filterkriterium am Anfang der Pipeline.

Eigene parallele Collector-Implementierungen

So mächtig die vorgefertigten Collectors der Java-Standardbibliothek auch sein mögen, manchmal reichen sie für spezifische Anforderungen nicht aus. Besonders wenn domänenspezifische Aggregationen, spezialisierte Datenstrukturen oder nicht-triviale Reduktionslogik benötigt werden, lohnt sich ein Blick auf die Möglichkeit, eigene Collector-Implementierungen zu erstellen.

In der Regel lässt sich ein eigener Collector mit der statischen Methode Collector.of(...) erstellen. Diese Methode erwartet fünf Parameter: einen Supplier, der einen neuen Akkumulator erzeugt; einen BiConsumer, der ein Element in den Akkumulator einfügt; einen BinaryOperator zum Kombinieren zweier Akkumulatoren; optional eine Function zur Konvertierung des Ergebnisses; und schließlich ein Array Collector.Characteristics..., das Metainformationen wie CONCURRENT oder UNORDERED bereitstellt.

Collector> toConcurrentQueue() {
    return Collector.of(
        ConcurrentLinkedQueue::new,
        Queue::add,
        (left, right) -> { left.addAll(right); return left; },
        Collector.Characteristics.CONCURRENT, Collector.Characteristics.UNORDERED
    );
}

Das Subreddit r/wallstreetbets ist ein riesiges Forum, auf dem die Mitglieder täglich interessante Aktien und die Entwicklungen am Markt diskutieren. Dabei entsteht kontinuierlich eine enorme Informationsmenge, die sich kaum überblicken lässt. Wertvolle Unterstützung kann eine Text-KI liefern, die ein Python-Skript mit Anweisungen und Informationen füttert.

In diesem Projekt erklären wir Schritt für Schritt, wie Sie eine KI zur Analyse der Gespräche über die Aktien erstellen, die im Forum gerade populär sind. Als Ergebnis liefert das Projekt ein Word-Dokument mit kurzen Diskussionsanalysen. Zu den jeweils relevanten Posts gibt die KI eine Einschätzung, ob die Diskussion auf eine positive oder negative Kursentwicklung hindeutet, begründet die Einschätzung mit den Inhalten aus den Gesprächen und hebt wichtige inhaltliche Punkte hervor.

Das Projekt eignet sich für Einsteiger in der Programmierung. Die Python-Skripte lassen wir auf Basis von Prompts von einer KI schreiben. Den Code können Sie herunterladen und für sich an wenigen Stellen anpassen. Das Projekt basiert auf unserem Artikel zum Reddit-Aktien-Crawler. Wir raten dazu, dass Sie zunächst das erste Projekt abschließen, bevor Sie die Erweiterung programmieren. Die Skripte haben wir in Windows 11 mit Python 3.13.5 erstellt.

Das war die Leseprobe unseres heise-Plus-Artikels „KI-Analyse für Aktien-Diskussionen auf Reddit: Anleitung zum Nachbauen“.
Mit einem heise-Plus-Abo können Sie den ganzen Artikel lesen.

Sogenannte Influencer sollen allein den Fiskus in Nordrhein-Westfalen um rund 300 Millionen Euro betrogen haben. Das bestätigte das Landesamt zur Bekämpfung der Finanzkriminalität (LBF NRW) auf Anfrage. Die Steuerfahnder analysieren nach eigenen Angaben aktuell ein Paket mehrerer Social-Media-Plattformen mit 6000 Datensätzen, aus denen sich der mutmaßliche Millionen-Schaden ergibt. Nach dpa-Informationen sind auch andere Bundesländer betroffen.

„Ziel der Ermittlungen sind professionelle Influencer, die ihre steuerlichen Pflichten mit hoher krimineller Energie umgehen“, hieß es von dem Landesamt, das seit Januar 2025 die gesamte nordrhein-westfälische Steuerfahndung mit rund 1200 Experten vereint. Es ist nach eigenen Angaben die erste Landesbehörde dieser Art in der Bundesrepublik. Innerhalb des LBF hatte man extra ein „Influencer-Team“ gebildet, um die Szene unter die Lupe zu nehmen.

„Die großen Fische im Visier“

„Im Fokus unseres Influencer-Teams stehen ausdrücklich nicht junge Menschen, die ein paar Follower gesammelt und ein paar Cremes oder Kleider beworben haben“, so Behördenleiterin Stephanie Thien: „Das LBF NRW hat auch auf den sozialen Netzwerken die großen Fische im Visier.“

Thien betonte: „Es gibt bei den großen Social-Media-Profilen Akteurinnen und Akteure, die mit hoher krimineller Energie jegliche Steuerverpflichtung zu umgehen versuchen. Es ist keine Seltenheit,dass eine Influencerin oder ein Influencer pro Monat mehrere zehntausend Euro verdient, aber nicht einmal eine Steuernummer hat. Da geht es nicht um Überforderung mit plötzlichem Ruhm, sondern umimmense Steuerhinterziehung mit Wissen und Willen.“

Die Ermittlungen seien aufwendig, so Thien: „Denn einen festen Arbeitsplatz gibt es nicht, oftmals melden sich die Content-Creators mit steigenden Umsätzen ins Ausland ab, um dem Finanzamt zuentgehen.“ So sei Dubai ein beliebtes Ziel.

Bei Werbung in sogenannten „Storys“, die nach 24 Stunden wieder verschwinden, sei die Beweisführung zudem schwierig. NRW habe aber „Ermittlungsmethoden initiiert, um Werbepartnerschaften und-einnahmen zurückverfolgen und beweissicher nachweisen zu können“, sagte Thien. Andere Länder hätten sich das zum Vorbild genommen.

Behörde führt bereits 200 Verfahren

Das LBF führt bereits rund 200 laufende Strafverfahren gegen in Nordrhein-Westfalen lebende Influencerinnen und Influencer – die Fälle aus dem aktuellen Datenpaket sind dabei noch nicht eingerechnet. Durchschnittlich geht es laut LBF um einen hohen fünfstelligen steuerlichen Fehlbetrag, in Einzelfällen auch um Fehlbeträge in Millionenhöhe.

NRW-Finanzminister Marcus Optendrenk (CDU) sagte der dpa: „Steuerhinterziehung im großen Stil tritt überall dort auf, wo Geld in großem Stil gemacht wird. Es ist ein Gebot der Gerechtigkeit, dass unsere Steuerfahndung ganz genau hinschaut.“

(dmk)