# Die vier Subject-Typen in RxJS URL: https://www.mikebild.dev/de/blog/rxjs-die-vier-subject-typen/ Ein Subject ist kein Observable mit angeschraubter `next()`-Methode. Es ist ein anderes Modell: hot statt kalt, multicast statt unicast, mit einem Lebenszyklus, der weiterläuft, ob gerade jemand zuhört oder nicht. Wer es nur als bequeme Abkürzung nimmt, um endlich selbst Werte in einen Stream zu schieben, handelt sich damit ein Timing-Verhalten ein, das er nicht bestellt hat. Vor der Wahl zwischen den vier Typen steht deshalb eine andere Frage: Lohnt sich der Stream an dieser Stelle überhaupt, oder tut es ein schlichteres Modell genauso? Wer sie überspringt, baut Reaktivität dort ein, wo eine Variable oder ein Promise gereicht hätte. Wer sie stellt und mit Ja beantwortet, hat danach noch genau eine Entscheidung zu treffen – und die spart, richtig getroffen, eine ganze Klasse von Timing-Bugs, die schwer zu reproduzieren und teuer zu debuggen sind. Ein Subject ist der Baustein, mit dem man aus imperativem Code heraus in die reaktive Welt hineinruft. 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. Und sobald diese Brücke steht, 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](https://www.mikebild.dev/de/blog/rxjs-push-und-pull/). Hier setze ich voraus, dass Observable und Subscription vertraut sind, und konzentriere mich auf die Entscheidung dahinter und auf die Subjects selbst. Stand ist RxJS 7. ## Lohnt sich Reaktivität hier überhaupt? Bevor ich zwischen den vier Typen wähle, stelle ich eine Frage davor: Braucht diese Stelle überhaupt einen Stream? Reaktivität ist kein Selbstzweck, sondern ein Modell mit Kosten. Ein Subject bringt einen Lebenszyklus mit, um den man sich kümmern muss – Subscriptions, die man wieder abbestellt, Puffer, die Speicher halten, ein Timing, das man im Kopf behalten muss. Diese Kosten lohnen sich, aber nicht überall. Für einen einzelnen Wert, der einmal berechnet und dann gelesen wird, reicht eine Variable. Für eine einzelne asynchrone Operation mit genau einem Ergebnis ist ein Promise das einfachere und vertrautere Werkzeug. Für eine fertige Liste, die vollständig vorliegt, tut es ein Array mit den üblichen Methoden. In all diesen Fällen einen Stream aufzuziehen, verteuert den Code, ohne etwas zurückzugeben – man handelt sich den Subscription-Lebenszyklus ein und bekommt dafür nichts, was das schlichtere Modell nicht auch geliefert hätte. Reaktivität zahlt sich in dem Moment aus, in dem drei Dinge zusammenkommen: Werte entstehen über die Zeit verteilt, nicht auf einen Schlag; es gibt mehr als einen Interessenten für dieselben Werte; und man will diese Werte zusammensetzen, filtern, abbrechen oder zusammenführen, statt sie nur einmal entgegenzunehmen. Ein Klickstrom, ein WebSocket, ein Zustand, den mehrere Komponenten teilen – das sind die Fälle, in denen ein Geflecht aus Variablen und Callbacks schnell unübersichtlich wird und ein Stream die Reibung herausnimmt. Die Vorabfrage kostet nichts und hält die Menge an reaktivem Code klein. Dort, wo Reaktivität trägt, spielt sie ihre Stärke aus, und der Rest bleibt einfach. Erst wenn die Frage mit Ja beantwortet ist, wird die zweite Frage relevant, um die es hier geht: welches Gedächtnis der Stream für späte Subscriber haben soll. ## Ein Subject ist beides zugleich Der entscheidende Satz über Subjects steht schon in der Klassendefinition: `Subject` erweitert `Observable` 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. ```ts import { Subject } from 'rxjs'; const subject = new Subject(); // 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. ```mermaid flowchart LR P["Quelle
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` schlichtweg 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. Die Liste beschreibt nur `next` und `complete`. Ein `error` beendet jeden dieser Streams. Danach liefert ein `BehaviorSubject` keinen letzten Wert mehr, ein `AsyncSubject` kein Endergebnis; ein `ReplaySubject` spielt seinen Puffer noch aus und meldet anschließend den Fehler. Diese terminalen Signale gehören deshalb genauso zur Wahl wie das Verhalten bei späten Subscriptions. 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. ```ts import { Subject } from 'rxjs'; const subject = new Subject(); 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. ```ts import { BehaviorSubject } from 'rxjs'; const state$ = new BehaviorSubject(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. ```ts const counter$ = new BehaviorSubject(0); counter$.next(42); console.log(counter$.getValue()); // 42 console.log(counter$.value); // 42 ``` Wer schon einmal in [RxJS in React](https://www.mikebild.dev/de/blog/rxjs-in-react-datenfluss/) 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. ```ts import { ReplaySubject } from 'rxjs'; const replay$ = new ReplaySubject(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(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. ```ts const cache$ = new ReplaySubject(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. ```ts import { AsyncSubject } from 'rxjs'; const async$ = new AsyncSubject(); 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. Der Fehler sieht harmlos aus und kostet trotzdem eine halbe Stunde: Die Werte sind geflossen, die Konsole bleibt leer, und man sucht zunächst überall – nur nicht beim fehlenden Abschluss. ## 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()`: ```ts import { Subject, Observable } from 'rxjs'; interface Command { type: string; } class CommandBus { private commandsSubject = new Subject(); // nach außen nur lesbar readonly commands$: Observable = 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](https://www.mikebild.dev/de/blog/rxjs-in-node-ereignisstroeme/) 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 die Frage davorstellt – lohnt sich der Stream hier überhaupt? – hält die Menge an reaktivem Code auf das begrenzt, was wirklich davon profitiert. Was dabei herauskommt, ist unspektakulär und genau deshalb brauchbar: Der passende Typ steht schneller fest, eine ganze Klasse von Timing-Bugs entsteht gar nicht erst, und der Datenfluss bleibt dort kontrolliert, wo er entsteht – hinter Observables nach außen, mit einem Subject als Motor im Inneren. Und wo eine Variable, ein Promise oder ein Array reicht, bleibt es dabei. Reaktivität ist das Werkzeug für Werte über die Zeit und viele Empfänger, nicht der Default für jeden Wert. ## Weiterführende Quellen - [RxJS Guide: Subject](https://rxjs.dev/guide/subject) - [RxJS API: BehaviorSubject](https://rxjs.dev/api/index/class/BehaviorSubject) - [RxJS API: ReplaySubject](https://rxjs.dev/api/index/class/ReplaySubject) - [RxJS API: AsyncSubject](https://rxjs.dev/api/index/class/AsyncSubject) - [RxJS: Push und Pull](https://www.mikebild.dev/de/blog/rxjs-push-und-pull/) - [RxJS in Node.js](https://www.mikebild.dev/de/blog/rxjs-in-node-ereignisstroeme/) - [RxJS in React](https://www.mikebild.dev/de/blog/rxjs-in-react-datenfluss/)