Реактивное программирование в C++: RxCpp и Boost.Signals2…

Реактивное программирование в C++: обзор библиотек RxCpp и Boost.Signals2

Реактивное программирование (Reactive Programming) — это парадигма, ориентированная на асинхронные потоки данных и распространение изменений. В C++ эта парадигма реализуется через библиотеки, которые позволяют обрабатывать события, сигналы и асинхронные последовательности данных декларативным способом. В данной статье мы рассмотрим две ключевые библиотеки для реактивного программирования на C++: RxCpp (Reactive Extensions for C++) и Boost.Signals2. Обе библиотеки решают задачу обработки событий, но с разной степенью сложности и функциональности.

Зачем нужно реактивное программирование в C++?

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

Декларативный подход: Вы описываете "что" должно произойти, а не "как".
Композиция потоков: Легко объединять, фильтровать и трансформировать потоки данных.
Управление многопоточностью: Встроенные механизмы для переключения контекстов и синхронизации.
Обработка ошибок: Единый конвейер для данных и исключений.

Библиотека RxCpp

RxCpp — это порт реактивных расширений (Rx) для C++. Она предоставляет мощный набор операторов для работы с асинхронными последовательностями данных. Основные концепции: observable (источник данных) и observer (подписчик).

Установка

RxCpp — header-only библиотека. Достаточно склонировать репозиторий и подключить заголовочные файлы:

#include <rxcpp/rx.hpp>
using namespace rxcpp;
using namespace rxcpp::sources;
using namespace rxcpp::operators;

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

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

#include <iostream>
#include <rxcpp/rx.hpp>

int main() {
    auto values = range(1, 5); // observable: 1,2,3,4,5

    values.
        subscribe([](int v) {
            std::cout << "Received: " << v << std::endl;
        },
        []() {
            std::cout << "Stream completed" << std::endl;
        });

    return 0;
}

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

Реактивные операторы позволяют легко манипулировать данными. Отфильтруем чётные числа и умножим их на 10:

auto values = range(1, 10);

values
    .filter([](int v) { return v % 2 == 0; }) // только чётные
    .map([](int v) { return v * 10; })        // умножаем на 10
    .subscribe([](int v) {
        std::cout << "Result: " << v << std::endl;
    });
// Вывод: 20, 40, 60, 80, 100

Пример 3: Объединение потоков

Объединим два потока с помощью оператора merge:

auto stream1 = interval(std::chrono::milliseconds(100)).take(3);
auto stream2 = interval(std::chrono::milliseconds(150)).take(3);

stream1.merge(stream2)
    .subscribe([](long v) {
        std::cout << v << " ";
    });
// Возможный вывод: 0 0 1 2 1 2 (порядок может отличаться)

Библиотека Boost.Signals2

Boost.Signals2 — это библиотека для реализации сигналов и слотов (signals & slots). Она является частью библиотек Boost и предоставляет типобезопасный механизм для обратных вызовов. В отличие от RxCpp, Signals2 не предназначена для работы с потоками данных, а скорее для обработки дискретных событий.

Основные концепции

Сигнал (signal): объект, который может быть "вызван" и уведомит всех своих подписчиков.
Слот (slot): функция или функциональный объект, подключенный к сигналу.
Соединение (connection): объект, представляющий связь между сигналом и слотом, позволяющий отключать слот.