PHP 怎么PHP 协程池

wen PHP项目 2026-07-27 3

本文目录导读：

PHP 怎么PHP 协程池

基于 Swoole 的协程池
基于 PHP 8.1+ Fiber 的协程池
基于第三方库 (如 Hyperf)
最佳实践建议
注意事项

在 PHP 中实现协程池通常是为了限制并发数量、复用协程资源以及控制任务调度，最常用的协程框架是 Swoole 或 Fiber（PHP 8.1+ 原生）。

基于 Swoole 的协程池

Swoole 提供了成熟的协程管理和 Channel 机制,非常适合实现协程池。

基本实现

<?php
use Swoole\Coroutine\Channel;
use Swoole\Coroutine;
use function Swoole\Coroutine\run;
class CoroutinePool
{
    private Channel $pool;
    private int $maxSize;
    public function __construct(int $maxSize = 10)
    {
        $this->maxSize = $maxSize;
        $this->pool = new Channel($maxSize);
        // 预先填充协程资源
        for ($i = 0; $i < $maxSize; $i++) {
            $this->pool->push($this->createCoroutine());
        }
    }
    private function createCoroutine(): \Closure
    {
        return function ($task) {
            // 模拟协程资源（例如数据库连接、HTTP客户端等）
            return $task();
        };
    }
    public function execute(callable $task): mixed
    {
        // 从池中获取一个协程资源
        $coroutine = $this->pool->pop();
        if ($coroutine === false) {
            throw new \RuntimeException('No available coroutine in pool');
        }
        try {
            // 执行任务
            $result = $coroutine($task);
            // 将资源放回池中
            $this->pool->push($coroutine);
            return $result;
        } catch (\Throwable $e) {
            // 如果执行失败，重新创建协程放回池中
            $this->pool->push($this->createCoroutine());
            throw $e;
        }
    }
    public function close(): void
    {
        while (!$this->pool->isEmpty()) {
            $this->pool->pop();
        }
        $this->pool->close();
    }
}
// 使用示例
run(function () {
    $pool = new CoroutinePool(5);
    $tasks = [];
    for ($i = 0; $i < 10; $i++) {
        $tasks[] = Coroutine::create(function () use ($pool, $i) {
            $result = $pool->execute(function () use ($i) {
                Coroutine::sleep(1); // 模拟耗时操作
                return "Task {$i} completed";
            });
            echo $result . PHP_EOL;
        });
    }
    // 等待所有任务完成
    foreach ($tasks as $task) {
        Coroutine::join([$task]);
    }
    $pool->close();
});

带连接复用的协程池

<?php
use Swoole\Coroutine\Channel;
use Swoole\Coroutine;
use Swoole\Database\PDOConfig;
use Swoole\Database\PDOPool;
class DatabasePool
{
    private PDOPool $pool;
    public function __construct(int $maxSize = 10)
    {
        $config = (new PDOConfig())
            ->withHost('127.0.0.1')
            ->withPort(3306)
            ->withDbName('test')
            ->withCharset('utf8mb4')
            ->withUsername('root')
            ->withPassword('password');
        $this->pool = new PDOPool($config, $maxSize);
    }
    public function query(string $sql, array $params = []): array
    {
        $pdo = $this->pool->get();
        try {
            $stmt = $pdo->prepare($sql);
            $stmt->execute($params);
            $result = $stmt->fetchAll(\PDO::FETCH_ASSOC);
            // 归还连接
            $this->pool->put($pdo);
            return $result;
        } catch (\Throwable $e) {
            // 归还连接（即使出错也要归还）
            $this->pool->put($pdo);
            throw $e;
        }
    }
}
// 使用示例
run(function () {
    $pool = new DatabasePool(5);
    $results = [];
    for ($i = 0; $i < 20; $i++) {
        $results[] = Coroutine::create(function () use ($pool, $i) {
            try {
                $data = $pool->query('SELECT ? as value', [$i]);
                echo "Query {$i}: " . json_encode($data) . PHP_EOL;
            } catch (\Throwable $e) {
                echo "Error in query {$i}: " . $e->getMessage() . PHP_EOL;
            }
        });
    }
    foreach ($results as $result) {
        Coroutine::join([$result]);
    }
});

基于 PHP 8.1+ Fiber 的协程池

如果你的 PHP 版本 >= 8.1，可以使用原生 Fiber 实现简单的协程池。

<?php
class FiberPool
{
    private array $fibers = [];
    private \SplQueue $taskQueue;
    private int $maxSize;
    private int $activeCount = 0;
    public function __construct(int $maxSize = 10)
    {
        $this->maxSize = $maxSize;
        $this->taskQueue = new \SplQueue();
    }
    public function execute(callable $task): void
    {
        $this->taskQueue->enqueue($task);
        $this->schedule();
    }
    private function schedule(): void
    {
        while (!$this->taskQueue->isEmpty() && $this->activeCount < $this->maxSize) {
            $task = $this->taskQueue->dequeue();
            $this->runTask($task);
        }
    }
    private function runTask(callable $task): void
    {
        $fiber = new \Fiber(function () use ($task) {
            $this->activeCount++;
            try {
                $task();
            } finally {
                $this->activeCount--;
                $this->schedule();
            }
        });
        $this->fibers[] = $fiber;
        $fiber->start();
    }
    public function run(): void
    {
        while ($this->activeCount > 0 || !$this->taskQueue->isEmpty()) {
            // 检查所有 Fiber 的状态
            foreach ($this->fibers as $key => $fiber) {
                if ($fiber->isTerminated()) {
                    unset($this->fibers[$key]);
                } elseif ($fiber->isSuspended()) {
                    $fiber->resume();
                }
            }
            // 调度新的任务
            $this->schedule();
            // 避免 CPU 空转
            if ($this->activeCount > 0) {
                usleep(10000); // 10ms
            }
        }
    }
}
// 使用示例
$pool = new FiberPool(5);
for ($i = 0; $i < 10; $i++) {
    $pool->execute(function () use ($i) {
        sleep(1); // 模拟耗时操作
        echo "Task {$i} completed\n";
    });
}
$pool->run();

基于第三方库 (如 Hyperf)

Hyperf 框架提供了更完善的协程池实现：

<?php
use Hyperf\Utils\Coroutine;
use Hyperf\Utils\Pool\SimplePool;
use Hyperf\Utils\Pool\PoolFactory;
class ConnectionPool
{
    private SimplePool $pool;
    public function __construct(int $maxSize = 10)
    {
        $factory = new PoolFactory();
        $this->pool = $factory->get('connection', function () {
            return new \PDO('mysql:host=127.0.0.1;dbname=test', 'root', 'password');
        }, [
            'max_connections' => $maxSize
        ]);
    }
    public function execute(callable $callback)
    {
        $connection = $this->pool->get();
        try {
            $result = $callback($connection);
            $this->pool->release($connection);
            return $result;
        } catch (\Throwable $e) {
            $this->pool->release($connection);
            throw $e;
        }
    }
}

最佳实践建议

选择合适的大小：根据系统资源和任务类型调整协程池大小
错误处理：确保异常时资源正确归还
超时处理：为长时间运行的任务设置超时
监控统计：添加活跃数、等待队列长度等监控指标
优雅关闭：在程序退出前正确关闭协程池

注意事项

Swoole 的协程池更适合生产环境
Fiber 协程池需要手动调度，性能不如 Swoole
协程池中的资源（如数据库连接）需要考虑连接有效性
避免协程池中的协程发生死锁

选择哪种实现方式取决于你的具体需求：如果已经在使用 Swoole，直接使用其内置的协程池机制；如果是原生 PHP，可以基于 Fiber 实现简单的协程池；对于企业级应用，推荐使用 Hyperf 等成熟框架。