# 告别阻塞！用 PHP TrueAsync 实现 PHP 脚本提速 10 倍

多年来，开发者们在任务并行化方面有过多种实践。最早的尝试基于 `pcntl_fork` 和 `posix_kill`，但这种方式在 Windows 上无法运行。于是转向 `proc_open` 和管道，但该方案也有其微妙之处，尤其在 Windows 平台上。

核心经验是：永远不要直接使用 `proc_open`，而应该使用 Symfony Process 这类库，它处理了许多边界情况并提供了简洁的进程操作接口。

本文将展示如何用 TrueAsync 构建进程池，在保持易用和可重构的同时，使用普通 PHP 函数实现。

## proc_open

与 `shell_exec` 等函数不同，`proc_open` 是创建进程的丰富工具。PHP 核心甚至为它引入了特殊的"hack"来正确处理管道。管道是进程间通信的最佳方式之一，也是最便捷的方式。唯一更好的方案是共享内存加文件事件，这仅仅是因为内存区域位于操作系统内核之外。

为什么管道方便？因为它们"假装"是文件。可以像普通文件一样用 `fread` 读取、`fwrite` 写入。这对开发者来说非常有价值，因为代码变得更通用、更可移植。

## 通信协议

管道是字节流，它不知道消息的概念，只是简单地在父子进程间传输字节。为了让代码正常工作，需要一种方式来界定消息边界。最简单的格式是 NDJSON：每条消息占一行，行与行之间用 `\n` 分隔。

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

为什么这种格式很合适？因为 PHP 对 JSON 支持很好，而且 PHP 有 `fgets` 函数可以从流中读取行。

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

下面编写普通同步代码，打开进程并通过 NDJSON 协议与之通信：

```php
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],
]);
```

现在把这段代码放到协程里，不是放一个，而是放 10 个。

```php
$tasks = [
    ['id' => 1, 'value' => 42],
    ['id' => 2, 'value' => 17],
    // ... 还有 28 个任务
];

for ($i = 0; $i < 10; $i++) {
    spawn(run_worker(...), 'worker.php', $tasks);
}
```

只写了一行异步代码：`spawn(run_worker(...))`，但实际上 `proc_open`、`fgets` 和 `fclose` 从一开始就变成了非阻塞 I/O 操作。现在十个协程（实际上是 11 个，因为有主代码）竞争 CPU 时间。大部分代码仍保持顺序执行。

## Channels

目前每个协程都收到相同的任务数组，这显然不是期望的行为。例如，可能想从 Twig 模板生成静态站点的 HTML，给定文件列表后让每个文件由独立的 worker 处理；或者可能需要解析日志。

Workers 相互独立运行，因为工作量可能不同，而且不知道 worker 是否忙碌。为此编写任务分发层相当复杂，也不理想。需要一个快速、便捷的解决方案，无需过多思考。

为此，TrueAsync 提供了 Channel，正是为这类任务设计的。Channel 是一个队列，用于在协程间交换数据。Channel 中的每条消息可以是一个任务。协程从同一个 Channel 读取并执行任务。无需检查 worker 是否忙碌，只需把任务发到 Channel，worker 准备好时就会执行。

```php
$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();
```

当然，这段代码没有错误处理或边界检查，但它尽可能简单且易读。然而在底层，一些非平凡的事情正在发生。假设有 100 个任务和 10 个进程的池。当所有 worker 都忙碌时会发生什么？`$channel->send($task)` 会挂起协程，直到队列有空间。队列大小与池大小匹配并非巧合；任务供给函数会暂停，防止内存被尚无法处理的任务填满。

## TaskGroup

已编写的代码离生产就绪还很远。不仅跳过了错误处理，而且没有控制协程。它们悬在空中，如果应用开始关闭或出现问题（异常），无法保证协程不会因逻辑错误而挂起。这很糟糕，而且在生产环境中代码挂起却无从排查原因，非常令人不快。

为最小化意外错误的风险，使用 TaskGroup 模式。它专门设计用于一起启动协程、一起等待它们、一起销毁它们。

```php
$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 完成
}
```

## 增强代码的容错性

前一个版本的代码不适合真实场景。如果 worker 进程崩溃，`fwrite($pipes[0], json_encode($task) . "\n")` 可能失败。

而且代码还隐藏着一个隐患。

```php
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: 0

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

=== DEADLOCK REPORT END ===
```

死锁调试输出可通过 `async.debug_deadlock` ini 指令控制。详见 https://true-async.github.io/en/docs/reference/ini-settings.html

原因是主协程试图向已满的 Channel 发送任务，但没有人会去读取。为避免此错误，需要在逻辑上将 TaskGroup 的生命周期与 `$taskQueue` 绑定。这可以通过 `TaskGroup::all()->await()` 等待所有活跃任务来实现。

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

更简洁的实现方式：

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

`Channel::send` 方法的第二个参数是取消令牌，允许在任意条件下取消操作。`$group->all()` 返回一个 Future，当所有任务完成时解析。

不幸的是，即使 `$taskQueue->send($task, $group->all())` 也不是 100% 正确，因为它没有处理 TaskGroup 中所有协程已完成但新协程尚未启动的情况。Channel 类的语义——在 TrueAsync 0.6.0+ 中与 Go 非常相似——是最有问题的实现之一，值得单独关注。

现在改进进程池代码本身的错误处理。

```php
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;
    }
}
```

在循环前添加超时设置：`stream_set_timeout($pipes[1], 5)`——五秒——防止父进程失控。

故意在 worker_fail.php 中破坏 worker 代码（例如添加 `sleep(10)`）看看会发生什么：

```bash
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 broken

Done.
```

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

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

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

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

```php
$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 逻辑，在进程崩溃时重启：

```php
$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 崩溃，进程池仍在运行：

```bash
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，可能进入无限重启循环。

```php
$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 处理这些块并返回结果。

```php
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 风格"。

```php
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);
    }
}
```

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

```php
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];
}
```

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

```php
$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。

[告别阻塞！用 PHP TrueAsync 实现 PHP 脚本提速 10 倍](https://catchadmin.com/post/2026-03/process-pool-php-trueasync)

多年来，开发者们在任务并行化方面有过多种实践。最早的尝试基于 pcntl_fork 和 posix_kill，但这种方式在 Windows 上无法运行。于是转向 proc_open 和管道，但该方案也有其微妙之处，尤其在 Windows 平台上。

告别阻塞！用 PHP TrueAsync 实现 PHP 脚本提速 10 倍

编程语言

PHPTrueAsync通过进程池和协程技术实现PHP脚本10倍提速，利用proc_open和管道通信处理并行任务。文章详解了NDJSON协议、Channel任务队列、TaskGroup管理等核心技术，提供高效可靠的异步编程方案，特别适合CPU密集型任务处理。

日志文件

调试器

Windows

JSON

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

告别阻塞！用 PHP TrueAsync 实现 PHP 脚本提速 10 倍-腾讯云开发者社区-腾讯云

告别阻塞！用 PHP TrueAsync 实现 PHP 脚本提速 10 倍

告别阻塞！用 PHP TrueAsync 实现 PHP 脚本提速 10 倍

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐