Reaktive Programmierung in C# (.NET)

Reaktive Programmierung in C# (.NET): Observable, Rx.NET und Arbeit mit Datenströmen

Reaktive Programmierung (Reactive Programming) ist ein Paradigma, das auf die asynchrone Verarbeitung von Datenströmen und die Verbreitung von Änderungen ausgerichtet ist. Im Ökosystem C# (.NET) wird dieses Konzept durch die Bibliothek Rx.NET (Reactive Extensions) und die Schnittstellen IObservable<T> / IObserver<T> umgesetzt. Im Gegensatz zum klassischen imperativen Ansatz ermöglicht die reaktive Programmierung, deklarativ zu beschreiben, wie das System auf Ereignisse reagieren soll, ohne den Hauptausführungs-Thread zu blockieren.

Auf der Seite PythonLib betrachten wir die Schlüsselkomponenten der reaktiven Programmierung in C#, einschließlich Observable-Sequenzen, LINQ-Operatoren zur Transformation von Strömen und praktische Anwendungsbeispiele.

Grundlegende Konzepte der reaktiven Programmierung in C#

Die reaktive Programmierung basiert auf dem Observer-Entwurfsmuster, bei dem:

Observable — die Datenquelle (Ereignisstrom).
Observer — der Abonnent, der auf die eingehenden Daten reagiert.
Subscription — der Mechanismus zur Verwaltung des Abonnements (Abbestellung vom Strom).

In .NET werden diese Konzepte durch die Schnittstellen IObservable<T> und IObserver<T> repräsentiert, in der Praxis wird jedoch häufiger die Bibliothek Rx.NET (NuGet-Paket System.Reactive) verwendet, die eine umfangreiche Sammlung von Operatoren für die Arbeit mit Sequenzen bereitstellt.

Installation von Rx.NET

Fügen Sie dem Projekt das NuGet-Paket hinzu:

Install-Package System.Reactive

Oder über die .NET CLI:

dotnet add package System.Reactive

Erstellen eines Observable und Abonnieren eines Datenstroms

Betrachten wir ein einfaches Beispiel: Wir erstellen ein Observable, das mit einer Verzögerung Zahlen von 1 bis 5 generiert, und abonnieren es.

using System;using System.Reactive.Linq;using System.Threading.Tasks;

class Program{    static async Task Main()    {        // Erstellen einer Observable-Sequenz        var observable = Observable.Interval(TimeSpan.FromSeconds(1))            .Select(x => x + 1) // Beginne mit 1            .Take(5);           // Nimm nur 5 Elemente

        // Abonnieren des Streams        var subscription = observable.Subscribe(            onNext: value => Console.WriteLine($"Empfangen: {value}"),            onError: error => Console.WriteLine($"Fehler: {error.Message}"),            onCompleted: () => Console.WriteLine("Stream abgeschlossen")        );

        // Warten auf Abschluss (in echtem Code await oder Blockierung verwenden)        await Task.Delay(6000);

        // Abbestellen (Ressourcen freigeben)        subscription.Dispose();    }}

LINQ-Operatoren zur Transformation von Strömen

Einer der Hauptvorteile von Rx.NET ist die Möglichkeit, auf Datenströme Operatoren anzuwenden, die aus LINQ bekannt sind: Where, Select, Aggregate und andere. Dies ermöglicht eine deklarative Beschreibung von Transformationen.

Beispiel: Filterung und Transformation

using System;using System.Reactive.Linq;

var numbers = Observable.Range(1, 10);

var processed = numbers    .Where(n => n % 2 == 0)    // Nur gerade Zahlen    .Select(n => n * n)        // Quadrieren    .Do(n => Console.WriteLine($"Verarbeitet: {n}")); // Nebeneffekt

processed.Subscribe(    onNext: n => Console.WriteLine($"Ergebnis: {n}"),    onCompleted: () => Console.WriteLine("Fertig"));