Реактивное программирование в C# (.NET) — Observable, Rx.NET и LINQ

Реактивное программирование в C# (.NET): Observable, Rx.NET и работа с потоками данных

Реактивное программирование (Reactive Programming) — это парадигма, ориентированная на асинхронную обработку потоков данных и распространение изменений. В экосистеме C# (.NET) эта концепция реализована через библиотеку Rx.NET (Reactive Extensions) и интерфейсы IObservable<T> / IObserver<T>. В отличие от классического императивного подхода, реактивное программирование позволяет декларативно описывать, как система должна реагировать на события, не блокируя основной поток выполнения.

На странице PythonLib мы рассмотрим ключевые компоненты реактивного программирования в C#, включая Observable последовательности, операторы LINQ для трансформации потоков и практические примеры использования.

Основные концепции реактивного программирования в C#

Реактивное программирование базируется на паттерне Observer, где:

• Observable — источник данных (поток событий).
• Observer — подписчик, который реагирует на поступающие данные.
• Subscription — механизм управления подпиской (отписка от потока).

В .NET эти концепции представлены интерфейсами IObservable<T> и IObserver<T>, но на практике чаще используется библиотека Rx.NET (NuGet пакет System.Reactive), которая предоставляет богатый набор операторов для работы с последовательностями.

Установка Rx.NET

Для начала работы добавьте в проект NuGet пакет:

Install-Package System.Reactive

Или через .NET CLI:

dotnet add package System.Reactive

Создание Observable и подписка на поток данных

Рассмотрим простейший пример: создадим Observable, который генерирует числа от 1 до 5 с задержкой, и подпишемся на него.

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

class Program
{
    static async Task Main()
    {
        // Создаем Observable последовательность
        var observable = Observable.Interval(TimeSpan.FromSeconds(1))
            .Select(x => x + 1) // Начинаем с 1
            .Take(5);           // Берем только 5 элементов

        // Подписываемся на поток
        var subscription = observable.Subscribe(
            onNext: value => Console.WriteLine($"Получено: {value}"),
            onError: error => Console.WriteLine($"Ошибка: {error.Message}"),
            onCompleted: () => Console.WriteLine("Поток завершен")
        );

        // Ожидаем завершения (в реальном коде используйте await или блокировку)
        await Task.Delay(6000);

        // Отписываемся (освобождаем ресурсы)
        subscription.Dispose();
    }
}

Операторы LINQ для трансформации потоков

Одно из главных преимуществ Rx.NET — возможность применять к потокам данных операторы, знакомые по LINQ: Where, Select, Aggregate и другие. Это позволяет декларативно описывать преобразования.

Пример: фильтрация и трансформация

using System;
using System.Reactive.Linq;

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

var processed = numbers
    .Where(n => n % 2 == 0)    // Только четные
    .Select(n => n * n)        // Возводим в квадрат
    .Do(n => Console.WriteLine($"Обработано: {n}")); // Побочный эффект

processed.Subscribe(
    onNext: n => Console.WriteLine($"Результат: {n}"),
    onCompleted: () => Console.WriteLine("Готово")
);

Объединение нескольких потоков

Rx.NET поддерживает слияние, объединение и комбинирование потоков. Например, Merge объединяет несколько Observable в один:

using System;
using System.Reactive.Linq;

var first = Observable.Interval(TimeSpan.FromSeconds(1)).Select(x => $"A{x}");
var second = Observable.Interval(TimeSpan.FromSeconds(0.5)).Select(x => $"B{x}");

var merged = first.Merge(second).Take(10);

merged.Subscribe(Console.WriteLine);

Console.ReadLine(); // Ждем завершения

Асинхронные операции и обработка ошибок

Реактивное программирование идеально подходит для работы с асинхронными источниками данных, такими как HTTP-запросы или файловые потоки. Обработка ошибок в Rx.NET осуществляется через операторы Catch, Retry и Finally.

Пример с обработкой ошибок:

using System;
using System.Reactive.Linq;

var faultyObservable = Observable.Create<int>(observer =>
{
    observer.OnNext(1);
    observer.OnNext(2);
    observer.OnError