Реактивное программирование в PHP: библиотека RxPHP для асинхронных потоков

Реактивное программирование в PHP: практическое руководство по RxPHP

Реактивное программирование — это парадигма, ориентированная на работу с асинхронными потоками данных. В PHP, традиционно синхронном языке, эта концепция реализована через библиотеку RxPHP (ReactiveX для PHP). Она позволяет обрабатывать события, HTTP-запросы, файловые потоки и пользовательские действия в едином, декларативном стиле.

Что такое RxPHP?

RxPHP — это порт популярной библиотеки ReactiveX. Основные компоненты:

• Observable — источник данных, который "излучает" события (onNext, onError, onCompleted).
• Observer — подписчик, который реагирует на события.
• Schedulers — управляют контекстом выполнения (синхронный, асинхронный, на новом потоке).
• Операторы — функции для трансформации, фильтрации и комбинирования потоков (map, filter, merge, zip).

Установка и настройка

Для работы с RxPHP используйте Composer:

composer require reactivex/rxphp

Базовый пример: создание Observable и подписка:

<?php
use Rx\Observable;
use Rx\ObserverInterface;

$observable = Observable::fromArray([1, 2, 3, 4, 5]);

$observer = new class implements ObserverInterface {
    public function onNext($value) {
        echo "Получено: $value\n";
    }
    public function onError(\Throwable $error) {
        echo "Ошибка: " . $error->getMessage() . "\n";
    }
    public function onCompleted() {
        echo "Завершено!\n";
    }
};

$observable->subscribe($observer);
// Вывод:
// Получено: 1
// Получено: 2
// Получено: 3
// Получено: 4
// Получено: 5
// Завершено!
?>

Основные операторы RxPHP

Операторы позволяют трансформировать данные без изменения исходного Observable:

1. map — преобразование каждого элемента

$observable = Observable::fromArray([1, 2, 3])
    ->map(function($value) {
        return $value * 2;
    });

$observable->subscribe(function($value) {
    echo $value . "\n";
});
// 2, 4, 6

2. filter — отбор по условию

$observable = Observable::fromArray([1, 2, 3, 4, 5])
    ->filter(function($value) {
        return $value % 2 === 0;
    });

// Результат: 2, 4

3. merge — объединение потоков

$stream1 = Observable::fromArray([1, 2]);
$stream2 = Observable::fromArray([3, 4]);

$merged = Observable::merge($stream1, $stream2);
// 1, 2, 3, 4 (порядок может зависеть от Scheduler)

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

Можно создавать Observable из любых источников: базы данных, сокетов или пользовательских событий.

<?php
use Rx\Observable;
use Rx\Scheduler\EventLoopScheduler;

// Эмуляция асинхронного чтения файла
function readFileAsync(string $filename): Observable {
    return Observable::create(function($observer) use ($filename) {
        $handle = fopen($filename, 'r');
        if (!$handle) {
            $observer->onError(new \RuntimeException("Не удалось открыть файл"));
            return;
        }
        
        while (($line = fgets($handle)) !== false) {
            $observer->onNext(trim($line));
        }
        fclose($handle);
        $observer->onCompleted();
    });
}

// Использование с EventLoop (требует react/event-loop)
$loop = \React\EventLoop\Loop::get();
$scheduler = new EventLoopScheduler($loop);

readFileAsync('data.log')
    ->subscribeOn($scheduler)
    ->filter(function($line) {
        return strpos($line, 'ERROR') !== false;
    })
    ->subscribe(function($line) {
        echo "Найдена ошибка: $line\n";
    });

$loop->run();
?>

Обработка ошибок и завершение

В реактивном стиле ошибки обрабатываются через onError, а завершение — через onCompleted. Это позволяет избежать вложенных try-catch.

$observable = Observable::fromArray([1, 2, 0, 3])
    ->map(function($value) {
        if ($value === 0) {
            throw new \InvalidArgumentException("Деление на ноль");
        }
        return 100 / $value;
    })
    ->onErrorResumeNext(function(\Throwable $e) {
        echo "Ошибка: " . $e->getMessage() . "\n";
        return Observable::empty(); // Продолжаем без ошибок
    });

$observable->subscribe(function($value) {
    echo $value