# RxJS in React: Observables im Datenfluss der Oberfläche

URL: https://www.mikebild.dev/de/blog/rxjs-in-react-datenfluss/

Ein Observable ist kein Zustandsspeicher. Es hält keinen Wert bereit, den eine Komponente auslesen könnte, und es ersetzt kein `useState`. Es beschreibt eine Ausführung, die Werte über die Zeit an ihre Abonnenten schiebt und die sich jederzeit abbrechen lässt. Wer diese Definition ernst nimmt, hat die Frage nach RxJS in React halb beantwortet: Wo es nur um den aktuellen Wert eines Feldes geht, trägt ein Observable nichts bei. Es zahlt sich erst aus, wo die Zeitachse selbst zum Problem wird.

**Redaktionelle Einordnung (Juli 2026):** Die gezeigten component-lokalen Subscriptions mit `useEffect` und Cleanup bleiben sinnvoll. Wird ein RxJS-basierter Zustand dagegen als externer Store von mehreren Komponenten gelesen, braucht React einen stabilen Snapshot-Vertrag, typischerweise über `useSyncExternalStore` oder eine darauf aufbauende Adapter-Bibliothek. Der Beitrag bleibt bewusst bei der Koordination von Ereignissen und macht RxJS nicht zur allgemeinen State-Lösung.

An dieser Unterscheidung hängt die ganze Diskussion, die in React-Teams meist verhakt läuft. Wer RxJS aus dem Angular-Umfeld kennt, verbindet damit oft Pipelines, die quer durch die Anwendung liefen und niemand mehr überblickte. Wer Streams nur von außen kennt, findet sie elegant und würde den kompletten Komponentenzustand in Observables gießen. Beide Positionen verfehlen den schmalen Bereich, um den es tatsächlich geht: Tastenanschläge, die schneller kommen, als eine API antwortet. Antworten, die in der falschen Reihenfolge eintreffen. Verbindungen, die einfach weitersenden. Diese drei Fälle nehme ich hier auseinander – und danach die Stellen, an denen die React-Bordmittel die bessere Wahl bleiben.

## Kurz eingeordnet: Was ein Observable in diesem Kontext ist

Ich setze die Grundlagen aus [RxJS: Push und Pull](https://www.mikebild.dev/de/blog/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.

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.

```ts
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.

```tsx
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 gehen in Code-Reviews regelmäßig unter. 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. Was RxJS trägt, 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.

```tsx
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. Der Unterschied wird greifbar, wenn man die Pipeline einmal bewusst mit `mergeMap` baut und gegen eine künstlich verlangsamte API laufen lässt: Die veralteten Treffer blitzen dann sichtbar in der Liste auf, bevor das aktuelle Ergebnis sie ersetzt.

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

```mermaid
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.

```mermaid
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.

```tsx
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>
  );
}
```

Die Grundlagen hinter solchen Message-Strömen beschreibe ich in [RxJS: Push und Pull](https://www.mikebild.dev/de/blog/rxjs-push-und-pull/). Bidirektionale Kanäle ändern an diesem Modell nichts: Senden und Empfangen bleiben zwei getrennte Ströme.

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:

```tsx
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 eine RxJS-React-Komponente 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.

```tsx
// 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 Reihenfolge, die zuverlässig funktioniert: 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

Die häufigste Fehlentscheidung ist, RxJS überall hineinzuziehen. Verhindern lässt sie sich mit zwei kurzen Listen.

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 beides: deklarative Streams dort, wo sie echte Probleme lösen, und schlanke Komponenten im Rest der Anwendung. Und wer konsequent im Cleanup `unsubscribe()` aufruft, umgeht den einen Fehler, der sonst zuverlässig zurückkommt.

## Weiterführende Quellen

- [RxJS: Push und Pull](https://www.mikebild.dev/de/blog/rxjs-push-und-pull/) – die Grundlagen zu Observables als Push-Producer.
- [Observable-Guide (rxjs.dev)](https://rxjs.dev/guide/observable)
- [Operators-Guide / pipe (rxjs.dev)](https://rxjs.dev/guide/operators)
- [Subject (rxjs.dev)](https://rxjs.dev/guide/subject)
- [switchMap (rxjs.dev)](https://rxjs.dev/api/operators/switchMap)
- [debounceTime (rxjs.dev)](https://rxjs.dev/api/operators/debounceTime)
- [fromEvent (rxjs.dev)](https://rxjs.dev/api/index/function/fromEvent)
- [Subscription / unsubscribe (rxjs.dev)](https://rxjs.dev/guide/subscription)
- [React: Using the Effect Hook (reactjs.org)](https://reactjs.org/docs/hooks-effect.html)