Die vier Subject-Typen in RxJS

In meinen RxJS-Schulungen kommt fast immer derselbe Moment. Wir haben Observables verstanden, wir haben die Operatoren durchgespielt, und dann fällt der Satz: "Aber wie schiebe ich denn selbst einen Wert in einen Stream hinein?" Ein normales Observable ist dafür nicht gebaut – es beschreibt eine Quelle, die man abonniert, und es startet seine Arbeit erst dann, verzögert, pro Subscription neu. Genau an dieser Stelle betritt das Subject die Bühne.

Ein Subject ist der Baustein, mit dem man aus imperativem Code heraus in die reaktive Welt hineinruft. Und sobald das klar ist, kommt die nächste Frage: Warum gibt es davon gleich vier Varianten? Die kurze Antwort lautet: Sie unterscheiden sich im Gedächtnis. Die lange Antwort ist dieser Artikel.

Wer die Grundlagen des Push-Modells noch einmal auffrischen möchte, findet sie in RxJS: Push und Pull. Hier setze ich voraus, dass Observable und Subscription vertraut sind, und konzentriere mich auf die Subjects. Stand ist RxJS 7.

Ein Subject ist beides zugleich

Der entscheidende Satz über Subjects steht schon in der Klassendefinition: Subject<T> erweitert Observable<T> und implementiert zugleich das Observer-Interface. Das heißt, ein Subject ist Observable und Observer in einer einzigen Instanz. Man kann es abonnieren wie jedes Observable – und man kann per next, error und complete Werte hineinfüttern wie in einen Observer.

import { Subject } from 'rxjs';

const subject = new Subject<number>();

// als Observable: abonnieren
subject.subscribe((value) => console.log('received', value));

// als Observer: Werte hineingeben
subject.next(1);
subject.next(2);
subject.complete();

Diese Doppelrolle hat eine wichtige Konsequenz. Ein reines Observable ist kalt und unicast: Jeder Subscriber bekommt seine eigene Ausführung, der Code im Observable läuft pro Subscription erneut. Ein Subject dagegen ist multicast. Es hält eine Liste seiner Subscriber, und wenn next aufgerufen wird, geht derselbe Wert an alle gleichzeitig. Es gibt eine Quelle und viele Empfänger.

flowchart LR
  P["Quelle<br/>next / error / complete"] --> S(("Subject"))
  S --> A["Subscriber A"]
  S --> B["Subscriber B"]
  S --> C["Subscriber C"]

Damit hängt der zweite zentrale Charakterzug zusammen: Subjects sind hot. Werte fließen unabhängig davon, ob gerade jemand zuhört. Rufe ich subject.next(1) auf, bevor irgendein Subscriber existiert, ist dieser Wert bei einem einfachen Subject schlicht weg. Er wurde emittiert, aber niemand war da, um ihn entgegenzunehmen. Genau dieser Punkt – was passiert mit Werten, die vor einer Subscription geflossen sind – trennt die vier Typen voneinander.

Die vier Typen und ihr Gedächtnis

Bevor wir in die Details gehen, hier der Überblick. Alle vier erben von Subject, alle vier sind hot und multicast. Der Unterschied liegt darin, was ein Subscriber sieht, der zu spät kommt:

Ein Subject hat kein Gedächtnis. Wer nach einem next abonniert, hat den Wert verpasst. Nur was ab dem Subscribe-Zeitpunkt fließt, kommt an.
Ein BehaviorSubject merkt sich den letzten Wert und braucht einen Initialwert. Jeder neue Subscriber bekommt sofort den aktuellen Stand – ideal für den Fall "aktueller Zustand".
Ein ReplaySubject puffert die letzten n Werte und spielt sie jedem neuen Subscriber ab, sogar nach complete.
Ein AsyncSubject emittiert nur den allerletzten Wert, und erst, wenn complete aufgerufen wurde.

Diese vier Zeilen sind der Kern des Artikels. Alles Weitere ist Ausbuchstabieren. Schauen wir uns jeden Typ mit einem kleinen Timing-Beispiel an, denn das Timing ist genau das, worauf es ankommt.

Subject – der Vergessliche

Beginnen wir mit dem einfachen Subject. Es hat kein Gedächtnis. Um das sichtbar zu machen, abonnieren wir zu unterschiedlichen Zeitpunkten und beobachten, wer was mitbekommt.

import { Subject } from 'rxjs';

const subject = new Subject<number>();

subject.next(1); // niemand hört zu -> verloren

subject.subscribe((v) => console.log('A', v));
subject.next(2);

subject.subscribe((v) => console.log('B', v));
subject.next(3);

// Ausgabe:
// A 2
// A 3
// B 3

Der Wert 1 ist verpufft, weil zum Zeitpunkt seiner Emission kein Subscriber existierte. Subscriber A steigt vor der 2 ein und sieht deshalb 2 und 3. Subscriber B kommt erst nach der 2 und sieht nur noch 3. Das ist kein Bug, das ist die Definition. Ein Subject ist ein reiner Verteiler für den Moment.

Wofür ist das gut? Für echte Ereignisse, bei denen Vergangenheit keine Rolle spielt. Ein Klick, ein WebSocket-Frame, ein Kommando, das ausgelöst wird. Wer nicht zuhört, wenn das Ereignis passiert, hat es eben nicht miterlebt – und das ist völlig in Ordnung.

BehaviorSubject – der Zustandshalter

Jetzt der Typ, den ich in der Praxis am häufigsten sehe und empfehle: das BehaviorSubject. Es unterscheidet sich in zwei Punkten. Erstens braucht sein Konstruktor einen Initialwert – man kann kein BehaviorSubject ohne definierten Startzustand erzeugen. Zweitens merkt es sich stets den zuletzt emittierten Wert und liefert ihn synchron an jeden neuen Subscriber aus.

import { BehaviorSubject } from 'rxjs';

const state$ = new BehaviorSubject<number>(0); // Initialwert Pflicht

state$.subscribe((v) => console.log('A', v)); // A bekommt sofort 0
state$.next(1);

state$.subscribe((v) => console.log('B', v)); // B bekommt sofort 1
state$.next(2);

// Ausgabe:
// A 0
// A 1
// B 1
// A 2
// B 2

Subscriber B steigt spät ein und bleibt trotzdem nicht leer. Er erhält beim Abonnieren sofort den aktuellen Wert 1 und danach den Rest. Genau das macht das BehaviorSubject zum Standardwerkzeug für Zustand. Ein Zustand hat immer einen aktuellen Wert, und wer ihn abfragt, will diesen Wert – nicht die Aussicht, irgendwann beim nächsten next etwas zu erfahren.

Praktisch dazu: das BehaviorSubject bietet mit getValue() und der Property value einen synchronen Zugriff auf den aktuellen Stand, ohne zu abonnieren.

const counter$ = new BehaviorSubject<number>(0);
counter$.next(42);

console.log(counter$.getValue()); // 42
console.log(counter$.value); // 42

Wer schon einmal in RxJS in React mit Streams als Datenquelle gearbeitet hat, kennt die Situation: Eine Komponente montiert und abonniert, und sie soll natürlich sofort etwas anzeigen können. Mit einem BehaviorSubject ist genau das garantiert.

Ein Detail, das gern übersehen wird: Nach complete() sendet ein BehaviorSubject seinen letzten Wert nicht mehr an neue Subscriber. Es liefert dann nur noch das Complete-Signal. Der aktuelle Wert steht also nur so lange bereit, wie der Stream aktiv ist.

ReplaySubject – das Gedächtnis mit Puffer

Das BehaviorSubject merkt sich genau einen Wert. Manchmal braucht man mehr Historie. Dafür gibt es das ReplaySubject. Sein Konstruktor nimmt eine Puffergröße entgegen und spielt diese letzten n Werte jedem neuen Subscriber ab.

import { ReplaySubject } from 'rxjs';

const replay$ = new ReplaySubject<number>(2); // die letzten 2 Werte puffern

replay$.next(1);
replay$.next(2);
replay$.next(3);

replay$.subscribe((v) => console.log('late', v));
replay$.next(4);

// Ausgabe:
// late 2
// late 3
// late 4

Der spät hinzukommende Subscriber bekommt nachträglich 2 und 3 – die beiden letzten gepufferten Werte – und danach 4 live. Der Wert 1 fällt aus dem Puffer, weil dieser nur zwei Plätze hat.

Zwei Eigenschaften machen das ReplaySubject besonders. Erstens gibt es neben der Puffergröße einen optionalen Zeitfenster-Parameter: new ReplaySubject<number>(bufferSize, windowTime). Damit werden nur Werte abgespielt, die innerhalb der letzten x Millisekunden geflossen sind. Zweitens – und das ist der große Unterschied zum BehaviorSubject – spielt das ReplaySubject seinen Puffer auch nach complete() noch ab.

const cache$ = new ReplaySubject<number>(3);

cache$.next(10);
cache$.next(20);
cache$.next(30);
cache$.complete();

// nach complete abonnieren
cache$.subscribe((v) => console.log('after complete', v));

// Ausgabe:
// after complete 10
// after complete 20
// after complete 30

Das eignet sich gut als kleiner Cache: Ein Ergebnis wird einmal produziert und dann jedem, der später fragt, erneut ausgeliefert. Vorsicht ist allerdings bei einer Puffergröße von Infinity geboten – die ist der Default. Ein ReplaySubject ohne Größenangabe merkt sich jeden jemals emittierten Wert und wächst unbegrenzt. In langlebigen Streams ist das eine Quelle für Speicherlecks. Deshalb setze ich die Puffergröße bewusst.

AsyncSubject – nur das Endergebnis

Der vierte Typ ist der eigenwilligste. Das AsyncSubject emittiert nur einen einzigen Wert – den letzten – und das auch erst, wenn complete() aufgerufen wird. Solange kein complete kommt, gibt es nichts.

import { AsyncSubject } from 'rxjs';

const async$ = new AsyncSubject<number>();

async$.subscribe((v) => console.log('result', v));

async$.next(1);
async$.next(2);
async$.next(3);
async$.complete(); // erst jetzt wird emittiert

// Ausgabe:
// result 3

Die Zwischenwerte 1 und 2 interessieren das AsyncSubject nicht. Es merkt sich nur den jeweils letzten, und beim complete gibt es diesen einen an alle Subscriber weiter – auch an die, die erst nach dem complete abonnieren. Das Verhalten erinnert bewusst an eine Promise: eine Berechnung, die läuft und am Ende genau ein Ergebnis liefert.

Der häufigste Stolperstein hier ist das fehlende complete. Ohne diesen Aufruf emittiert das AsyncSubject niemals. In der Schulung ist das regelmäßig der Moment, in dem jemand ratlos auf eine leere Konsole schaut – die Werte sind geflossen, aber ohne den Abschluss gibt das Subject nichts heraus.

Wann welcher – die Entscheidung

Fassen wir die Wahl zusammen. Die Frage, die man sich stellen sollte, lautet immer: Was soll ein Subscriber sehen, der zu spät kommt?

Es geht um ein Ereignis, und Vergangenheit ist irrelevant? Subject.
Es geht um einen aktuellen Zustand, der immer verfügbar sein muss? BehaviorSubject.
Neue Subscriber sollen die letzten n Werte oder ein zeitliches Fenster nachgeliefert bekommen? ReplaySubject.
Es interessiert nur ein einziges Endergebnis nach Abschluss? AsyncSubject.

In der Praxis dominieren zwei davon. BehaviorSubject für Zustand, Subject für Ereignisse. ReplaySubject und AsyncSubject sind Spezialwerkzeuge für konkrete Bedürfnisse.

Zwei Fallstricke aus der Praxis

Zum Schluss die zwei Fehler, die mir am häufigsten begegnen. Beide sind leicht zu vermeiden, wenn man sie einmal gesehen hat.

Der erste ist die Verwechslung von Subject und BehaviorSubject bei Zustand. Man baut einen Service, der einen aktuellen Wert hält, verwendet aber ein einfaches Subject. Dann montiert eine Komponente, abonniert – und bleibt leer, weil das letzte next schon vorbei war. Der Fehler äußert sich als "manchmal ist der Wert da, manchmal nicht", je nach Timing. Für Zustand gilt: immer BehaviorSubject, damit es einen definierten aktuellen Wert gibt.

Der zweite Fallstrick ist subtiler und betrifft die Architektur: Subjects überall statt reiner Observables nach außen zu geben. Wenn ich ein Subject als öffentliche Property exponiere, kann jeder von außen next() oder complete() darauf aufrufen. Damit ist die Kapselung dahin – jede beliebige Stelle im Code kann in den Stream hineinfunken oder ihn abschließen. Die saubere Lösung ist ein privates Subject und ein öffentliches Observable über asObservable():

import { Subject, Observable } from 'rxjs';

interface Command {
  type: string;
}

class CommandBus {
  private commandsSubject = new Subject<Command>();

  // nach außen nur lesbar
  readonly commands$: Observable<Command> = this.commandsSubject.asObservable();

  dispatch(command: Command): void {
    this.commandsSubject.next(command);
  }
}

Von außen kann man commands$ abonnieren, aber nicht hineinschreiben. Das next läuft ausschließlich über die kontrollierte dispatch-Methode. Das ist dasselbe Prinzip, das auch in RxJS in Node.js trägt: Ereignisse fließen über eine definierte Schnittstelle hinein und über abonnierbare Observables wieder hinaus. Ein Subject als Motor im Inneren, ein Observable als Fassade nach außen.

Damit man es nicht falsch versteht: Subjects sind kein Anti-Pattern. Sie sind das richtige Werkzeug, wann immer man von imperativem Code aus in einen Stream schreiben muss. Nur sollte das Subject selbst nach Möglichkeit im Inneren bleiben.

Fazit

Ein Subject ist Observable und Observer in einem und multicastet dieselben Werte an alle seine Subscriber. Es ist hot – Werte fließen, ob jemand zuhört oder nicht. Die vier Varianten unterscheiden sich nur in einer einzigen Dimension: im Gedächtnis für Subscriber, die zu spät kommen. Kein Gedächtnis beim Subject, der letzte Wert beim BehaviorSubject, die letzten n Werte beim ReplaySubject, das Endergebnis nach Abschluss beim AsyncSubject.

Wer diese eine Frage – was sieht ein später Subscriber? – zur Gewohnheit macht, trifft die Wahl fast automatisch. Und wer daneben die Kapselung im Blick behält und Subjects nach innen versteckt, hinter Observables nach außen, hält den Datenfluss überschaubar. Mehr braucht es für den souveränen Umgang mit Subjects nicht.