Основы обработки асинхронных событий с помощью Rx в C#

Привет, Хабр!

Reactive Extensions, известные также как Rx.NET, представляют собой библиотеку для .NET, предназначенную для композиции асинхронных и событийно-ориентированных программ с помощью наблюдаемых последовательностей и операторов запросов в стиле LINQ.

Основная фича Rx заключается в управлении асинхронными данными так, как если бы они были синхронными коллекциями.

В этой статье мы рассмотрми, как работать с Rx в C#.

Немного теории

Observables – это основа Rx, представляющая потоки данных, которые можно наблюдать. Observables могут генерировать данные с течением времени, будь то из массивов, событий или тех же асинхронных запросов.

Observers – объекты, которые подписываются на Observables и реагируют на данные, поступающие из них. Observer реализует три метода, которые вызываются соответственно при получении данных, ошибки или завершении потока. С помощью метода Subscribe, Observer подключается к Observable и начинает получать уведомления.

Операторы выполняю основную роль манипуляции и управлении потоками данных. Они позволяют трансформировать, фильтровать, комбинировать и даже создавать новые потоки на основе существующих.

Синтаксис Rx

Для создания Observable последовательностей в Rx используется метод Observable.Create. Этот метод позволяет определить логику создания потока данных, которая будет исполняться при подписке на Observable:

IObservable<int> observable = Observable.Create(observer => {
    observer.OnNext(1);
    observer.OnNext(2);
    observer.OnNext(3);
    observer.OnCompleted();
    return Disposable.Empty;
});

Операторы:

Select:

var observable = Observable.Range(1, 5);
var selected = observable.Select(x => x * 10);
selected.Subscribe(Console.WriteLine); // вывод: 10, 20, 30, 40, 50

Where:

var observable = Observable.Range(1, 10);
var filtered = observable.Where(x => x % 2 == 0);
filtered.Subscribe(Console.WriteLine); // вывод: 2, 4, 6, 8, 10

Aggregate:

var observable = Observable.Range(1, 5);
var aggregated = observable.Aggregate((acc, x) => acc + x);
aggregated.Subscribe(Console.WriteLine); // вывод: 15

Работа с множественными источниками данных:

Merge:

var first = Observable.Interval(TimeSpan.FromSeconds(1)).Take(3);
var second = Observable.Interval(TimeSpan.FromSeconds(2)).Take(2);
var merged = Observable.Merge(first, second);
merged.Subscribe(Console.WriteLine); // вывод: 0, 0, 1, 1, 2

Concat:

var first = Observable.Range(1, 3);
var second = Observable.Range(4, 3);
var concatenated = Observable.Concat(first, second);
concatenated.Subscribe(Console.WriteLine); // вывод: 1, 2, 3, 4, 5, 6

Zip:

var numbers = Observable.Range(1, 3);
var letters = Observable.From(new[] {'a', 'b', 'c'});
var zipped = Observable.Zip(numbers, letters, (n, l) => $"{n}{l}");
zipped.Subscribe(Console.WriteLine); // вывод: 1a, 2b, 3c

Управление памятью и избежание утечек с помощью Dispose

Часто при работе с Rx используют Dispose для предотвращения утечек памяти. Когда вы подписываетесь на Observable, Rx.NET возвращает объект IDisposable, который нужно корректно утилизировать после окончания использования подписки. Это используют в проектах, где подписки могут динамически создаваться и удаляться.

var subscription = observable.Subscribe(item => Console.WriteLine(item));
// после завершения использования подписки, не забываем освободить ресурсы
subscription.Dispose();

Примеры использования

Обработка кнопочных событий в пользовательском интерфейсе

Допустим, есть форма с кнопкой, и хочется реагировать на клики по этой кнопке, но только если клики происходят с интервалом не менее одной секунды, чтобы избежать двойного клика:

// подключение Rx и Windows Forms
using System;
using System.Windows.Forms;
using System.Reactive.Linq;

public class ReactiveForm : Form
{
    private Button myButton;

    public ReactiveForm()
    {
        myButton = new Button { Text = "Click me!" };
        Controls.Add(myButton);

        // создание потока событий клика
        var clicks = Observable.FromEventPattern(
            h => myButton.Click += h,
            h => myButton.Click -= h);

        // фильтрация событий с интервалом в 1 секунду
        clicks
            .Throttle(TimeSpan.FromSeconds(1))
            .Subscribe(_ => Console.WriteLine("Button clicked!"));
    }
}

Комбинирование данных из нескольких источников

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

IObservable<decimal> prices = Observable.Return(100m); // пример потока цен
IObservable<decimal> discounts = Observable.Return(0.1m); // пример потока скидок

var finalPrice = prices.Zip(discounts, (price, discount) => price * (1 - discount));
finalPrice.Subscribe(price => Console.WriteLine($"Final price: {price}"));

Отслеживание состояния кнопки с задержкой

Если хочется отслеживать состояние кнопки, но с задержкой, чтобы предотвратить реакцию на случайные или кратковременные нажатия:

var buttonStates = Observable.FromEventPattern<StateChangedEventArgs>(button, "StateChanged");

buttonStates
    .Select(evt => evt.EventArgs.NewState)
    .DistinctUntilChanged() // игнорирование повторяющихся состояний
    .Throttle(TimeSpan.FromMilliseconds(500)) // задержка для стабилизации состояния
    .Subscribe(state => Console.WriteLine($"Button state stabilized to: {state}"));

Реактивное управление потоками данных API

Используем Rx для управления асинхронными вызовами API для получения погоды. Пример того, как можно организовать повторные запросы и обработку ошибок:

IObservable<string> weatherData = Observable.Interval(TimeSpan.FromMinutes(1))
    .SelectMany(_ => Observable.FromAsync(() => GetWeatherAsync()))  // асинхронный вызов API
    .Retry(3)  // повтор при ошибке до 3 раз
    .Catch<string, Exception>(ex => Observable.Return("Error fetching weather data"));

weatherData.Subscribe(data => Console.WriteLine($"Current weather: {data}"),
                      error => Console.WriteLine($"An error occurred: {error}"));

---

Reactive Extensions позволяет писать более чистый и реактивный код.

Статья подготовлена в преддверии старта специализации C# Developer. На странице специализации вы можете подробнее узнать о программе, а также зарегистрироваться на бесплатные вебинары.