PHP 怎么PHP 线程池

wen PHP项目 2026-07-27 179

本文目录导读：

PHP 怎么PHP 线程池

使用 ext-parallel（PHP 8.0+ 推荐）
使用 Swoole（推荐生产环境）
使用进程模拟线程池（兼容性好）
注意事项
推荐方案

在 PHP 中实现线程池，最常用的扩展是 pthreads，不过需要注意：

pthreads 只能在 CLI 模式下使用，不能用于 Web 服务器环境（如 Apache/FPM）。
PHP 7.4+ 对 pthreads 支持有限，推荐使用 ext-parallel 或 Swoole。
PHP 8.0+ 已不再支持 pthreads（需要自行编译），官方推荐使用 ext-parallel。

以下是几种线程池实现方式：

使用 ext-parallel（PHP 8.0+ 推荐）

parallel 是一个更现代的并行扩展，提供了线程池功能。

安装

pecl install parallel

示例代码

<?php
use parallel\{Runtime, Future, Channel, Events};
class ThreadPool {
    private array $workers = [];
    private Channel $channel;
    private int $size;
    public function __construct(int $size) {
        $this->size = $size;
        $this->channel = Channel::make('pool');
        // 创建工作线程
        for ($i = 0; $i < $size; $i++) {
            $this->workers[$i] = new Runtime();
            $this->workers[$i]->run(function(int $id, Channel $channel) {
                while ($task = $channel->recv()) {
                    [$func, $args] = $task;
                    try {
                        $result = $func(...$args);
                        echo "[Worker $id] Task completed\n";
                    } catch (\Throwable $e) {
                        echo "[Worker $id] Error: " . $e->getMessage() . "\n";
                    }
                }
            }, [$i, $this->channel]);
        }
    }
    public function submit(callable $func, array $args = []): void {
        $this->channel->send([$func, $args]);
    }
    public function shutdown(): void {
        // 发送结束信号
        for ($i = 0; $i < $this->size; $i++) {
            $this->channel->send(null);  // 空值表示结束
        }
    }
}
// 使用示例
$pool = new ThreadPool(4);
// 提交任务
for ($i = 0; $i < 10; $i++) {
    $pool->submit(function($n) {
        sleep(1);
        return $n * $n;
    }, [$i]);
}
$pool->shutdown();
echo "All tasks submitted\n";

使用 Swoole（推荐生产环境）

Swoole 提供了更完善的进程/协程池功能。

安装

pecl install swoole

进程池示例

<?php
use Swoole\Process;
use Swoole\Process\Pool;
$pool = new Pool(4);
$pool->on("WorkerStart", function (Pool $pool, int $workerId) {
    echo "Worker#{$workerId} is started\n";
    // 接收消息
    $pool->read(function ($data) use ($workerId) {
        echo "Worker#{$workerId} received: $data\n";
        // 处理任务
        return "Processed by Worker#$workerId";
    });
});
$pool->start();

协程方式（更推荐）

<?php
use Swoole\Coroutine;
use Swoole\Coroutine\Channel;
$poolSize = 4;
$channel = new Channel($poolSize);
$results = [];
// 创建协程池
for ($i = 0; $i < $poolSize; $i++) {
    Coroutine::create(function () use ($channel, $i) {
        while (true) {
            $task = $channel->pop();
            if ($task === null) break;  // 结束信号
            [$id, $data] = $task;
            echo "Worker $i processing task $id\n";
            Coroutine::sleep(1);  // 模拟耗时操作
            $results[$id] = "Result of task $id";
        }
    });
}
// 提交任务
for ($i = 0; $i < 10; $i++) {
    $channel->push([$i, "Data $i"]);
}
// 关闭通道
$channel->close();
echo "All tasks completed\n";

使用进程模拟线程池（兼容性好）

如果无法安装扩展,可以使用 pcntl_fork() 实现进程池。

<?php
class ProcessPool {
    private array $children = [];
    private int $maxWorkers;
    private array $tasks = [];
    public function __construct(int $maxWorkers = 4) {
        $this->maxWorkers = $maxWorkers;
    }
    public function addTask(callable $task): void {
        $this->tasks[] = $task;
    }
    public function execute(): void {
        $taskQueue = $this->tasks;
        $running = 0;
        while (!empty($taskQueue) || $running > 0) {
            // 清理已完成进程
            foreach ($this->children as $key => $child) {
                $status = null;
                $pid = pcntl_waitpid($child, $status, WNOHANG);
                if ($pid > 0) {
                    unset($this->children[$key]);
                    $running--;
                }
            }
            // 启动新进程
            while ($running < $this->maxWorkers && !empty($taskQueue)) {
                $task = array_shift($taskQueue);
                $pid = pcntl_fork();
                if ($pid == -1) {
                    die("Fork failed");
                } elseif ($pid == 0) {
                    // 子进程
                    $task();
                    exit(0);
                } else {
                    // 父进程
                    $this->children[] = $pid;
                    $running++;
                }
            }
            // 避免空转
            if ($running >= $this->maxWorkers) {
                usleep(100000);  // 100ms
            }
        }
        // 等待所有子进程结束
        foreach ($this->children as $child) {
            pcntl_waitpid($child, $status);
        }
    }
}
// 使用示例
$pool = new ProcessPool(3);
for ($i = 0; $i < 10; $i++) {
    $pool->addTask(function() use ($i) {
        echo "Task $i started\n";
        sleep(1);
        echo "Task $i completed\n";
    });
}
$pool->execute();

注意事项

1 共享数据问题

线程/进程之间不能直接共享变量
需要使用 Channel、队列 或 共享内存 进行通信

2 资源限制

注意 PHP 的 max_execution_time 限制
控制最大并发数,避免系统过载

3 错误处理

// 使用 try-catch 捕获线程内的异常
$pool->submit(function() {
    try {
        // 任务逻辑
    } catch (\Throwable $e) {
        // 记录错误
        return ['error' => $e->getMessage()];
    }
});