IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

PHP 怎么PHP 协程池

wen PHP项目 187

本文目录导读：

PHP 怎么PHP 协程池

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

在 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;
        }
    }
}

最佳实践建议

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

注意事项

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

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

版权声明：NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播

上一篇PHP 怎么PHP 资源限制

下一篇PHP 怎么PHP 健康传输

相关文章

抱歉，评论功能暂时关闭!

友情链接攻略窝 · 礼包窝 · 配置窝 · 换算窝 · 算法窝 · 查文窝 · 手册窝 · 问答窝 · 全文搜 · 手册库
相关站点礼包窝 · 查文窝