post

RxJS in Node.js: Ereignisströme reaktiv verarbeiten

RxJS in Node.js: Wie fromEvent einen EventEmitter in ein Observable verwandelt, Operatoren serverseitige Ströme deklarativ transformieren und warum RxJS keine echte Backpressure hat.

In meinen Node.js-Schulungen kommt regelmäßig der Moment, in dem ein Teilnehmer eine Verarbeitung gebaut hat, die funktioniert – aber niemand mehr versteht, wie. Es beginnt harmlos mit einem emitter.on('data', ...). Dann kommt der Wunsch, nur bestimmte Ereignisse durchzulassen. Dann sollen sie pro Sekunde gebündelt gespeichert werden. Dann kommt ein zweiter Konsument dazu. Und plötzlich stehen fünf verschachtelte Callbacks im Code, mit einem Zwischenzustand in Modulvariablen, den keiner mehr sauber testen kann.

Genau an dieser Stelle setze ich RxJS ins Spiel. Nicht als Selbstzweck und nicht für jede Node-Anwendung, aber überall dort, wo aus einem Ereignisstrom ein komponierbares Programm werden soll. Wer sich das imperative Fundament schon angesehen hat – etwa im Beitrag zu EventEmitter und Ereignismodell – erkennt in RxJS dieselbe Idee, nur von der anderen Seite aufgezäumt.

Von on/emit zu deklarativen Strömen

Der EventEmitter ist imperativ. Ich sage: „Wenn data passiert, führe diese Funktion aus.“ Das ist ein Befehl. Jeder Handler ist eine eigene kleine Insel, und wenn ich mehrere Schritte verketten will, verknüpfe ich sie von Hand über gemeinsamen Zustand.

RxJS beschreibt denselben Ereignisstrom deklarativ. Ich sage nicht mehr „tu dies, wenn jenes passiert“, sondern „dieser Strom ist die Quelle, transformiert durch diese Kette von Schritten“. Die Kette ist eine Beschreibung, kein Ablauf. Erst beim subscribe wird sie aktiviert.

Die Brücke zwischen beiden Welten ist fromEvent. Der Operator nimmt einen Node-EventEmitter und einen Ereignisnamen und liefert ein Observable:

const { EventEmitter } = require('events');
const { fromEvent } = require('rxjs');

const emitter = new EventEmitter();
const data$ = fromEvent(emitter, 'data');

const subscription = data$.subscribe({
  next: value => console.log('received', value),
});

emitter.emit('data', { level: 'info', message: 'server started' });

fromEvent(emitter, 'data') ändert nichts an der Semantik des Emitters. Es registriert bei jedem subscribe einen Listener über addListener und entfernt ihn bei unsubscribe wieder. Das Observable ist damit ein Adapter: Der Emitter bleibt die Quelle, RxJS legt nur eine komponierbare Schicht darüber. Wer den Unterschied zwischen dem Push-Modell von Observables und dem Pull-Modell von Iteratoren vertiefen will, findet das im Beitrag zu RxJS: Push und Pull.

Ein Hinweis zur Konvention: Ich hänge an Variablen, die ein Observable halten, ein $ an (data$). Das ist kein Zwang, hat sich in der RxJS-Welt aber eingebürgert und macht im Code sofort sichtbar, was ein Strom ist und was ein einzelner Wert.

Die Pipeline: Operatoren als reine Funktionen

Der eigentliche Gewinn liegt in den Operatoren. Sie sind reine Funktionen, die ein Observable entgegennehmen und ein neues zurückgeben. Verkettet werden sie über .pipe(...). In RxJS 7 importiere ich alles – Erzeugungsfunktionen wie fromEvent und Operatoren wie map – direkt aus 'rxjs'. Der frühere Import aus 'rxjs/operators' funktioniert noch, ist aber überholt.

Ein typisches serverseitiges Beispiel ist Log-Verarbeitung. Ich möchte eingehende Rohzeilen zu Objekten parsen, nur die Fehler behalten und diese pro Sekunde gebündelt wegschreiben, statt bei jeder einzelnen Zeile die Datenbank anzufassen:

const { fromEvent } = require('rxjs');
const { map, filter, bufferTime } = require('rxjs');

const logStream$ = fromEvent(logEmitter, 'line').pipe(
  map(line => JSON.parse(line)),
  filter(entry => entry.level === 'error'),
  bufferTime(1000),
);

const subscription = logStream$.subscribe({
  next: batch => saveErrors(batch),
  error: err => console.error('stream failed', err),
});

Was hier steht, liest sich fast wie eine Beschreibung der Anforderung. map transformiert jeden Wert, filter lässt nur passende durch, bufferTime(1000) sammelt über ein Zeitfenster von einer Sekunde und emittiert dann ein Array. Der next-Handler bekommt also nicht einzelne Einträge, sondern jeweils ein Bündel – oder ein leeres Array, wenn im Fenster nichts kam.

Jeder Schritt ist für sich verständlich und für sich testbar. Und wenn morgen die Anforderung kommt, zusätzlich nach Dienst zu gruppieren, füge ich einen Operator ein, statt eine verschachtelte Callback-Struktur umzubauen.

Ein zweites Muster aus der Praxis: Sensordaten drosseln und dabei einen laufenden Mittelwert bilden. Angenommen, ein Gerät feuert deutlich häufiger, als ich verarbeiten will:

const { fromEvent } = require('rxjs');
const { throttleTime, scan } = require('rxjs');

const average$ = fromEvent(sensor, 'reading').pipe(
  map(event => event.value),
  throttleTime(500),
  scan((sum, value, index) => (sum * index + value) / (index + 1), 0),
);

average$.subscribe({ next: avg => console.log('running average', avg) });

throttleTime(500) lässt einen Wert durch und ignoriert dann für 500 Millisekunden alle weiteren. scan verhält sich wie reduce, emittiert aber jeden Zwischenstand – so bekomme ich bei jeder durchgelassenen Messung den aktualisierten Durchschnitt. Diese Kombination aus laufender Aggregation und Zeitfenster ist im Repository unter operators.js vorgezeichnet, wenn auch dort noch mit der älteren RxJS-6-Importweise.

Wenn ich pro Ereignis eine asynchrone Weiterverarbeitung anstoßen will – etwa Details nachladen – hilft mergeMap. Es mappt jeden Wert auf ein inneres Observable und flacht die Ergebnisse nebenläufig ab. Der zweite Parameter begrenzt, wie viele innere Verarbeitungen gleichzeitig laufen dürfen:

const { mergeMap } = require('rxjs');

const details$ = ids$.pipe(
  mergeMap(id => fetchDetails(id), 4),
);

Mit mergeMap(id => fetchDetails(id), 4) laufen höchstens vier Abrufe parallel. Diese Begrenzung ist wichtig, und zwar aus einem Grund, der gleich noch eine eigene Rolle spielt.

Der grobe Fluss durch so eine Pipeline sieht immer gleich aus:

flowchart LR
  E[EventEmitter] -->|emit data| F["fromEvent<br/>(Observable)"]
  F --> M[map]
  M --> FI[filter]
  FI --> B[bufferTime]
  B --> S[Subscriber]

Ein Subject: aus einem Strom viele Konsumenten machen

Bis hierher hatte jeder Strom genau einen Verbraucher. Sobald mehrere Teile der Anwendung denselben Ereignisstrom brauchen, kommt das Subject ins Spiel. Ein Subject ist zugleich Observer und Observable: Ich kann Werte hineingeben und andere können es abonnieren. Es verteilt jeden Wert an alle Subscriber gleichzeitig – Multicast.

const { fromEvent, Subject } = require('rxjs');

const bus = new Subject();

// one source feeds the bus
fromEvent(emitter, 'data').subscribe(bus);

// multiple consumers share the same stream
const loggerSub = bus.subscribe({ next: v => logger.write(v) });
const metricsSub = bus.subscribe({ next: v => metrics.record(v) });

// clean up later
loggerSub.unsubscribe();
metricsSub.unsubscribe();

Der Unterschied zum direkten Abonnieren des Quell-Observables ist grundlegend. Würden logger und metrics beide fromEvent(emitter, 'data') abonnieren, hinge am Emitter zweimal ein Listener und die Quelle würde zweimal ausgeführt. Über das Subject gibt es eine Ausführung, deren Werte an alle Subscriber verteilt werden.

Genau das ist der Kern der Unterscheidung zwischen Unicast und Multicast:

flowchart LR
  P[Producer] --> SUB[Subject]
  SUB --> C1[Logger-Subscriber]
  SUB --> C2[Metrics-Subscriber]
  SUB --> C3[Audit-Subscriber]

Neben dem einfachen Subject gibt es Varianten, die je nach Anwendungsfall passen:

  • BehaviorSubject(seed) hält den letzten Wert und liefert ihn jedem neuen Subscriber sofort. Praktisch für Zustand, den ein spät hinzukommender Konsument kennen muss.
  • ReplaySubject(bufferSize) puffert die letzten n Werte und spielt sie neuen Subscribern vor.
  • AsyncSubject gibt nur den allerletzten Wert aus, und das erst beim complete.

Im Repository unter subject.js sieht man das Grundprinzip: dieselbe Quelle, mehrere subscribe-Aufrufe. Dort steht noch die positionsbasierte subscribe(next, error, complete)-Form. In RxJS 7 nutze ich stattdessen durchgängig den Objekt-Observer subscribe({ next, error, complete }) – dazu gleich mehr bei den Fallstricken.

Hot und cold: der Unterschied, der Ergebnisse verfälscht

Ein Punkt, an dem in Schulungen zuverlässig Verwirrung entsteht, ist die Frage, wann die Quelle eines Observables entsteht.

Ein kaltes Observable erzeugt seinen Producer pro subscribe. Jeder Subscriber bekommt eine eigene, frische Ausführung – Unicast. Ein from([1, 2, 3]) ist kalt: Jeder, der es abonniert, sieht die Werte 1, 2, 3 von vorne.

Ein heißes Observable hat seinen Producer außerhalb. Die Werte entstehen unabhängig davon, ob jemand zuhört. Wer spät abonniert, verpasst, was vorher lief – Multicast. Ein Subject ist heiß, und ein fromEvent über einen bereits laufenden Emitter verhält sich ebenfalls heiß: Die Ereignisse feuern ohnehin, ich klinke mich nur ein.

Diese Unterscheidung ist keine Theorie. Wer erwartet, dass „jeder Subscriber frisch startet“, aber tatsächlich einen heißen Strom vor sich hat, bekommt bei spät hinzukommenden Konsumenten schlicht falsche Ergebnisse – fehlende Anfangswerte, eine Aggregation, die mitten im Strom beginnt. Wenn ich will, dass auch ein später Subscriber den letzten Stand oder die Historie sieht, ist genau das der Fall für BehaviorSubject, ReplaySubject oder den Operator shareReplay.

Die wichtige Abgrenzung: RxJS hat keine echte Backpressure

Jetzt kommt der Punkt, den ich in Schulungen bewusst früh und deutlich mache, weil er sonst zu falschen Erwartungen führt. Wer Node-Streams kennt, kennt Backpressure: Ein lesbarer Stream kann pausieren, ein highWaterMark begrenzt den Puffer, pipe regelt den Fluss zwischen Producer und Konsument automatisch. Der schnelle Producer wird gebremst, wenn der langsame Konsument nicht hinterherkommt.

RxJS hat das nicht. Es gibt keinen eingebauten Mechanismus, mit dem ein Subscriber die Quelle zum Langsamerwerden zwingt. Ein Subject oder ein fromEvent pusht Werte, sobald sie da sind, egal ob die Verarbeitung mitkommt.

Was RxJS stattdessen anbietet, sind zeitbasierte und puffernde Operatoren:

  • throttleTime(ms) lässt einen Wert durch und verwirft dann für eine Weile alle weiteren.
  • debounceTime(ms) wartet, bis eine Ruhephase eintritt, und emittiert dann den letzten Wert.
  • sampleTime(ms) greift in festen Abständen den jeweils aktuellen Wert ab.
  • bufferTime(ms) sammelt über ein Fenster und liefert Bündel.
  • das concurrent-Limit von mergeMap begrenzt die Zahl paralleler innerer Verarbeitungen.

Der entscheidende Unterschied: Diese Operatoren verwerfen oder aggregieren Werte, aber sie bremsen die Quelle nicht. Der Producer feuert unverändert weiter; ich reduziere nur, was hinten ankommt. Für eine Log-Anzeige oder Sensordaten ist das oft genau richtig – ein paar Zwischenwerte wegzuwerfen ist unkritisch. Für eine Verarbeitung, bei der jeder einzelne Wert zählen muss und die Quelle tatsächlich gedrosselt gehört, sind Node-Streams das richtige Werkzeug, nicht RxJS.

Man kann beide Welten verbinden. Das Repository zeigt unter node.js die Brücke über rxjs-stream:

const { spawn } = require('child_process');
const { streamToRx } = require('rxjs-stream');
const { map } = require('rxjs');

streamToRx(spawn('ls', ['-lha']).stdout)
  .pipe(map(chunk => chunk.toString()))
  .subscribe({ next: text => console.log(text) });

streamToRx macht aus einem Node-Readable ein Observable und behält dabei die Stream-Backpressure im unteren Teil. So bekomme ich die Komponierbarkeit von RxJS für die Transformation, ohne die Flusskontrolle des Streams aufzugeben. Wer die Grundlagen des Event-Loops und der Stream-Architektur dahinter auffrischen will, findet sie im Node.js-Training.

Fallstricke, die ich in jeder Schulung sehe

Zwei Fehler tauchen so regelmäßig auf, dass ich sie hier ausdrücklich nenne.

Der erste ist der Subscription-Leak. fromEvent registriert bei jedem subscribe einen Listener am Emitter. Ohne unsubscribe() bleibt dieser Listener hängen – auch dann, wenn die Verarbeitung längst nicht mehr gebraucht wird. Das Ergebnis ist ein Memory-Leak und eine Pipeline, die unbemerkt weiterläuft. Die Regel ist einfach: Ich halte die Subscription, die .subscribe() zurückgibt, und rufe unsubscribe() auf, wenn ich fertig bin. Alternativ begrenze ich den Strom deklarativ über take(n) oder takeUntil(stop$), sodass er sich selbst beendet:

const { fromEvent, timer } = require('rxjs');
const { takeUntil } = require('rxjs');

const stop$ = timer(60000);

fromEvent(emitter, 'data')
  .pipe(takeUntil(stop$))
  .subscribe({ next: v => process(v) });

Hier räumt takeUntil nach 60 Sekunden auf, ohne dass ich die Subscription selbst verwalten muss. Bei langlebigen Servern ist eine der beiden Strategien Pflicht, sonst sammeln sich Listener über die Laufzeit an.

Der zweite Fallstrick ist das Verwechseln von hot und cold, das ich oben schon beschrieben habe – es äußert sich fast immer als „mein zweiter Subscriber bekommt nicht dieselben Werte“. Die Antwort ist meist ein Subject mit passender Variante oder shareReplay.

Und ein dritter, eher handwerklicher Punkt: Viele Beispiele im Netz und auch die älteren Snippets im Repository nutzen noch RxJS-6-Muster. Konkret heißt das für RxJS 7:

  • Statt subscribe(next, error, complete) mit einzelnen Argumenten den Objekt-Observer subscribe({ next, error, complete }) verwenden – die positionsbasierte Form ist seit 6.4 überholt.
  • Statt tap(fn, errFn) mit mehreren Callbacks ebenfalls den Objekt-Observer nutzen.
  • Operatoren aus 'rxjs' importieren statt aus 'rxjs/operators'.

Diese Anpassungen sind mechanisch, aber wichtig, damit der Code nicht auf einem Muster aufsetzt, das perspektivisch verschwindet.

Wann RxJS statt rohem EventEmitter?

Nicht jede Node-Anwendung braucht RxJS. Ein einzelner on('data', ...)-Handler bleibt mit purem EventEmitter schlanker und hat keine zusätzliche Abhängigkeit. Ich greife dann zu RxJS, wenn mindestens einer dieser Punkte zutrifft:

  • Ich muss mehrere Verarbeitungsschritte verketten – transformieren, filtern, aggregieren – und will das nicht über verschachtelte Callbacks und geteilten Zustand lösen.
  • Ich brauche zeitbasierte Steuerung: drosseln, entprellen, in Fenstern bündeln.
  • Denselben Ereignisstrom sollen mehrere Konsumenten unabhängig verarbeiten (Multicast über ein Subject).
  • Der Strom soll sich sauber und deklarativ beenden lassen, etwa über takeUntil.
  • Der Verarbeitungscode soll in kleine, für sich testbare Bausteine zerfallen.

Umgekehrt bleibe ich beim rohen Emitter oder greife zu Node-Streams, wenn es nur ein einziger simpler Handler ist oder wenn ich echte Backpressure über den ganzen Weg brauche.

Fazit

RxJS verschiebt in Node.js die Perspektive vom Befehl zur Beschreibung. Der EventEmitter sagt, was bei jedem Ereignis zu tun ist; RxJS beschreibt den ganzen Ereignisstrom als komponierbare Kette von reinen Funktionen. fromEvent ist die Brücke dorthin, ohne dass sich an der Semantik des Emitters etwas ändert, und ein Subject macht aus einem Strom viele Konsumenten.

Zwei Dinge sollte man dabei nie vergessen. Erstens die Disziplin beim Aufräumen: Ohne unsubscribe, take oder takeUntil bleiben Listener und Verarbeitung am Leben. Zweitens die ehrliche Grenze des Werkzeugs: RxJS steuert Überlast nur zeitbasiert und puffernd, nicht über echte Backpressure. Wer diese beiden Punkte im Kopf behält, bekommt mit RxJS eine sehr angenehme Art, serverseitige Ereignisströme zu verarbeiten – lesbar, testbar und erweiterbar, ohne im Callback-Dickicht zu versinken.

Weiterführende Quellen

Kommentare

Kommentar schreiben