Reaktive Programmierung in C# (.NET): Rx.NET und IObservable

Reaktive Programmierung in C# (.NET): Vollständiger Leitfaden mit Rx.NET

\n\n

Reaktive Programmierung ist ein Entwicklungsparadigma, das sich auf asynchrone Datenströme und die Verbreitung von Änderungen konzentriert. Im C# (.NET)-Ökosystem ist die Bibliothek Reactive Extensions (Rx.NET) das wichtigste Werkzeug zur Umsetzung dieses Ansatzes. Sie ermöglicht es, mit Datenfolgen, die sich im Laufe der Zeit ändern, genauso bequem zu arbeiten wie mit normalen Sammlungen über LINQ.

\n\n

In diesem Artikel werden wir die wichtigsten Konzepte der reaktiven Programmierung erläutern, die grundlegenden Komponenten von Rx.NET betrachten und praktische Codebeispiele für die Bildungswebsite PythonLib geben.

\n\n

Grundlegende Konzepte der reaktiven Programmierung

\n\n

Die Grundlage von Rx.NET bilden zwei Schnittstellen:

\n

\n
• IObservable<T> — die Datenquelle (observable sequence), die Elemente, Fehler oder ein Abschlusssignal senden kann.
\n
• IObserver<T> — der Abonnent (observer), der Benachrichtigungen von der Quelle empfängt: OnNext, OnError, OnCompleted.
\n

\n\n

Die Kernidee: Anstatt die Quelle abzufragen (Pull-Modell), reagiert der Abonnent auf Ereignisse (Push-Modell). Dies ist besonders nützlich für die Verarbeitung von UI-Ereignissen, Netzwerkanfragen, Datenströmen von Sensoren oder anderen asynchronen Operationen.

\n\n

Erste Schritte mit Rx.NET

\n\n

Um Rx.NET in einem C#-Projekt zu verwenden, installieren Sie das NuGet-Paket:

\n

dotnet add package System.Reactive

\n\n

Beispiel 1: Erstellen und Abonnieren eines Observable

\n

Erstellen wir eine einfache Zahlenfolge, die Werte mit einer Verzögerung generiert:

\n

using System;\nusing System.Reactive.Linq;\nusing System.Threading.Tasks;\n\nclass Program\n{\n    static async Task Main()\n    {\n        var observable = Observable\n            .Interval(TimeSpan.FromSeconds(1)) // generiert 0, 1, 2... jede Sekunde\n            .Take(5); // nimmt nur die ersten 5 Zahlen\n\n        var subscription = observable.Subscribe(\n            onNext: number => Console.WriteLine($"Empfangen: {number}"),\n            onError: error => Console.WriteLine($"Fehler: {error.Message}"),\n            onCompleted: () => Console.WriteLine("Stream abgeschlossen")\n        );\n\n        // Warten auf das Ende des Streams\n        await Task.Delay(6000);\n        subscription.Dispose(); // abmelden\n    }\n}

\n\n

Beispiel 2: Filtern und Transformieren mit LINQ

\n

Rx.NET unterstützt alle standardmäßigen LINQ-Operatoren, was den Code prägnant macht:

\n

var values = Observable.Range(1, 10);\n\nvar processed = values\n    .Where(x => x % 2 == 0) // nur gerade\n    .Select(x => x * 10)    // mit 10 multiplizieren\n    .Timestamp();           // Zeitstempel hinzufügen\n\nprocessed.Subscribe(x => Console.WriteLine($"[{x.Timestamp}] Wert: {x.Value}"));

\n\n

Fortgeschrittene Funktionen von Rx.NET

\n\n

Zusammenführen von Streams

\n

Sie können mehrere Observable-Sequenzen kombinieren:

\n

var stream1 = Observable.Interval(TimeSpan.FromSeconds(1)).Select(x => $"A{x}");\nvar stream2 = Observable.Interval(TimeSpan.FromSeconds(1.5)).Select(x => $"B{x}");\n\nvar merged = stream1.Merge(stream2); // zu einem Stream zusammenführen\nmerged.Take(6).Subscribe(Console.WriteLine);

\n\n

Multithreading-Verwaltung

\n

Rx.NET ermöglicht das einfache Umschalten des Ausführungskontexts:

\n

var uiScheduler = new SynchronizationContextScheduler(SynchronizationContext.Current);\n\nObservable\n    .Interval(TimeSpan.FromSeconds(1))\n    .ObserveOn(uiScheduler) // Ausführung im UI-Thread sicherstellen\n    .Subscribe(x => UpdateUI(x)); // Benutzeroberfläche aktualisieren

\n\n

Fehlerbehandlung

\n

Rx.NET bietet leistungsstarke Mechanismen zur Wiederherstellung nach Fehlern:

\n

var faultyStream = Observable.Throw<int>(new Exception("Fehler!"));\n\nvar safeStr