Programación reactiva en C# (.NET)

Programación reactiva en C# (.NET): Observable, Rx.NET y trabajo con flujos de datos

La programación reactiva (Reactive Programming) es un paradigma orientado al procesamiento asíncrono de flujos de datos y la propagación de cambios. En el ecosistema C# (.NET), este concepto se implementa a través de la biblioteca Rx.NET (Reactive Extensions) y las interfaces IObservable<T> / IObserver<T>. A diferencia del enfoque imperativo clásico, la programación reactiva permite describir de forma declarativa cómo debe reaccionar el sistema ante eventos, sin bloquear el hilo de ejecución principal.

En la página PythonLib exploraremos los componentes clave de la programación reactiva en C#, incluyendo secuencias Observable, operadores LINQ para la transformación de flujos y ejemplos prácticos de uso.

Conceptos fundamentales de la programación reactiva en C#

La programación reactiva se basa en el patrón Observer, donde:

Observable — fuente de datos (flujo de eventos).
Observer — suscriptor que reacciona a los datos entrantes.
Subscription — mecanismo de gestión de la suscripción (cancelación de la suscripción al flujo).

En .NET, estos conceptos están representados por las interfaces IObservable<T> e IObserver<T>, pero en la práctica se utiliza más a menudo la biblioteca Rx.NET (paquete NuGet System.Reactive), que proporciona un rico conjunto de operadores para trabajar con secuencias.

Instalación de Rx.NET

Para empezar, agregue el paquete NuGet al proyecto:

Install-Package System.Reactive

O mediante .NET CLI:

dotnet add package System.Reactive

Creación de un Observable y suscripción a un flujo de datos

Veamos un ejemplo sencillo: crearemos un Observable que genera números del 1 al 5 con un retardo, y nos suscribiremos a él.

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

class Program{    static async Task Main()    {        // Creamos una secuencia Observable        var observable = Observable.Interval(TimeSpan.FromSeconds(1))            .Select(x => x + 1) // Empezamos desde 1            .Take(5);           // Tomamos solo 5 elementos

        // Nos suscribimos al flujo        var subscription = observable.Subscribe(            onNext: value => Console.WriteLine($"Recibido: {value}"),            onError: error => Console.WriteLine($"Error: {error.Message}"),            onCompleted: () => Console.WriteLine("Flujo completado")        );

        // Esperamos a que termine (en código real use await o bloqueo)        await Task.Delay(6000);

        // Cancelamos la suscripción (liberamos recursos)        subscription.Dispose();    }}

Operadores LINQ para la transformación de flujos

Una de las principales ventajas de Rx.NET es la posibilidad de aplicar a los flujos de datos operadores familiares de LINQ: Where, Select, Aggregate y otros. Esto permite describir transformaciones de forma declarativa.

Ejemplo: filtrado y transformación

using System;using System.Reactive.Linq;

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

var processed = numbers    .Where(n => n % 2 == 0)    // Solo pares    .Select(n => n * n)        // Elevamos al cuadrado    .Do(n => Console.WriteLine($"Procesado: {n}")); // Efecto secundario

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