PHP 多进程编程：pcntl 扩展与进程池的实现

PHP 作为一种广泛使用的脚本语言，通常运行在 Web 请求的单进程模型中。然而，在 CLI 环境下处理批量任务、队列消费、数据抓取或高并发场景时，单进程往往成为性能瓶颈。此时，利用 PHP 的 pcntl 扩展进行多进程编程，并构建一个稳定的进程池，可以显著提升任务处理能力。本文将介绍 pcntl 的核心函数，并手把手实现一个简易但实用的进程池。

一、为什么需要多进程？

在 Web 环境中，Nginx + PHP-FPM 已经通过多进程模型处理并发请求。但在命令行下，比如需要处理 10 万条数据同步任务，单进程只能顺序执行，耗时极长。虽然可以使用消息队列加多个消费者进程，但手动管理进程的创建、回收和负载均衡仍然麻烦。多进程编程允许我们在一个 PHP 脚本内 fork 出多个子进程并行工作，充分利用多核 CPU。

二、pcntl 扩展核心函数

pcntl（Process Control）是 PHP 内置的进程控制扩展，默认在编译时需加上 --enable-pcntl，Windows 平台不支持。常用函数如下：

• pcntl_fork()：创建子进程。父进程返回子进程 PID，子进程返回 0，失败返回 -1。
• pcntl_waitpid($pid, &$status, $options)：等待或返回子进程状态。
• pcntl_wait(&$status)：等待任意子进程。
• pcntl_signal($signo, $handler)：安装信号处理器。
• pcntl_signal_dispatch()：调用挂起的信号处理器（PHP 7.1+ 需手动调用或使用 pcntl_async_signals(true)）。
• pcntl_exec()：在子进程中执行外部程序。
• posix_getpid()：获取当前进程 ID。

一个最基础的 fork 示例：

$pid = pcntl_fork();
if ($pid == -1) {
    die('fork failed');
} elseif ($pid == 0) {
    // 子进程逻辑
    echo "child pid: " . posix_getpid() . PHP_EOL;
    exit(0);
} else {
    // 父进程逻辑
    pcntl_waitpid($pid, $status);
    echo "parent: child exited" . PHP_EOL;
}

三、进程池要解决的问题

直接 fork 大量子进程会带来几个问题：

1. 资源耗尽：无限制 fork 可能导致系统内存或 PID 耗尽。
2. 僵尸进程：父进程未回收已结束的子进程，会留下僵尸进程。
3. 负载不均：某些子进程处理慢任务，另一些空闲。
4. 信号处理：需要优雅地响应 SIGTERM、SIGINT 等信号，避免任务中断。

进程池的核心思想是：预先创建固定数量的工作进程（worker），由父进程（master）管理它们，任务通过某种方式分发给空闲 worker。由于 PHP 的 fork 模型下父子进程不共享内存，通常采用以下两种方式分发任务：

• 父进程预分配：master 将任务列表切分，每个 worker 处理固定部分。
• 抢占式：多个 worker 竞争同一个任务源（如 Redis 队列、文件锁）。

本文实现一个基于“父进程预分配 + 动态重启”的通用进程池。

四、实现一个简易进程池

1. 设计目标

• 指定最大工作进程数 $workerNum。
• 每个 worker 从任务数组里取任务执行。
• 父进程监控子进程，异常退出时重新拉起。
• 支持 Ctrl+C 优雅退出。

2. 核心代码

class ProcessPool
{
    private int $workerNum;
    private array $workers = [];
    private array $tasks = [];
    private bool $running = true;

    public function __construct(int $workerNum, array $tasks)
    {
        $this->workerNum = $workerNum;
        $this->tasks = $tasks;
        pcntl_async_signals(true);
        pcntl_signal(SIGINT, [$this, 'shutdown']);
        pcntl_signal(SIGTERM, [$this, 'shutdown']);
    }

    public function shutdown(): void
    {
        $this->running = false;
        foreach ($this->workers as $pid) {
            posix_kill($pid, SIGTERM);
        }
        exit(0);
    }

    public function run(): void
    {
        for ($i = 0; $i < $this->workerNum; $i++) {
            $this->forkWorker();
        }
        // 父进程循环回收
        while ($this->running) {
            $pid = pcntl_wait($status);
            if ($pid > 0) {
                unset($this->workers[$pid]);
                if ($this->running) {
                    $this->forkWorker(); // 补充新 worker
                }
            }
        }
    }

    private function forkWorker(): void
    {
        $pid = pcntl_fork();
        if ($pid == -1) {
            die('fork failed');
        } elseif ($pid == 0) {
            $this->workerLoop();
            exit(0);
        } else {
            $this->workers[$pid] = $pid;
        }
    }

    private function workerLoop(): void
    {
        // 每个 worker 从共享任务中抢占（此处用文件锁模拟）
        while ($this->running) {
            $task = $this->fetchTask();
            if ($task === null) {
                break;
            }
            $this->handleTask($task);
        }
    }

    private function fetchTask(): ?string
    {
        $file = __DIR__ . '/tasks.txt';
        $fp = fopen($file, 'r+');
        if (!$fp) return null;
        if (!flock($fp, LOCK_EX)) {
            fclose($fp);
            return null;
        }
        $task = fgets($fp);
        if ($task === false) {
            flock($fp, LOCK_UN);
            fclose($fp);
            return null;
        }
        // 将已读取的行标记为已处理：简单用空格覆盖
        fseek($fp, -strlen($task), SEEK_CUR);
        fwrite($fp, str_repeat(' ', strlen($task) - 1) . "\n");
        flock($fp, LOCK_UN);
        fclose($fp);
        return trim($task);
    }

    private function handleTask(string $task): void
    {
        // 模拟耗时任务
        usleep(random_int(100000, 500000));
        echo sprintf("[pid %d] handled: %s\n", posix_getpid(), $task);
    }
}

使用示例：

file_put_contents(__DIR__ . '/tasks.txt', implode("\n", range(1, 100)) . "\n");
$pool = new ProcessPool(5, []);
$pool->run();

3. 代码解析

• 信号处理：pcntl_async_signals(true) 让信号自动分发，无需手动调用 pcntl_signal_dispatch()。收到 SIGINT 后，父进程向所有子进程发送 SIGTERM。
• 任务抢占：这里用文件锁 + 文件指针实现了一个极简的“任务队列”。实际生产中可以替换为 Redis 的 LPOP 或数据库行锁。
• 动态补充：父进程在 pcntl_wait 返回后，若池中 worker 数量不足，立即 fork 新的 worker，保证并发度稳定。
• 避免僵尸进程：pcntl_wait 自动回收子进程。

五、进阶优化建议

1. 使用 Redis 队列：将任务放入 Redis List，worker 通过 BLPOP 阻塞获取，避免文件锁竞争。
2. 限制单进程内存：在 workerLoop 中定期检查 memory_get_usage()，超过阈值主动退出，由父进程重新拉起，防止内存泄漏。
3. 超时控制：使用 pcntl_alarm 为每个任务设置超时，超时后杀死子进程。
4. 日志与监控：每个 worker 将处理结果写入独立日志或消息队列，便于追踪。
5. 协程替代：如果 PHP 版本支持 Swoole 或 Swow，可以使用协程获得更高并发，但 pcntl 方案无需额外扩展，适合轻量场景。

六、总结

pcntl 扩展为 PHP 提供了底层的进程控制能力，配合进程池模式，可以稳定地并行处理大量任务。本文实现的进程池虽然简单，但涵盖了 fork、信号处理、子进程回收和任务分发等核心环节。在实际项目中，你可以根据任务类型选择文件、Redis 或数据库作为任务源，并加入超时、内存监控等机制，构建出健壮的多进程处理系统。对于 CPU 密集型或 IO 密集型任务，多进程编程都能让 PHP 在命令行下发挥出远超单进程的吞吐能力。

未经允许不得转载：任鹏个人博客 » PHP 多进程编程：pcntl 扩展与进程池的实现