Rx.NET в реальном продакшене где реактивные потоки окупаются и где достаточно обычных событий

Реактивные расширения родились в Microsoft на рубеже 2000-х и 2010-х и разрослись в семейство: RxJS, RxJava и RxSwift повторяют идеи родоначальника, а сам он живёт под крылом .NET Foundation. Пакет System.Reactive дошёл до версии 7.0.0, вышел летом 2026 года и накопил около 440 миллионов загрузок на NuGet: сборки для .NET 8, netstandard2.0 и .NET Framework 4.7.2 и анализатор кода. Разработчик открывает для себя операторы Throttle, Buffer, Switch, и внезапно любое событие программы выглядит кандидатом на реактивную обёртку: кнопки, таймеры, показания датчиков. Опыт сопровождения отрезвляет быстро. Rx.NET экономит дни в одних сценариях и бесполезно тратит их в других, и границу зон стоит провести до первой подписки.

Механика Rx.NET и реальная цена входа в реактивный стиль

Ядро библиотеки умещается в два интерфейса. IObservable<T> уведомляет подписчиков о новых значениях, IObserver<T> принимает их, и оба живут в пространстве имён System со времён .NET Framework 4. Источник проталкивает значения наружу, подписчик реагирует, а System.Reactive добавляет поверх многие десятки LINQ-операторов: событийный поток запрашивается так же свободно, как обычная коллекция. Установка занимает команду:

dotnet add package System.Reactive --version 7.0.0

Минимальный пример:

using System.Reactive.Linq;

var ticks = Observable.Timer(
    dueTime: TimeSpan.Zero,
    period: TimeSpan.FromSeconds(1));

using var subscription = ticks.Subscribe(
    tick => Console.WriteLine($"Tick {tick}"));

Подписка возвращает IDisposable. Обычный обработчик события снимается оператором -=, реактивная подписка снимается методом Dispose, и пренебрежение этой мелочью оборачивается утечками. Готовое событие существующего класса оборачивается методом FromEventPattern, ниже это показано на примере FileSystemWatcher.

Цена входа не в синтаксисе, а в объёме понятий: планировщики, время жизни подписок, разница между холодными и горячими источниками. Команда без опыта разбирает цепочку из пяти операторов дольше, чем автор писал. Для зрелого коллектива это рабочий инструмент, для случайного набора людей обуза.

Поиск в интерфейсе с подавлением лишних запросов

Классика жанра: поле поиска с подсказками по мере ввода. Наивная реализация шлёт HTTP-запрос на каждое нажатие, быстрый набор в пике даёт восемь-десять нажатий в секунду, и нужен только последний запрос. Вдобавок ответы приезжают в произвольном порядке: ответ на первые буквы может прийти позже ответа на полную строку и затереть правильные результаты. Пользователь видит ложные подсказки и стучит по Backspace.

Реактивный конвейер гасит обе проблемы одним выражением:

var search = Observable
    .FromEventPattern<TextChangedEventArgs>(
        searchBox, nameof(searchBox.TextChanged))
    .Select(_ => searchBox.Text)
    .Throttle(TimeSpan.FromMilliseconds(250))
    .DistinctUntilChanged()
    .Select(term => Observable.FromAsync(ct => api.SearchAsync(term, ct)))
    .Switch()
    .ObserveOn(SynchronizationContext.Current)
    .Subscribe(results => resultsPanel.ItemsSource = results);

Throttle пропускает значение дальше только после 250 мс тишины, промежуточные буквы выбрасываются. DistinctUntilChanged убирает повторы, когда пользователь стёр символ и вернул его на место. Switch при каждом новом значении отписывается от предыдущего внутреннего потока, а FromAsync принимает CancellationToken, так что устаревший запрос прерывается, а не игнорируется. ObserveOn возвращает значения в контекст потока интерфейса.

Императивный эквивалент требует поля для CancellationTokenSource, таймера со сбросом на каждом нажатии, сравнения последнего запроса и синхронизации. Выходит полсотни строк ручного управления временем жизни против восьми операторов. Выигрыш Rx в том, что время здесь значение первого класса, и конвейер выражает работу с ним явно.

Прореживание потока показаний датчиков через Sample Buffer и скользящее среднее

Цеховой датчик температуры отдаёт сто измерений в секунду, сутки работы дают 8 640 000 значений. Интерфейсу нужен один взгляд в секунду, базе достаточно агрегата раз в пять секунд, графику нужна сглаженная линия. Три потребности закрываются тремя операторами:

IObservable<double> sensor = device.Temperature; // 100 значений в секунду

var uiFeed = sensor.Sample(TimeSpan.FromSeconds(1));

var dbFeed = sensor
    .Buffer(TimeSpan.FromSeconds(5), 100)
    .Where(chunk => chunk.Count > 0)
    .Select(chunk => new
    {
        avg = chunk.Average(),
        max = chunk.Max(),
        count = chunk.Count
    });

var smooth = sensor.Scan(seed: 20.0, (acc, x) => acc + 0.2 * (x - acc));

Sample выдаёт последнее значение в интервале и срезает поток в сто раз: вместо 8 640 000 значений в сутки интерфейс получает 86 400. Buffer собирает пачки по пять секунд или по сотне измерений и превращает поток в готовые агрегаты: 17 280 строк в сутки со средним и максимумом вместо миллионов вставок. Scan работает как агрегатная функция, но выдаёт каждое промежуточное значение, а формула acc + 0.2 * (x - acc) описывает экспоненциальное сглаживание с коэффициентом 0.2: дрожащий сигнал превращается в ровную линию.

Ручная версия потребует System.Timers.Timer и потокобезопасной очереди, реактивная умещается в три выражения.

Всплески событий FileSystemWatcher и обработка затиший в наблюдаемой папке

FileSystemWatcher дарит сюрприз каждому второму: копирование двухсот файлов порождает лавину уведомлений, сохранение одного файла поднимает несколько событий Changed подряд. Наивный обработчик гоняет один и тот же файл по кругу, а при плотном потоке событий наблюдатель бросает InternalBufferOverflowException, его внутренний буфер не безразмерен.

Реактивное решение собирает события в партии:

var watcher = new FileSystemWatcher(@"C:\Incoming")
{
    EnableRaisingEvents = true
};

var arrivals = Observable
    .FromEventPattern<FileSystemEventHandler, FileSystemEventArgs>(
        h => watcher.Created += h,
        h => watcher.Created -= h)
    .Select(e => e.EventArgs.FullPath)
    .Buffer(TimeSpan.FromSeconds(2))
    .Where(batch => batch.Count > 0);

using var handle = arrivals.Subscribe(files => ProcessIncoming(files));

Buffer с интервалом в две секунды собирает значения в список, Where отсекает пустые интервалы, обработчик получает готовую партию путей. Второй вариант давно стал классикой: копить значения, пока события идут потоком, и обрабатывать накопленное после периода тишины. Смысл один: всплеск допустимо дождаться целиком, прежде чем реагировать.

Повторные попытки с экспоненциальной задержкой для неустойчивых внешних сервисов

Внешний HTTP-сервис мигает: раз в неделю отвечает таймаутом, и наивный Retry(5) повторяет запрос мгновенно, добивая сервер, который и без того захлёбывается. В реактивном стиле экспоненциальная задержка пишется без единого цикла:

static IObservable<T> RetryWithBackoff<T>(
    Func<IObservable<T>> source,
    int attempts,
    TimeSpan delay)
{
    if (attempts <= 1)
    {
        return source();
    }

    return Observable.Catch(source(), (Exception _) =>
        Observable.Timer(delay).SelectMany(__ =>
            RetryWithBackoff(
                source,
                attempts - 1,
                TimeSpan.FromTicks(delay.Ticks * 2))));
}

Использование сворачивается в один вызов:

var status = RetryWithBackoff(
    () => Observable.FromAsync(ct => client.GetAsync("/status", ct)),
    attempts: 5,
    delay: TimeSpan.FromSeconds(1));

Первый сбой уводит повтор на секунду, второй на две, дальше четыре, восемь и шестнадцать. Observable.Catch подменяет упавшую последовательность, Observable.Timer изображает паузу, рекурсия уменьшает счётчик попыток, и до подписчика доезжает только ошибка последней. Для чистых HTTP-повторов есть библиотека Polly, но внутри реактивного конвейера родной оператор органичнее.

Где реактивность лишняя и выручают события Channel и IAsyncEnumerable

Не каждое событие заслуживает потока. Одиночный клик кнопки, меняющий один флаг, пишется обычным обработчиком за минуту, и оборачивание его в IObservable ради галочки лишь умножает сущности. У платформы хватает штатных альтернатив.

Очередь между производителем и потребителем ложится на Channel:

using System.Threading.Channels;

var queue = Channel.CreateBounded<Job>(
    new BoundedChannelOptions(300)
    {
        FullMode = BoundedChannelFullMode.Wait
    });

await queue.Writer.WriteAsync(new Job(payload));

await foreach (var job in queue.Reader.ReadAllAsync())
{
    await job.ProcessAsync();
}

Ограниченный канал честно пережимает давление: когда он заполнен, WriteAsync приостанавливает производителя, быстрый источник не топит медленную обработку. У Rx с давлением хуже: источник проталкивает значения независимо от готовности потребителя, ObserveOn лишь переносит очередь в другой поток.

Итерация по записям, которые потребитель тянет сам, ложится на IAsyncEnumerable:

async IAsyncEnumerable<Order> ReadOrdersAsync(
    HttpClient http, CancellationToken ct)
{
    int page = 1;
    while (true)
    {
        var batch = await http.GetFromJsonAsync<List<Order>>(
            $"/orders?page={page}", ct);

        if (batch is null || batch.Count == 0)
        {
            yield break;
        }

        foreach (var order in batch)
        {
            yield return order;
        }

        page++;
    }
}

Потребитель сам решает, когда запросить следующий элемент, и медленный клиент не даёт источнику улететь вперёд. С .NET 9 есть и Task.WhenEach для обработки задач по мере готовности:

await foreach (var done in Task.WhenEach(job1, job2, job3))
{
    Report(await done);
}

Правило простое: события приходят извне, темп задаёт источник, прошлое надо уметь отменять, и здесь конвейер Rx окупается; поток между своими компонентами с темпом потребителя дешевле обслужить Channel и await foreach.

Типичные ловушки Rx.NET от утечки подписок до Subject вместо архитектуры

Первая ловушка, подписка переживает экран. Замыкание держит ссылки на модель и элементы интерфейса, и в долгоживущем приложении память копится неделями. Лечение одно:

var subscriptions = new CompositeDisposable();

subscriptions.Add(searchFeed.Subscribe(ShowResults));
subscriptions.Add(statusFeed.Subscribe(ShowStatus));

void OnClosed() => subscriptions.Dispose(); // при закрытии экрана

Вторая ловушка, Subject<T> на роли глобальной шины событий. Получается сервис-локатор на минималках: зависимости прячутся в неявные потоки, тесты требуют разогрева всей кухни, порядок инициализации влияет на результат. BehaviorSubject, помнящий последнее значение, полезен для текущего состояния вроде выбранной вкладки, но не годится для передачи каждого клика.

Третья ловушка, потоки. События приезжают из пула потоков, интерфейс требует возвращения в свой контекст, и ObserveOn(SynchronizationContext.Current) спасает приложения WPF и WinForms. Различают и планировщики: SubscribeOn выбирает, где начнётся подписка, ObserveOn решает, где значения доедут до подписчика.

Четвёртая ловушка, холодные последовательности выполняются на каждую подписку. FromAsync превратит два вызова Subscribe в два HTTP-запроса, а связка Publish с RefCount делает один источник на всех подписчиков.

Отдельная радость реактивного кода в тестируемости. Пакет Microsoft.Reactive.Testing даёт TestScheduler: Throttle(TimeSpan.FromMilliseconds(250)) в unit-тесте срабатывает мгновенно по команде AdvanceBy, без Thread.Sleep.

Проверочный фильтр перед выбором состоит из четырёх вопросов:
1. событие одиночное, без временной логики и без отмены прошлой работы, обычный обработчик проще и честнее;
2. источник внешний, темп важен, прошлую работу надо отменять при новых событиях, конвейер из Throttle и Switch окупается сразу;
3. записи тянет потребитель по своей готовности, подходят IAsyncEnumerable либо Channel с ограниченной ёмкостью;
4. нужны повторы с паузами и координация источников, остаются Retry с задержкой, Merge и Zip.

У System.Reactive 7.0 заслуженное место на полке инструментов .NET-разработчика, и место это там, где время и события сплетаются в сложные узоры: поиск с подавлением, телеметрия с прореживанием, повторы с задержками. За пределами этих сценариев платформа предлагает Channel, IAsyncEnumerable и Task.WhenEach, которые дешевле в сопровождении. Зрелость инженера видна не по числу операторов Rx в коде, а по трезвой оценке, где конвейер решает задачу, а где лишь демонстрирует энтузиазм автора. Скальпель хорош в руках хирурга, но консервы им не открывают, и признание этого экономит недели.