如何用ReactPHP实现PHP并发？解决任务无法同时启动问题

问题描述 #

尝试用ReactPHP实现多任务同时处理，发现各任务启动时间不一致，期望所有任务在同一时间启动。

原代码 #

<?php
require 'vendor/autoload.php';
use React\EventLoop\Factory;
use React\Promise\Promise;

$loop = Factory::create();

function asyncTask($taskName)
{
    return new Promise(function ($resolve) use ($taskName) {
        $startTime = (int)microtime(true);
        echo "Starting $taskName at " . date('Y-m-d H:i:s', $startTime) . "\n";

        $loop = React\EventLoop\Factory::create();
        $loop->addTimer(rand(1, 5), function () use ($resolve, $taskName, $startTime) {
            $endTime = (int)microtime(true);
            echo "Completed $taskName at " . date('Y-m-d H:i:s', $endTime) . "\n";

            $resolve([
                'taskName' => $taskName,
                'startTime' => $startTime,
                'endTime' => $endTime,
            ]);
        });

        $loop->run();
    });
}

$promises = [
    'Task 1' => asyncTask('Task 1'),
    'Task 2' => asyncTask('Task 2'),
    'Task 3' => asyncTask('Task 3'),
];

\React\Promise\all($promises)->then(function ($results) {
    echo "All tasks completed: <br>";
    foreach ($results as $result) {
        echo "{$result['taskName']} started at " . date('Y-m-d H:i:s', $result['startTime']) . " and ended at " . date('Y-m-d H:i:s', $result['endTime']) . "\n";
    }
});

$loop->run();
?>

当前输出 #

Starting Task 1 at 2024-01-18 17:37:09

Completed Task 1 at 2024-01-18 17:37:10

Starting Task 2 at 2024-01-18 17:37:10

Completed Task 2 at 2024-01-18 17:37:14

Starting Task 3 at 2024-01-18 17:37:14

Completed Task 3 at 2024-01-18 17:37:15

All tasks completed:

Task 1 started at 2024-01-18 17:37:09 and ended at 2024-01-18 17:37:10

Task 2 started at 2024-01-18 17:37:10 and ended at 2024-01-18 17:37:14

Task 3 started at 2024-01-18 17:37:14 and ended at 2024-01-18 17:37:15

问题根源 #

1. 每个asyncTask内部新建独立的EventLoop并调用run()，会阻塞当前进程直到该任务定时器执行完毕，导致任务串行执行而非并行。
2. 任务启动时间在各自EventLoop启动时才记录，自然出现时间差。

修正后的代码 #

<?php
require 'vendor/autoload.php';
use React\EventLoop\Factory;
use React\Promise\Promise;

$loop = Factory::create();
// 统一记录所有任务的启动时间
$globalStartTime = (int)microtime(true);

function asyncTask($taskName, $loop, $startTime)
{
    return new Promise(function ($resolve) use ($taskName, $loop, $startTime) {
        echo "Starting $taskName at " . date('Y-m-d H:i:s', $startTime) . "\n";

        $loop->addTimer(rand(1, 5), function () use ($resolve, $taskName, $startTime) {
            $endTime = (int)microtime(true);
            echo "Completed $taskName at " . date('Y-m-d H:i:s', $endTime) . "\n";

            $resolve([
                'taskName' => $taskName,
                'startTime' => $startTime,
                'endTime' => $endTime,
            ]);
        });
    });
}

$promises = [
    'Task 1' => asyncTask('Task 1', $loop, $globalStartTime),
    'Task 2' => asyncTask('Task 2', $loop, $globalStartTime),
    'Task 3' => asyncTask('Task 3', $loop, $globalStartTime),
];

\React\Promise\all($promises)->then(function ($results) {
    echo "All tasks completed:\n";
    foreach ($results as $result) {
        echo "{$result['taskName']} started at " . date('Y-m-d H:i:s', $result['startTime']) . " and ended at " . date('Y-m-d H:i:s', $result['endTime']) . "\n";
    }
});

$loop->run();
?>

修正说明 #

1. 共用全局EventLoop，所有任务的定时器添加到同一个loop中，由loop统一调度实现并行。
2. 创建任务前统一记录globalStartTime，确保所有任务使用相同的启动时间。
3. 移除任务内部的$loop->run()调用，仅保留全局loop的run()，避免阻塞串行执行。

预期输出 #

Starting Task 1 at 2024-01-18 17:37:09
Starting Task 2 at 2024-01-18 17:37:09
Starting Task 3 at 2024-01-18 17:37:09
Completed Task 1 at 2024-01-18 17:37:10
Completed Task 3 at 2024-01-18 17:37:12
Completed Task 2 at 2024-01-18 17:37:14
All tasks completed:
Task 1 started at 2024-01-18 17:37:09 and ended at 2024-01-18 17:37:10
Task 2 started at 2024-01-18 17:37:09 and ended at 2024-01-18 17:37:14
Task 3 started at 2024-01-18 17:37:09 and ended at 2024-01-18 17:37:12

内容的提问来源于stack exchange，提问作者Naveed Ahmad Baig