Reaktive Programmierung in JavaScript (Node.js): die Bibliothek RxJS

Reaktive Programmierung in JavaScript (Node.js) mit RxJS

\n\n

Reaktive Programmierung ist ein Paradigma, das sich auf asynchrone Datenströme und die Verbreitung von Änderungen konzentriert. Im Ökosystem von JavaScript (Node.js) ist eine der leistungsstärksten Bibliotheken zur Umsetzung dieses Paradigmas RxJS (Reactive Extensions for JavaScript). Sie bietet Werkzeuge für die Arbeit mit Observable (beobachtbaren Sequenzen) und ermöglicht es Entwicklern, Ereignisse, HTTP-Anfragen, Timer und andere asynchrone Datenquellen effizient zu verarbeiten.

\n\n

Grundlegende Konzepte von RxJS

\n

Im Kern von RxJS stehen drei Schlüsselkomponenten:

\n

\n
• Observable (Beobachtbares Objekt) — stellt einen Datenstrom dar, der im Laufe der Zeit Werte erzeugen kann.
\n
• Observer (Beobachter) — ein Objekt, das ein Observable abonniert und auf neue Daten (next), Fehler (error) oder den Abschluss des Stroms (complete) reagiert.
\n
• Subscription (Abonnement) — das Ergebnis des Aufrufs der Methode subscribe(). Ermöglicht die Verwaltung des Lebenszyklus des Abonnements (z. B. dessen Kündigung).
\n

\n

Eine wichtige Rolle spielen auch die Operatoren — reine Funktionen, die es ermöglichen, Datenströme zu transformieren, zu filtern, zu kombinieren und zu verknüpfen.

\n\n

Installation und Einrichtung

\n

Um RxJS in einem Node.js-Projekt zu verwenden, reicht es aus, das Paket über npm zu installieren:

\n

npm install rxjs

\n

Nach der Installation ist die Bibliothek einsatzbereit. Importieren Sie die erforderlichen Komponenten:

\n

const { Observable, of, from, interval } = require('rxjs');\nconst { map, filter, take } = require('rxjs/operators');

\n\n

Erstellen eines Observable

\n

Es gibt mehrere Möglichkeiten, Datenströme zu erstellen. Betrachten wir die grundlegenden:

\n\n

1. Erstellung aus einem Array

\n

const { from } = require('rxjs');\n\nconst numbers$ = from([1, 2, 3, 4, 5]);\nnumbers$.subscribe({\n    next: value => console.log(`Empfangen: ${value}`),\n    complete: () => console.log('Strom abgeschlossen')\n});

\n

Ergebnis der Ausführung:

\n

Empfangen: 1\nEmpfangen: 2\nEmpfangen: 3\nEmpfangen: 4\nEmpfangen: 5\nStrom abgeschlossen

\n\n

2. Periodische Generierung von Werten

\n

const { interval } = require('rxjs');\nconst { take } = require('rxjs/operators');\n\nconst timer$ = interval(1000).pipe(\n    take(4) // auf 4 Werte begrenzen\n);\n\ntimer$.subscribe({\n    next: value => console.log(`Tick ${value}`),\n    complete: () => console.log('Timer abgeschlossen')\n});

\n

Ausgabe (im Abstand von 1 Sekunde):

\n

Tick 0\nTick 1\nTick 2\nTick 3\nTimer abgeschlossen

\n\n

Arbeit mit Operatoren

\n

Operatoren sind das Herz von RxJS. Sie ermöglichen die Transformation von Daten im laufenden Betrieb. Betrachten wir zwei beliebte Operatoren:

\n

\n
• map — transformiert jeden Wert des Stroms.
\n
• filter — lässt nur die Werte durch, die eine Bedingung erfüllen.
\n

\n\n

const { of } = require('rxjs');\nconst { map, filter } = require('rxjs/operators');\n\nconst source$ = of(1, 2, 3, 4, 5, 6);\n\nsource$.pipe(\n    filter(value => value % 2 === 0), // nur gerade Zahlen behalten\n    map(value => value * 10)          // jede Zahl mit 10 multiplizieren\n).subscribe({\n    next: value => console.log(`Ergebnis: ${value}`),\n    complete: () => console.log('Fertig')\n});

\n

Ergebnis:

\n

Ergebnis: 20\nErgebnis: 40\nErgebnis: 60\nFertig

\n\n

Fehlerbehandlung in reaktiven Strömen

\n

RxJS bietet integrierte Mechanismen zur Fehlerbehandlung. Zum Beispiel ermöglicht der Operator catchError das Abfangen einer Ausnahme und die Fortsetzung der Ausführung:

\n

const { throwError, of } = require('rxjs');\nconst { catchError } = require('rxjs/operators');\n\nconst error$ = throwError(() =>