Реактивное программирование в C# (.NET): Rx.NET и IObservable

Реактивное программирование в C# (.NET): Полное руководство с Rx.NET

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

В этой статье мы разберем ключевые концепции реактивного программирования, рассмотрим основные компоненты Rx.NET и приведем практические примеры кода для образовательного сайта PythonLib.

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

В основе Rx.NET лежат два интерфейса:

• IObservable<T> — источник данных (observable sequence), который может отправлять элементы, ошибки или сигнал завершения.
• IObserver<T> — подписчик (observer), который получает уведомления от источника: OnNext, OnError, OnCompleted.

Ключевая идея: вместо того чтобы опрашивать источник (pull-модель), подписчик реагирует на события (push-модель). Это особенно полезно для обработки UI-событий, сетевых запросов, потоков данных с датчиков или любых других асинхронных операций.

Начало работы с Rx.NET

Для использования Rx.NET в проекте C# установите NuGet-пакет:

dotnet add package System.Reactive

Пример 1: Создание и подписка на observable

Создадим простую последовательность чисел, которая генерирует значения с задержкой:

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

class Program
{
    static async Task Main()
    {
        var observable = Observable
            .Interval(TimeSpan.FromSeconds(1)) // генерирует 0, 1, 2... каждую секунду
            .Take(5); // берем только первые 5 чисел

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

        // Ждем завершения потока
        await Task.Delay(6000);
        subscription.Dispose(); // отписываемся
    }
}

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

Rx.NET поддерживает все стандартные операторы LINQ, что делает код лаконичным:

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

var processed = values
    .Where(x => x % 2 == 0) // только четные
    .Select(x => x * 10)    // умножаем на 10
    .Timestamp();           // добавляем временную метку

processed.Subscribe(x => Console.WriteLine($"[{x.Timestamp}] Значение: {x.Value}"));

Продвинутые возможности Rx.NET

Объединение потоков

Вы можете комбинировать несколько observable-последовательностей:

var stream1 = Observable.Interval(TimeSpan.FromSeconds(1)).Select(x => $"A{x}");
var stream2 = Observable.Interval(TimeSpan.FromSeconds(1.5)).Select(x => $"B{x}");

var merged = stream1.Merge(stream2); // объединяем в один поток
merged.Take(6).Subscribe(Console.WriteLine);

Управление многопоточностью

Rx.NET позволяет легко переключать контекст выполнения:

var uiScheduler = new SynchronizationContextScheduler(SynchronizationContext.Current);

Observable
    .Interval(TimeSpan.FromSeconds(1))
    .ObserveOn(uiScheduler) // гарантируем выполнение в UI-потоке
    .Subscribe(x => UpdateUI(x)); // обновление интерфейса

Обработка ошибок

Rx.NET предоставляет мощные механизмы восстановления после сбоев:

var faultyStream = Observable.Throw<int>(new Exception("Ошибка!"));

var safeStream = faultyStream
    .Catch((Exception ex) => Observable.Return(-1)) // заменяем ошибку значением
    .Retry(3); // повторяем до 3 раз

safeStream.Subscribe(
    x => Console.WriteLine(x),
    ex => Console.WriteLine($"Финальная ошибка: {ex.Message}")
);

Практическое применение: Поиск с автодополнением

Реактивное программирование идеально подходит для реализации debounce-механизма в поисковых строках:

public class SearchViewModel
{
    private readonly Subject<string> _searchSubject = new Subject<string>();

    public IObservable<string> SearchResults { get; }

    public SearchViewModel()
    {
        SearchResults = _searchSubject
            .Throttle(TimeSpan.FromMilliseconds(300)) // ждем 300 мс после последнего ввода