使用 RxPHP 在 PHP 中实现响应式编程

响应式编程是一种面向异步数据流和变化传播的编程范式。在 PHP 中，这种范式通过 RxPHP（PHP 的 ReactiveX）库实现，它允许将事件、异步请求和数据序列作为可观察流（Observables）进行处理。

什么是 RxPHP？

RxPHP 是流行框架 ReactiveX (Rx) 的 PHP 移植版本。它通过可观察序列和类似 LINQ 的查询操作符，提供了组合异步和事件驱动程序的工具。

RxPHP 的核心概念：

Observable — 数据源，可以发出值、错误或完成信号。
Observer — 订阅者，响应 Observable 发出的事件。
Subscription — Observable 和 Observer 之间的连接，允许取消订阅。
Operators — 用于转换、过滤和组合流的函数（如 map、filter、merge、zip 等）。
Scheduler — 执行管理（同步、异步、延迟）。

安装 RxPHP

通过 Composer 安装：

composer require reactivex/rxphp

示例 1：创建简单的 Observable

创建一个流，依次发出三个数字并完成：

<?phprequire 'vendor/autoload.php';

use Rx\\Observable;

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

$disposable = $observable->subscribe(    function ($value) {        echo "收到: $value";    },    function (\\Throwable $error) {        echo "错误: " . $error->getMessage() . "";    },    function () {        echo "完成!";    });

执行结果：

收到: 10收到: 20收到: 30完成!

示例 2：使用 map 和 filter 操作符

转换数字流：过滤出偶数并乘以 10：

<?phpuse 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";});

输出：

结果: 20结果: 40

示例 3：使用 HttpClient 进行异步操作

RxPHP 非常适合处理异步 HTTP 请求。使用 Observable::fromPromise 和 GuzzleHttp 库：

<?phpuse 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 — 合并流，按到达顺序交错排列元素：

<?phpuse 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