RxJS in React: Observables im Datenfluss der Oberfläche

Wenn ich RxJS in einer React-Schulung auf den Tisch lege, geht regelmäßig ein Ruck durch den Raum. Die eine Hälfte hat mit RxJS im Angular-Umfeld gelitten und will es nie wieder sehen. Die andere Hälfte hat davon gehört, findet Streams elegant und würde am liebsten den gesamten Komponentenzustand in Observables gießen. Beide Lager liegen daneben – und genau an dieser Spannung lässt sich am besten erklären, wofür push-basierte Streams in einer React-Oberfläche taugen und wofür nicht.

Meine These vorweg, damit der rote Faden klar ist: Die allermeiste React-UI braucht kein RxJS. useState, useEffect und useReducer decken das ab, was die Oberfläche an Zustand und Seiteneffekten mitbringt. RxJS wird erst dann zum richtigen Werkzeug, wenn asynchrone Ereignisse über die Zeit koordiniert werden müssen: Eingaben entprellen, veraltete Requests verwerfen, kontinuierliche Ströme aus WebSockets oder Timern verarbeiten. Genau diese Fälle schaue ich mir hier an.

Kurz eingeordnet: Was ein Observable in diesem Kontext ist

Ich setze die Grundlagen aus RxJS: Push und Pull hier voraus – der Beitrag erklärt, warum ein Observable ein Push-Producer ist und wie er sich von Promises und Iteratoren unterscheidet. Für den React-Kontext sind drei Eigenschaften entscheidend, die in Schulungen immer wieder für Aha-Momente sorgen.

Ein Observable ist erstens lazy: Der Code im subscribe-Callback läuft erst, wenn tatsächlich jemand subscribe() aufruft. Ein Observable, das niemand abonniert, tut nichts. Zweitens ist ein normales Observable per Default cold und unicast – jede Subscription startet ihre eigene Ausführung. Drittens gibt subscribe() eine Subscription zurück, und über deren unsubscribe() wird die Ausführung wieder beendet.

Diese drei Punkte zusammengenommen erklären, warum RxJS und der useEffect-Lebenszyklus so gut zusammenpassen: Ein Effekt hat genau die Struktur „aufsetzen beim Mount, aufräumen beim Unmount", die ein Observable braucht.

import { Observable } from 'rxjs';

// A cold observable: the callback runs once per subscriber
const ticker$ = new Observable<number>((subscriber) => {
  let count = 0;
  const id = setInterval(() => subscriber.next(count++), 1000);

  // teardown logic runs on unsubscribe
  return () => clearInterval(id);
});

const subscription = ticker$.subscribe((value) => console.log(value));
// later, to stop and clean up:
subscription.unsubscribe();

Wer den Aufbau dieses subscribe/unsubscribe-Musters verinnerlicht hat, erkennt sofort die Parallele zur Cleanup-Funktion von useEffect. Das ist kein Zufall, sondern der eigentliche Grund, warum die Integration so geradlinig ist.

Der erste sinnvolle Fall: Eingaben entprellen

Das klassische Beispiel ist ein Suchfeld. Bei jedem Tastenanschlag einen HTTP-Request abzufeuern ist verschwenderisch – der Nutzer tippt „React", und schon sind sechs Requests unterwegs, von denen fünf sofort veraltet sind. Die naive React-Lösung baut hier einen setTimeout mit clearTimeout im Effekt zusammen. Das funktioniert, wird aber schnell unübersichtlich, sobald mehr als nur Entprellung dazukommt.

Mit RxJS wird die Entprellung zu einem einzigen deklarativen Operator. debounceTime(300) emittiert einen Wert erst, wenn 300 Millisekunden lang kein neuer Wert nachgekommen ist. Ich speise die Eingaben über ein Subject ein – ein Subject ist zugleich Observable und Observer, ich kann also im onChange-Handler next() aufrufen und am anderen Ende die Pipeline abonnieren.

import { useEffect, useRef, useState } from 'react';
import { Subject } from 'rxjs';
import { debounceTime, distinctUntilChanged } from 'rxjs/operators';

function SearchField() {
  const query$ = useRef(new Subject<string>()).current;
  const [debounced, setDebounced] = useState('');

  useEffect(() => {
    const subscription = query$
      .pipe(debounceTime(300), distinctUntilChanged())
      .subscribe(setDebounced);

    return () => subscription.unsubscribe();
  }, [query$]);

  return (
    <input
      type="search"
      onChange={(event) => query$.next(event.target.value)}
      placeholder="Enter search term"
    />
  );
}

Zwei Details, die in Schulungen gern übersehen werden. Das Subject lege ich mit useRef an, damit es über Re-Renders hinweg stabil bleibt – würde ich es im Render-Body neu erzeugen, hätte jeder Render sein eigenes Subject, und die Verbindung zwischen Handler und Pipeline bräche. Und distinctUntilChanged() filtert identische aufeinanderfolgende Werte heraus, damit ein Fokuswechsel ohne Textänderung keinen neuen Durchlauf auslöst.

Der Fall, in dem RxJS wirklich glänzt: Autocomplete mit switchMap

Entprellen alleine rechtfertigt RxJS noch nicht – das bekommt man auch mit einem Timeout hin. Der Punkt, an dem ich Teilnehmer regelmäßig überzeuge, ist das Verwerfen veralteter Requests. Genau hier liegt das eigentliche Race-Condition-Problem eines Autocomplete: Ich tippe „re", ein Request geht raus. Ich tippe „rea", ein zweiter Request geht raus. Wenn nun die Antwort auf „re" nach der Antwort auf „rea" eintrifft, überschreibt das veraltete Ergebnis das aktuelle. Die Oberfläche zeigt Treffer zu einem Suchbegriff an, der gar nicht mehr im Feld steht.

switchMap löst das per Konstruktion. Der Operator abonniert das innere Observable – hier den HTTP-Request – und cancelt die vorherige innere Subscription, sobald ein neuer Wert von außen kommt. Kombiniert mit fromFetch und dessen AbortController wird der veraltete Request nicht nur ignoriert, sondern real abgebrochen.

import { useEffect, useRef, useState } from 'react';
import { Subject } from 'rxjs';
import { fromFetch } from 'rxjs/fetch';
import {
  debounceTime,
  distinctUntilChanged,
  filter,
  switchMap,
} from 'rxjs/operators';

type Hit = { id: string; label: string };

function Autocomplete() {
  const query$ = useRef(new Subject<string>()).current;
  const [results, setResults] = useState<Hit[]>([]);

  useEffect(() => {
    const subscription = query$
      .pipe(
        debounceTime(300),
        distinctUntilChanged(),
        filter((term) => term.length > 1),
        switchMap((term) =>
          fromFetch(`/api/search?q=${encodeURIComponent(term)}`).pipe(
            switchMap((response) => response.json() as Promise<Hit[]>),
          ),
        ),
      )
      .subscribe(setResults);

    return () => subscription.unsubscribe();
  }, [query$]);

  return (
    <div>
      <input onChange={(event) => query$.next(event.target.value)} />
      <ul>
        {results.map((hit) => (
          <li key={hit.id}>{hit.label}</li>
        ))}
      </ul>
    </div>
  );
}

Der entscheidende Unterschied hängt an einem einzigen Operatornamen. Wer hier aus Gewohnheit mergeMap schreibt, lässt alle Requests parallel weiterlaufen, und das oben beschriebene Race-Condition-Problem ist wieder da. Für „nur die letzte Eingabe zählt" ist switchMap die richtige Wahl. Ich lasse Teilnehmer das gerne einmal mit mergeMap bauen und mit einer künstlich verlangsamten API testen – die veralteten Treffer, die dann aufblitzen, prägen sich besser ein als jede Folie.

Das folgende Diagramm zeigt den Datenfluss der Pipeline von der Eingabe bis zum Re-Render.

flowchart LR
  A["Input-Event<br/>onChange"] --> B["Subject.next(term)"]
  B --> C["debounceTime(300)<br/>Ruhephase abwarten"]
  C --> D["distinctUntilChanged<br/>Duplikate filtern"]
  D --> E["switchMap<br/>alten Request verwerfen"]
  E --> F["subscribe<br/>setResults"]
  F --> G["React Re-Render"]

Wichtig ist die Cancellation-Semantik von switchMap, weil sie der Kern des Ganzen ist. Das zweite Diagramm stellt sie als Sequenz dar: Die Eingabe „rea" bricht den noch offenen inneren Stream von „re" ab, sodass nur die Antwort zur letzten Eingabe im State landet.

sequenceDiagram
  participant U as Nutzer
  participant P as Pipeline (switchMap)
  participant A as API
  U->>P: Eingabe "re"
  P->>A: Request "re"
  U->>P: Eingabe "rea"
  Note over P: switchMap cancelt<br/>inneren Stream "re"
  P--xA: Request "re" abgebrochen
  P->>A: Request "rea"
  A-->>P: Antwort "rea"
  P->>U: setResults (nur letzte Antwort)

Kontinuierliche Ströme: WebSockets und Events

Der dritte Fall, in dem sich RxJS auszahlt, sind Quellen, die nicht ein einzelnes Ergebnis liefern, sondern über die Zeit immer weiter emittieren. WebSockets sind das offensichtliche Beispiel. RxJS bringt mit webSocket aus rxjs/webSocket einen bidirektionalen Wrapper mit, der zugleich Observable und Observer ist – ich kann eintreffende Nachrichten abonnieren und über next() selbst welche senden. Das Detail, das hier Gold wert ist: unsubscribe() schließt die Verbindung sauber.

import { useEffect, useState } from 'react';
import { webSocket } from 'rxjs/webSocket';

type Message = { id: string; text: string };

function LiveFeed() {
  const [messages, setMessages] = useState<Message[]>([]);

  useEffect(() => {
    const socket$ = webSocket<Message>('wss://example.com/feed');

    const subscription = socket$.subscribe((message) =>
      setMessages((prev) => [...prev, message]),
    );

    // unsubscribe closes the underlying socket
    return () => subscription.unsubscribe();
  }, []);

  return (
    <ul>
      {messages.map((message) => (
        <li key={message.id}>{message.text}</li>
      ))}
    </ul>
  );
}

Wer tiefer in die Muster hinter Message-Strömen einsteigen will, findet in Web-Messaging mit RxJS den ausführlicheren Kontext dazu, wie sich bidirektionale Kanäle als Streams modellieren lassen.

Eine Variante zu fromEvent gehört noch dazu, weil sie eine Stolperstelle offenlegt. fromEvent hängt sich an einen echten DOM-Knoten, nicht an Reacts synthetisches Event-System. Man braucht also eine ref, und die ist erst nach dem Mount gesetzt:

import { useEffect, useRef, useState } from 'react';
import { fromEvent } from 'rxjs';
import { map, debounceTime } from 'rxjs/operators';

function ScrollTracker() {
  const boxRef = useRef<HTMLDivElement>(null);
  const [position, setPosition] = useState(0);

  useEffect(() => {
    const node = boxRef.current;
    if (!node) return;

    const subscription = fromEvent(node, 'scroll')
      .pipe(
        map((event) => (event.target as HTMLElement).scrollTop),
        debounceTime(50),
      )
      .subscribe(setPosition);

    return () => subscription.unsubscribe();
  }, []);

  return <div ref={boxRef}>{/* scrollable content */}</div>;
}

Der Fallstrick, der jeden einmal erwischt: fehlendes unsubscribe

Wenn in einer Schulung ein RxJS-React-Beispiel unerklärlich Speicher frisst oder eine Warnung „setState on an unmounted component" wirft, ist die Ursache fast immer dieselbe: Die Cleanup-Funktion fehlt oder gibt die Subscription nicht frei. Ohne unsubscribe() läuft der Stream nach dem Unmount weiter. Der Event-Listener bleibt hängen, der Timer tickt weiter, der WebSocket bleibt offen – und jeder emittierte Wert ruft ein setState auf einer Komponente auf, die es nicht mehr gibt.

// Wrong: no cleanup, the stream keeps running after unmount
useEffect(() => {
  interval(1000).subscribe((n) => setTick(n)); // memory leak
}, []);

// Right: keep the subscription and release it in cleanup
useEffect(() => {
  const subscription = interval(1000).subscribe((n) => setTick(n));
  return () => subscription.unsubscribe();
}, []);

Der zweite verwandte Fehler ist subtiler: die Pipeline im Render-Body statt im Effekt aufzubauen. Ein fromEvent(...).pipe(...), das bei jedem Render neu entsteht, produziert doppelte Listener und doppelte Subscriptions. Die Regel, die ich mitgebe: Die Quelle stabil halten (per useRef oder außerhalb der Komponente), die Subscription im Effekt aufsetzen, im Cleanup abbauen, und das Dependency-Array sauber pflegen.

Als Randnotiz für 2022: React 18 bringt mit useSyncExternalStore einen Mechanismus, um sich sauber an externe Stores – auch RxJS-basierte – zu binden. Für die hier gezeigten Fälle ist der useEffect-Weg aber völlig ausreichend, und ich würde useSyncExternalStore erst dann heranziehen, wenn ein Store wirklich außerhalb der Komponente liegt und von mehreren Stellen aus genutzt wird.

Wann RxJS sich lohnt – und wann nicht

Am Ende steht bei mir immer dieselbe Abgrenzung, weil sie die häufigste Fehlentscheidung verhindert: RxJS überall reinzuziehen. Die Entscheidung lässt sich auf zwei kurze Listen eindampfen.

RxJS lohnt sich, wenn:

Eingaben zeitbasiert entprellt oder gedrosselt werden müssen, bevor ein teurer Effekt läuft.
veraltete asynchrone Ergebnisse aktiv verworfen werden sollen (Autocomplete, konkurrierende Requests) – der Kernfall für switchMap.
kontinuierliche Ströme wie WebSockets, Server-Sent-Events oder Timer verarbeitet werden.
mehrere Event- und Zeitquellen kombiniert, gemischt oder zeitlich koordiniert werden müssen.

RxJS lohnt sich nicht, wenn:

es um einfachen lokalen Komponentenzustand geht – dafür sind useState und useReducer da.
ein einzelner Request beim Mount geladen wird; ein fetch mit AbortController im Effekt reicht, oder man greift zu einer Data-Fetching-Bibliothek.
das Team RxJS nicht beherrscht und die Anforderung keine echte asynchrone Koordination enthält – die Lernkurve zahlt sich dann nicht aus.

Diese Trennung ist der eigentliche Kern des Beitrags. RxJS ist kein Ersatz für Reacts Zustandsmodell, sondern eine Ergänzung für einen schmalen, aber wichtigen Ausschnitt: die Koordination asynchroner Ereignisse über die Zeit. Wer Observables gezielt an diesen wenigen Stellen einsetzt und überall sonst bei den React-Bordmitteln bleibt, bekommt das Beste aus beiden Welten – deklarative Streams dort, wo sie echte Probleme lösen, und schlanke Komponenten überall sonst. Und wer konsequent im Cleanup unsubscribe() aufruft, umgeht den einen Fehler, der sonst zuverlässig zurückkommt.

Weiterführende Quellen

RxJS: Push und Pull – die Grundlagen zu Observables als Push-Producer.
Web-Messaging mit RxJS – bidirektionale Message-Ströme als Streams modelliert.
Observable-Guide (rxjs.dev)
Operators-Guide / pipe (rxjs.dev)
Subject (rxjs.dev)
switchMap (rxjs.dev)
debounceTime (rxjs.dev)
fromEvent (rxjs.dev)
Subscription / unsubscribe (rxjs.dev)
React: Using the Effect Hook (reactjs.org)