别再让阻塞拖垮 PHP!异步开发让脚本执行效率飙升 10 倍

{"id":1,"value":42}{"id":2,"value":17}{"id":3,"value":88}

$line = fgets(STDIN);          // 读取到 \n 为止$task = json_decode($line);    // 解析 JSON

function run_worker(string $script, array $tasks): void{    $process = proc_open(        [PHP_BINARY, $script],        [            0 => ['pipe', 'r'],   // 子进程 STDIN            1 => ['pipe', 'w'],   // 子进程 STDOUT            2 => STDERR,        ],        $pipes    );    foreach ($tasks as $task) {        // 发送任务        fwrite($pipes[0], json_encode($task) . "\n");        // 等待响应 ← PHP 在这里阻塞，直到读取到一行        $line   = fgets($pipes[1]);        $result = json_decode(trim($line), true);        echo "Task #{$result['task_id']}: {$result['result']}\n";    }    fclose($pipes[0]);   // 关闭 STDIN → worker.php 将退出    fclose($pipes[1]);    proc_close($process);}run_worker('worker.php', [    ['id' => 1, 'value' => 42],    ['id' => 2, 'value' => 17],    ['id' => 3, 'value' => 88],]);

$tasks = [    ['id' => 1, 'value' => 42],    ['id' => 2, 'value' => 17],    // ... 还有 28 个任务];for ($i = 0; $i < 10; $i++) {    spawn(run_worker(...), 'worker.php', $tasks);}

$taskQueue = new Async\Channel(10);for ($i = 0; $i < 10; $i++) {    spawn(run_worker(...), 'worker.php', $taskQueue);}foreach ($tasks as $task) {    $channel->send($task);}$channel->close();

$taskQueue = new Async\Channel(10);$group     = new Async\TaskGroup();try {    // 在组管理下启动 workers    for ($i = 0; $i < 10; $i++) {        $group->spawn(run_worker(...), 'worker.php', $taskQueue);    }    // 向 Channel 投喂任务    foreach ($tasks as $task) {        $taskQueue->send($task);    }} finally {    // 关闭任务 Channel —— 尝试读取的协程    // 会收到 `ChannelClosedException` 并优雅关闭    $taskQueue->close();    // 封印组，防止意外添加新协程    $group->seal();                // 防止添加新协程    // 等待尚未完成的任务    $group->all()->await();        // 等待所有 worker 完成}

function run_worker(string $script, Channel $taskQueue): void{    return;}$taskQueue = new Channel(POOL_SIZE);$group     = new TaskGroup();try {    for ($i = 0; $i < POOL_SIZE; $i++) {        $group->spawn(run_worker(...), __DIR__ . '/worker.php', $taskQueue);    }    foreach (generate_tasks(TASK_COUNT) as $task) {        $taskQueue->send($task);    }} finally {    $taskQueue->close();    $group->seal();    $group->all()->await();    echo "\nDone.\n";}

=== DEADLOCK REPORT START ===Coroutines waiting: 1, active_events: 0Coroutine 4 spawned at main:0, suspended at main.php:96 (main)  waiting for:    - Channel(capacity=3, receivers=0, senders=1)=== DEADLOCK REPORT END ===

spawn(function () use ($taskQueue, $group) {    try {        $group->all()->await();    } finally {        $taskQueue->close();    }});

foreach (generate_tasks(TASK_COUNT) as $task) {    $taskQueue->send($task, $group->all());}

foreach ($taskQueue as $task) {    $encoded = json_encode($task);    if ($encoded === false) {        echo "[worker] json_encode failed for task #{$task['id']}: " . json_last_error_msg() . "\n";        return;    }    if (fwrite($pipes[0], $encoded . "\n")) {        echo "[worker] fwrite failed for task #{$task['id']}: pipe may be broken\n";        return;    }    $line = fgets($pipes[1]);    if ($line === false && stream_get_meta_data($pipes[1])['timed_out']) {        echo "[worker] timeout waiting for response on task #{$task['id']}\n";        return;    } else if ($line === false) {        echo "[worker] fgets failed for task #{$task['id']}: pipe closed or EOF\n";        return;    }    $result = json_decode(trim($line), true);    if ($result === null) {        echo "[worker] json_decode failed for task #{$task['id']}: " . json_last_error_msg() . " (raw: " . trim($line) . ")\n";        return;    }}

php.exe E:\php\examples\workers_process_cli\main2.php[worker] fwrite failed for task #1: pipe may be broken[worker] fwrite failed for task #2: pipe may be broken[worker] fwrite failed for task #3: pipe may be brokenDone.

现在可以确信，即使 worker 崩溃，主进程也不会挂起，而是继续运行。

如果进程池能自动启动新进程以保持总数恒定，那就更好了。怎么做？可以复杂化协程代码，添加显式重启进程的循环。把这个练习留给喜欢复杂代码和分形的爱好者；这不是我们的选择，因为很容易不小心破坏什么。协程内部的线性、简单和最小分支是它的主要优点。让我们保持这样。

既然协程在概念上等同于进程，那么在进程交互结束时同时结束协程是合理的。这意味着可以响应"协程完成"事件。TaskGroup 类有合适的 race() 方法，在协程完成时触发。唯一的问题是 TaskGroup 是幂等的；多次调用 race() 总是返回相同结果。

所以需要不同的工具。TaskSet 类非常适合这个需求——它是 TaskGroup 的孪生兄弟，非常适合 supervisor 逻辑。大多数方法相同；区别在于 join* 方法组，它不仅返回任务结果，同时从 TaskSet 中移除它。正是需要的！

$taskQueue = new Channel(POOL_SIZE);$group     = new TaskSet();try {    for ($i = 0; $i < POOL_SIZE; $i++) {        $group->spawn(run_worker(...), __DIR__ . '/worker.php', $taskQueue);    }    foreach (generate_tasks(TASK_COUNT) as $task) {        $taskQueue->send($task, $group->joinAll());    }} finally {    $taskQueue->close();    $group->seal();    $group->joinAll()->await();    echo "\nDone.\n";}

现在添加 supervisor 逻辑，在进程崩溃时重启：

$supervisor = spawn(function () use ($group, $taskQueue): void {    while (true) {        try {            // 等待至少一个任务完成            $group->joinNext()->await();        } catch (\Exception $exception) {            // 发生了真正意想不到的事            echo "[supervisor] Exception: {$exception->getMessage()}\n";            $group->seal();            $taskQueue->close();            return;        }        // 如果组被封印或 Channel 已关闭，退出循环 ——        // 进程被有意停止，不应重启        if($group->isSealed() || $taskQueue->isClosed()) {            return;        }        $group->spawn(run_worker(...), __DIR__ . '/worker.php', $taskQueue);    }});

用 worker_fail.php 运行代码，观察到尽管 worker 崩溃，进程池仍在运行：

php.exe E:\php\examples\workers_process_cli\main2.php[worker] fwrite failed for task #1: pipe may be broken[worker] fwrite failed for task #2: pipe may be broken

条目数应该等于 TASK_COUNT，尽管没有任务被成功执行。太好了。supervisor 代码对任务如何执行一无所知。执行任务的协程对 supervisor 一无所知。这不仅产生了相当可读的代码，而且产生了低耦合的代码，易于维护。

稍微改进 supervisor，让它对重启不要过于宽容：如果有无限多的任务且 worker.php 有 bug，可能进入无限重启循环。

$supervisor = spawn(function () use ($group, $taskQueue): void {    $cooldown = 5;    $threshold = 2;    $restarts = 0;    $lastFail = time();    while (true) {        try {            $group->joinNext()->await();        } catch (\Exception $e) {            echo "[supervisor] Exception: {$e->getMessage()}\n";            $group->seal();            $taskQueue->close();            return;        }        if ($group->isSealed() || $taskQueue->isClosed()) {            return;        }        $now = time();        if (($now - $lastFail) < $cooldown) {            if ($restarts >= $threshold) {                echo "[supervisor] Too many worker failures, shutting down pool.\n";                $group->seal();                $taskQueue->close();                return;            }            $restarts++;        }        $lastFail = $now;        $group->spawn(run_worker(...), __DIR__ . '/worker.php', $taskQueue);    }});

现在给 worker 添加类似真实任务的东西。假设有一个需要解析并计算统计信息的大日志文件。解析文件是重度的 CPU 密集型操作：非常适合进程池。统计聚合应该在主进程中发生。为最小化内存使用，拆分工作：主进程只读取日志文件并干净地分成块，workers 处理这些块并返回结果。

function feed_chunks(string $filePath, int $chunkBytes, Channel $queue): int{    $fileSize = filesize($filePath);    $fp       = fopen($filePath, 'r');    $offset   = 0;    $chunkId  = 0;    try {        while ($offset < $fileSize) {            $end = min($offset + $chunkBytes, $fileSize);            // 对齐到下一行，避免把行切半            if ($end < $fileSize) {                fseek($fp, $end);                fgets($fp);                   // 跳过被切分行的剩余部分                $end = ftell($fp);            // 现在处于干净的行边界            }            $queue->send([                'id'     => ++$chunkId,                'file'   => $filePath,                'offset' => $offset,                'length' => $end - $offset,            ]);            $offset = $end;        }    } finally {        fclose($fp);    }    return $chunkId;}

只向 worker 发送偏移量和块长度，worker 自己打开文件并读取所需部分。这种方式让主进程快速把大任务拆分成部分，无需通过管道泵送真实数据。

添加另一个 Channel 用于结果，在单独的协程中处理。虽然可以直接从 worker 协程修改数组来达到相同效果，因为处于单个进程内，所有变量对任何协程都可见。但使用 Channel 看起来更"Go 风格"。

function run_worker(int $workerId, string $script, Channel $taskQueue, Channel $resultQueue): void{    $process = proc_open(        [PHP_BINARY, $script],        [            0 => ['pipe', 'r'],            1 => ['pipe', 'w'],            2 => STDERR,        ],        $pipes    );    if ($process === false) {        throw new RuntimeException("Worker #$workerId: proc_open failed");    }    echo "  Worker #$workerId started\n";    // 这个函数只在 TrueAsync 中有效 —— 在普通 PHP 中无法给管道设置超时    stream_set_timeout($pipes[1], 30);    try {        foreach ($taskQueue as $task) {            // 只发送轻量级描述符 —— 日志数据不通过管道传输            $encoded = json_encode($task);            if (!fwrite($pipes[0], $encoded . "\n")) {                throw new RuntimeException("Worker #$workerId: pipe broken on chunk #{$task['id']}");            }            $line = fgets($pipes[1]);            if ($line === false && stream_get_meta_data($pipes[1])['timed_out']) {                throw new RuntimeException("Worker #$workerId: timeout on chunk #{$task['id']}");            } elseif ($line === false) {                throw new RuntimeException("Worker #$workerId: process died on chunk #{$task['id']}");            }            $result = json_decode(trim($line), true);            if ($result === null) {                throw new RuntimeException("Worker #$workerId: invalid JSON on chunk #{$task['id']}");            }            printf(                "  [Worker #%d | pid %5d] chunk #%03d  (%s KB) → %d lines parsed\n",                $workerId,                $result['pid'],                $task['id'],                number_format($task['length'] / 1024),                $result['parsed'],            );            $resultQueue->send($result);        }    } finally {        fclose($pipes[0]);        fclose($pipes[1]);        proc_close($process);    }}

再添加一个协程，在结果到达时显示进度：

function run_collector(Channel $resultQueue): array{    $totals = [        'ip_counts'     => [],        'status_counts' => [],        'path_counts'   => [],        'method_counts' => [],        'total_bytes'   => 0,        'bot_requests'  => 0,    ];    $totalParsed = 0;    $totalErrors = 0;    try {        foreach ($resultQueue as $result) {            $totalParsed += $result['parsed'];            $totalErrors += $result['errors'];            merge_stats($totals, $result['stats']);        }    } catch (Async\ChannelException) {    }    return [$totals, $totalParsed, $totalErrors];}

最后把所有东西连接起来：

$taskQueue    = new Channel(POOL_SIZE);$resultQueue  = new Channel(POOL_SIZE);$group        = new TaskSet();$workerId     = 0;try {    // 启动收集器协程 —— 合并到达的结果    $collector = spawn(run_collector(...), $resultQueue);    // 启动 worker 池    for ($i = 1; $i <= POOL_SIZE; $i++) {        $group->spawn(run_worker(...), ++$workerId, __DIR__ . '/log_worker.php', $taskQueue, $resultQueue);    }    echo "Start\n";    $startTime    = microtime(true);    // 分割文件，直接把块描述符投喂到 Channel。    // send() 在所有 worker 忙碌时阻塞 → 自然的背压。    // joinAll() 作为取消令牌：如果所有 worker 死亡，停止投喂。    $totalChunks = feed_chunks($logFile, CHUNK_BYTES, $taskQueue, $group);    $taskQueue->close();    printf("\n  Split into %d chunks of ~%d MB each\n", $totalChunks, CHUNK_BYTES / 1024 / 1024);    try {        $group->joinAll()->await();    } catch (\Exception $e) {        echo "Exception: {$e->getMessage()}\n";    }} finally {    $taskQueue->close();    $supervisor->cancel();    $group->seal();    $group->joinAll()->await();    $resultQueue->close();    // 从收集器获取合并后的结果    [$stats, $totalParsed, $totalErrors] = \Async\await($collector);    $elapsed = microtime(true) - $startTime;    print_report($stats, $totalParsed, $totalErrors, $elapsed);}

这样，标准 PHP 函数可以在异步原语旁边获得第二次生命。如果还没有异步编程经验，建议尝试在 XDebug 调试器中运行程序，PHP TrueAsync 附带了适配版本的 XDebug。