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

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

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

Что такое RxPHP?

RxPHP — это PHP-портированная версия популярного фреймворка ReactiveX (Rx). Она предоставляет инструменты для композиции асинхронных и событийно-ориентированных программ с помощью наблюдаемых последовательностей и LINQ-подобных операторов запросов.

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

• Observable — источник данных, который может испускать значения, ошибки или сигнал завершения.
• Observer — подписчик, который реагирует на события от Observable.
• Subscription — связь между Observable и Observer, позволяющая отписываться.
• Operators — функции для трансформации, фильтрации и комбинирования потоков (map, filter, merge, zip и др.).
• Scheduler — управление выполнением (синхронное, асинхронное, с задержкой).

Установка RxPHP

Установка производится через Composer:

composer require reactivex/rxphp

Пример 1: Создание простого Observable

Создадим поток, который последовательно выдает три числа и завершается:

<?php
require 'vendor/autoload.php';

use Rx\Observable;

$observable = Observable::fromArray([10, 20, 30]);

$disposable = $observable->subscribe(
    function ($value) {
        echo "Получено: $value\n";
    },
    function (\Throwable $error) {
        echo "Ошибка: " . $error->getMessage() . "\n";
    },
    function () {
        echo "Завершено!\n";
    }
);

Результат выполнения:

Получено: 10
Получено: 20
Получено: 30
Завершено!

Пример 2: Использование операторов map и filter

Трансформируем поток чисел: отфильтруем чётные и умножим их на 10:

<?php
use Rx\Observable;

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

$observable->subscribe(function ($value) {
    echo "Результат: $value\n";
});

Вывод:

Результат: 20
Результат: 40

Пример 3: Асинхронная работа с HttpClient

RxPHP отлично подходит для работы с асинхронными HTTP-запросами. Используем Observable::fromPromise с библиотекой GuzzleHttp:

<?php
use Rx\Observable;
use GuzzleHttp\Client;
use GuzzleHttp\Promise;

$client = new Client();
$promise1 = $client->getAsync('https://api.example.com/data1');
$promise2 = $client->getAsync('https://api.example.com/data2');

Observable::fromArray([$promise1, $promise2])
    ->flatMap(function ($promise) {
        return Observable::fromPromise($promise);
    })
    ->map(function ($response) {
        return json_decode($response->getBody(), true);
    })
    ->subscribe(function ($data) {
        print_r($data);
    });

Пример 4: Объединение потоков (merge и zip)

Часто требуется объединить несколько потоков в один. Рассмотрим два оператора:

merge — объединяет потоки, чередуя их элементы по мере поступления:

<?php
use Rx\Observable;

$stream1 = Observable::fromArray(['A1', 'A2', 'A3']);
$stream2 = Observable::fromArray(['B1', 'B2']);

Observable::merge([$stream1, $stream2])
    ->subscribe(function ($value) {
        echo "$value ";
    });
// Вывод: A1 B1 A2 B2 A3

zip — попарно объединяет элементы из разных потоков (как застежка-молния):

<?php
use Rx\Observable;

$names = Observable::fromArray(['Alice', 'Bob', 'Charlie']);
$ages = Observable::fromArray([25, 30, 35]);

Observable::zip([$names, $ages], function ($name, $age) {
    return "$name is $age years old";
})->subscribe(function ($result) {
    echo "$result\n";
});
// Вывод:
// Alice is 25 years old
// Bob is 30 years old
// Charlie is 35