Observables von null an

Ein Observable ist ein Strom von Werten über die Zeit. Es kann nichts, einen einzigen Wert oder eine endlose Folge ausgeben und endet mit complete oder error — oder nie. In Angular begegnet es dir überall, wo Events passieren: HttpClient, Router-Events, valueChanges in Reactive Forms, WebSockets.

Signals haben die Aufgabe übernommen, Zustand zu halten; Observables bleiben das Werkzeug für Events in der Zeit — Klicks, Tastendrücke, Antworten, die man abbrechen kann. Diese Seite deckt ab, was du vor Signals und RxJS und RxJS für Fortgeschrittene brauchst.

Anatomie eines Observables

Ein Observable ist eine Funktion, die darauf wartet, aufgerufen zu werden. new Observable(subscriber => …) speichert sie; jedes subscribe() ruft sie mit einem neuen Subscriber auf, der Werte über next() und das Ende über error() oder complete() bekommt. Die Funktion gibt ein Teardown zurück: was gestoppt werden muss, wenn sich jemand abmeldet.

import { Observable } from 'rxjs';

// die Funktion läuft bei JEDEM subscribe(), nicht hier
const ticker$ = new Observable<number>((subscriber) => {
  let n = 0;
  const id = setInterval(() => subscriber.next(++n), 1000);

  // Teardown: läuft bei unsubscribe(), complete() oder error()
  return () => clearInterval(id);
});

Lazy, und eine Ausführung pro Subscriber

Wenn du ein Observable erzeugst, passiert nichts — nicht einmal ein HTTP-Request. Die Arbeit beginnt bei subscribe(), und jeder Subscriber bekommt seine eigene Ausführung: zwei Subscriber auf http.get() bedeuten zwei Requests. So ein Observable nennt man cold.

Ohne Teilen startet jeder Subscriber sein eigenes setInterval: Die Werte unterscheiden sich, und die Zahl der laufenden Ausführungen wächst. share() macht den Strom hot: eine Ausführung, deren Werte an alle gehen, gestoppt, wenn der letzte Subscriber geht. shareReplay({ bufferSize: 1, refCount: true }) gibt zusätzlich jedem, der später kommt, den letzten Wert — ein neuer Subscriber zeigt kein „…“ mehr.

const user$ = this.http.get<User>('/api/user'); // noch wird nichts gesendet

user$.subscribe(); // Request 1
user$.subscribe(); // Request 2 — jeder Subscriber bekommt seine eigene Ausführung

// ein Request, geteilt von allen, die abonnieren, solange er läuft
const shared$ = user$.pipe(shareReplay({ bufferSize: 1, refCount: true }));

Operatoren: pipe() baut einen neuen Strom

Ein Operator nimmt ein Observable und gibt ein anderes zurück; die Quelle ändert sich nicht. pipe() verkettet sie von oben nach unten, und jeder Wert läuft durch die ganze Kette, bevor der nächste startet.

source$.pipe(
  filter((n) => n % 2 === 0),
  map((n) => n * 10),
).subscribe({ next, error, complete });

Ein paar Dinge zum Beobachten. filter und take entscheiden, ob ein Wert durchkommt; map und scan verändern ihn; debounceTime hält einen Wert zurück, bis die Quelle 500 ms still war — aus der Salve wird ein einziger Wert. take(3) beendet den Subscriber, obwohl die Quelle weiterläuft. Und nach complete() oder error() ist die Quelle endgültig fertig: Es kommt nichts mehr durch.

import { debounceTime, distinctUntilChanged, filter, map } from 'rxjs';

const query$ = this.searchControl.valueChanges.pipe(
  map((text) => text.trim()),
  filter((text) => text.length >= 2), // nicht für einen Buchstaben
  debounceTime(300), // warten, bis das Tippen pausiert
  distinctUntilChanged(), // dieselbe Suche nochmal: überspringen
);

Die Operatoren, die du jede Woche brauchst

Wer meldet sich ab?

Ein Observable, das endet — ein HTTP-Request —, räumt selbst auf. Eines, das nie endet — interval, ein WebSocket, Router-Events —, läuft weiter, bis sich jemand abmeldet. In Angular rufst du unsubscribe() selten selbst auf: Die async-Pipe und toSignal() tun es, wenn die Komponente zerstört wird, und takeUntilDestroyed() tut es für ein subscribe() in der Klasse. Was passiert, wenn es niemand tut, steht unter Memory Leaks.

// 1. im Template: die async-Pipe meldet sich ab, wenn die Komponente zerstört wird
// <p>{{ ticker$ | async }}</p>

// 2. als Signal: genauso, und es liest sich wie jedes andere Signal
readonly tick = toSignal(ticker$, { initialValue: 0 });

// 3. ein subscribe() in der Klasse: an die Lebensdauer der Komponente binden
constructor() {
  ticker$.pipe(takeUntilDestroyed()).subscribe((n) => this.log(n));
}

Subject und BehaviorSubject

Ein Subject ist ein Observable, in das du selbst Werte mit next() schiebst — praktisch, um Callbacks in einen Strom zu verwandeln. Ein BehaviorSubject merkt sich zusätzlich den letzten Wert. Für Zustand in einer Komponente oder einem Service erledigt ein Signal die Aufgabe des BehaviorSubjects mit weniger Code; behalte Subjects für Events.

// ein Subject: du schiebst die Werte hinein
const clicks$ = new Subject<MouseEvent>();
button.addEventListener('click', (e) => clicks$.next(e));

// ein BehaviorSubject merkt sich den letzten Wert…
const user$ = new BehaviorSubject<User | null>(null);
user$.getValue();

// …aber für Zustand ist ein Signal einfacher
readonly user = signal<User | null>(null);

Signal oder Observable?

Die übliche Form in Angular 22: Observables an den Rändern (HTTP, Sockets, DOM-Events), mit toSignal() in Signals verwandelt, sobald sie eine Komponente erreichen. Mehr dazu in Signals und RxJS.

@Component({
  template: `
    @if (user(); as u) {
      <h2>{{ u.name }}</h2>
      <p>{{ u.email }}</p>
    }
  `,
})
export class Profile {
  private readonly http = inject(HttpClient);
  // das Observable bleibt am Rand; die Komponente liest ein Signal
  readonly user = toSignal(this.http.get<User>('/api/user'));
}

Häufige Stolperfallen

• Das Abonnieren vergessen. this.http.post(…) ohne subscribe() sendet nichts.
• Zwei async-Pipes auf demselben HTTP-Observable. Zwei Requests. Wandle es einmal mit toSignal() um oder lies es einmal mit @let.
• subscribe() in subscribe(). Verschachtelte Abos lassen sich nicht gemeinsam abbrechen und laufen durcheinander. Nimm switchMap und seine Geschwister — siehe RxJS für Fortgeschrittene.
• Ein endloser Strom ohne Ausgang. interval läuft weiter, wenn die Komponente schon weg ist. Nimm takeUntilDestroyed() oder toSignal().
• shareReplay(1) ohne refCount. Die Quelle läuft weiter, auch wenn alle Subscriber gegangen sind.

Spickzettel

Observable             eine Funktion, die bei jedem subscribe() läuft; gibt 0..∞ Werte aus, dann complete oder error
lazy                   Erzeugen tut nichts; subscribe() startet die Arbeit (sogar einen HTTP-Request)
cold                   eine Ausführung pro Subscriber: zwei Subscriber = zwei Requests
share()                eine Ausführung für alle; stoppt, wenn der letzte Subscriber geht
shareReplay(…)         wiederholt auch den letzten Wert; immer { bufferSize: 1, refCount: true }
pipe(a, b, c)          ein neuer Strom; Werte laufen durch a, dann b, dann c
Teardown               die Funktion, die new Observable(…) zurückgibt; läuft bei unsubscribe
Abmelden               async-Pipe, toSignal(), takeUntilDestroyed(); HTTP endet von selbst
Subject                du rufst next() selbst auf; BehaviorSubject behält auch den letzten Wert
Signal vs. Observable  ein Wert jetzt vs. Events über die Zeit; toSignal() an der Komponentengrenze