Swoole 协程框架

概述

Swoole 是一个高性能的 PHP 协程框架，它为 PHP 提供了异步 I/O、协程（Coroutine）、HTTP/WebSocket 服务器、异步 MySQL 客户端、定时器、进程管理等能力。Swoole 通过在 C 层面实现协程调度器，使得 PHP 能够像 Go、Node.js 一样处理高并发 I/O 操作，在单线程中同时处理数万个连接，性能远超传统的 PHP-FPM 模式。

PHP 版本要求

Swoole 5.x 支持 PHP 8.0+。本文基于 Swoole 5.x 和 PHP 8.1+ 编写。Swoole 仅支持 Linux、macOS 和 FreeBSD，不支持 Windows。

基础概念

协程（Coroutine）

协程是一种用户态的轻量级线程，由程序自身控制调度（非操作系统内核调度）。协程在遇到 I/O 操作时自动让出执行权，在 I/O 完成后恢复执行，整个过程无需操作系统参与线程切换，因此开销极小。

协程 vs 线程 vs 进程

特性	进程	线程	协程
创建开销	大	中	极小
调度方式	操作系统	操作系统	用户程序
内存占用	MB 级	KB 级	字节级
切换开销	大	中	极小
并发数量	百级	千级	万级
数据共享	需 IPC	共享内存	直接共享

Swoole 架构

                ┌──────────────────┐
                │   Master Process │  管理进程
                └────────┬─────────┘
                         │
          ┌──────────────┼──────────────┐
          │              │              │
    ┌─────┴─────┐ ┌─────┴─────┐ ┌─────┴─────┐
    │  Worker 1 │ │  Worker 2 │ │  Worker N │  工作进程（协程）
    └───────────┘ └───────────┘ └───────────┘
          │              │              │
          └──────────────┼──────────────┘
                         │
                ┌────────┴─────────┐
                │  Task Worker     │  异步任务进程
                └──────────────────┘

安装与配置

bash

# 通过 PECL 安装
pecl install swoole

# 或使用 composer（Swoole 需要扩展安装）
# composer require swoole/ide-helper:~5.0.0  # IDE 自动补全

# 验证安装
php -m | grep swoole
php --ri swoole

详细说明

协程基础

创建协程

php

<?php
declare(strict_types=1);

use Swoole\Coroutine;

Coroutine\run(function (): void {
    echo "协程 1 开始" . PHP_EOL;

    // 创建子协程
    Coroutine::create(function (): void {
        Coroutine::sleep(1); // 协程 sleep（不阻塞）
        echo "协程 2 完成" . PHP_EOL;
    });

    Coroutine::create(function (): void {
        Coroutine::sleep(0.5);
        echo "协程 3 完成" . PHP_EOL;
    });

    echo "协程 1 结束" . PHP_EOL;
});

// 输出顺序：
// 协程 1 开始
// 协程 1 结束
// 协程 3 完成
// 协程 2 完成

协程调度

php

<?php
declare(strict_types=1);

use Swoole\Coroutine;
use function Swoole\Coroutine\run;

run(function (): void {
    $cid1 = Coroutine::create(function (): void {
        for ($i = 0; $i < 5; $i++) {
            echo "A{$i} ";
            Coroutine::yield(); // 主动让出
        }
    });

    $cid2 = Coroutine::create(function (): void {
        for ($i = 0; $i < 5; $i++) {
            echo "B{$i} ";
            Coroutine::yield();
        }
    });

    echo PHP_EOL;
});
// 输出: A0 B0 A1 B1 A2 B2 A3 B3 A4 B4

HTTP 服务器

php

<?php
declare(strict_types=1);

use Swoole\Http\Server;
use Swoole\Http\Request;
use Swoole\Http\Response;

$server = new Server('0.0.0.0', 9501);

$server->on('request', function (Request $request, Response $response): void {
    $path = $request->server['request_uri'] ?? '/';
    $method = $request->server['request_method'] ?? 'GET';

    $response->header('Content-Type', 'application/json');
    $response->end(json_encode([
        'path' => $path,
        'method' => $method,
        'time' => date('Y-m-d H:i:s'),
    ]));
});

$server->start();

// 启动后: curl http://localhost:9501/api/hello
// {"path":"/api/hello","method":"GET","time":"2024-01-01 12:00:00"}

路由系统

php

<?php
declare(strict_types=1);

use Swoole\Http\Server;
use Swoole\Http\Request;
use Swoole\Http\Response;

$server = new Server('0.0.0.0', 9502);

// 简易路由
$routes = [
    'GET /' => function (Request $req, Response $res): void {
        $res->end('Welcome to Swoole!');
    },
    'GET /api/users' => function (Request $req, Response $res): void {
        $res->header('Content-Type', 'application/json');
        $res->end(json_encode([
            ['id' => 1, 'name' => 'Alice'],
            ['id' => 2, 'name' => 'Bob'],
        ]));
    },
    'GET /api/users/{id}' => function (Request $req, Response $res): void {
        $path = $req->server['request_uri'];
        preg_match('#/api/users/(\d+)#', $path, $matches);
        $id = $matches[1] ?? 'unknown';
        $res->end(json_encode(['id' => (int) $id, 'name' => "User {$id}"]));
    },
];

$server->on('request', function (Request $req, Response $res) use ($routes): void {
    $method = $req->server['request_method'];
    $path = $req->server['request_uri'];
    $key = "{$method} {$path}";

    if (isset($routes[$key])) {
        $routes[$key]($req, $res);
    } else {
        // 尝试匹配带参数的路由
        foreach ($routes as $pattern => $handler) {
            $regex = preg_replace('#\{[^}]+\}#', '([^/]+)', $pattern);
            if (preg_match("~^{$regex}$~", $key)) {
                $handler($req, $res);
                return;
            }
        }
        $res->status(404);
        $res->end('Not Found');
    }
});

$server->start();

WebSocket 服务器

php

<?php
declare(strict_types=1);

use Swoole\WebSocket\Server;
use Swoole\Http\Request;
use Swoole\WebSocket\Frame;

$server = new Server('0.0.0.0', 9503);

$server->on('start', function (Server $server): void {
    echo "WebSocket 服务器启动于 ws://0.0.0.0:9503" . PHP_EOL;
});

$server->on('open', function (Server $server, Request $request): void {
    echo "客户端 #{$request->fd} 已连接" . PHP_EOL;
});

$server->on('message', function (Server $server, Frame $frame): void {
    echo "收到消息: {$frame->data}" . PHP_EOL;

    // 广播给所有客户端
    foreach ($server->connections as $fd) {
        if ($fd !== $server->master_fd) {
            $server->push($fd, "广播: {$frame->data}");
        }
    }
});

$server->on('close', function (Server $server, int $fd): void {
    echo "客户端 #{$fd} 已断开" . PHP_EOL;
});

$server->on('disconnect', function (Server $server, int $fd): void {
    echo "客户端 #{$fd} 异常断开" . PHP_EOL;
});

$server->start();

异步 MySQL 客户端

php

<?php
declare(strict_types=1);

use Swoole\Coroutine;
use function Swoole\Coroutine\run;

run(function (): void {
    // 异步 MySQL 查询
    $mysql = new Swoole\Coroutine\MySQL();

    $connected = $mysql->connect([
        'host' => '127.0.0.1',
        'port' => 3306,
        'user' => 'root',
        'password' => '',
        'database' => 'test',
    ]);

    if (!$connected) {
        echo "连接失败: " . $mysql->error . PHP_EOL;
        return;
    }

    // 并发执行多个查询
    $results = [];
    $queries = [
        'SELECT * FROM users LIMIT 10',
        'SELECT COUNT(*) as total FROM orders',
        'SELECT * FROM products WHERE price > 100 LIMIT 5',
    ];

    foreach ($queries as $sql) {
        Coroutine::create(function () use ($mysql, $sql, &$results): void {
            $stmt = $mysql->query($sql);
            $results[] = $stmt ? $stmt->fetchAll() : [];
        });
    }

    // 等待所有协程完成
    Coroutine::sleep(1);

    echo "查询结果数: " . count($results) . PHP_EOL;
    print_r($results);

    $mysql->close();
});

定时器

php

<?php
declare(strict_types=1);

use Swoole\Timer;
use function Swoole\Coroutine\run;

run(function (): void {
    // 一次性定时器
    Timer::after(3000, function (): void {
        echo "3 秒后执行一次性定时器" . PHP_EOL;
    });

    // 循环定时器（每秒执行）
    $timerId = Timer::tick(1000, function (): void {
        echo "Tick: " . date('H:i:s') . PHP_EOL;
    });

    // 10 秒后取消循环定时器
    Timer::after(10000, function () use ($timerId): void {
        echo "取消定时器 #{$timerId}" . PHP_EOL;
        Timer::clear($timerId);
    });

    // 保持协程运行
    Coroutine::sleep(15);
});

进程管理

php

<?php
declare(strict_types=1);

use Swoole\Process;
use Swoole\Process\Pool;

/**
 * 子进程工作函数
 */
function workerProcess(int $workerId): void
{
    echo "Worker #{$workerId} 启动 (PID: " . getmypid() . ")" . PHP_EOL;

    while (true) {
        $data = "Worker #{$workerId} 数据: " . date('H:i:s');
        echo $data . PHP_EOL;
        sleep(mt_rand(1, 3));
    }
}

// 创建进程池
$pool = new Pool(4); // 4 个 worker 进程

$pool->on('WorkerStart', function (Pool $pool, int $workerId): void {
    echo "Worker #{$workerId} 启动" . PHP_EOL;
});

$pool->on('WorkerStop', function (Pool $pool, int $workerId): void {
    echo "Worker #{$workerId} 停止" . PHP_EOL;
});

$pool->on('Message', function (Pool $pool, string $data): void {
    echo "收到消息: {$data}" . PHP_EOL;
});

$pool->start();

实战示例

协程并发 HTTP 客户端

php

<?php
declare(strict_types=1);

use Swoole\Coroutine;
use Swoole\Coroutine\Http\Client;
use function Swoole\Coroutine\run;

/**
 * 协程并发请求
 */
function concurrentRequests(): array
{
    $urls = [
        'https://httpbin.org/get?req=1',
        'https://httpbin.org/get?req=2',
        'https://httpbin.org/get?req=3',
        'https://httpbin.org/get?req=4',
        'https://httpbin.org/get?req=5',
    ];

    $results = [];
    $start = microtime(true);

    foreach ($urls as $i => $url) {
        Coroutine::create(function () use ($url, $i, &$results): void {
            $client = new Client('httpbin.org', 443);
            $client->setHeaders(['User-Agent' => 'Swoole']);
            $client->get('/get?req=' . ($i + 1));

            $results[$i] = [
                'url' => $url,
                'status' => $client->statusCode,
                'time_ms' => round($client->totalTimeMs, 2),
            ];
        });
    }

    // 等待所有请求完成
    Coroutine::sleep(3);

    $totalTime = round((microtime(true) - $start) * 1000, 2);
    echo "总耗时: {$totalTime}ms (5 个并发请求)" . PHP_EOL;
    return $results;
}

run(function (): void {
    $results = concurrentRequests();
    foreach ($results as $i => $result) {
        echo "请求 {$i}: status={$result['status']} time={$result['time_ms']}ms" . PHP_EOL;
    }
});

生产级 HTTP 服务器

php

<?php
declare(strict_types=1);

use Swoole\Http\Server;
use Swoole\Http\Request;
use Swoole\Http\Response;

$server = new Server('0.0.0.0', 9501, SWOOLE_PROCESS);

// 配置
$server->set([
    'worker_num' => 4,           // worker 进程数
    'daemonize' => false,        // 守护进程模式
    'max_request' => 10000,      // 每个 worker 最大请求数（防止内存泄漏）
    'dispatch_mode' => 2,         // 固定模式分配请求
    'reload_async' => true,      // 异步重启
    'log_file' => '/tmp/swoole.log',
    'upload_max_filesize' => 10 * 1024 * 1024,
]);

// 静态文件服务
$server->on('request', function (Request $req, Response $res): void {
    $path = $req->server['request_uri'];

    if ($path === '/health') {
        $res->end('OK');
        return;
    }

    $res->header('Content-Type', 'application/json');
    $res->end(json_encode([
        'code' => 0,
        'message' => 'success',
        'data' => ['path' => $path],
    ], JSON_UNESCAPED_UNICODE));
});

$server->on('start', function (Server $server): void {
    echo "Swoole HTTP 服务器启动" . PHP_EOL;
    echo "监听: http://0.0.0.0:9501" . PHP_EOL;
});

$server->on('WorkerStart', function (Server $server, int $workerId): void {
    echo "Worker #{$workerId} 启动" . PHP_EOL;
});

$server->on('WorkerStop', function (Server $server, int $workerId): void {
    echo "Worker #{$workerId} 停止" . PHP_EOL;
});

$server->start();

注意事项

协程安全

协程环境下不可用的函数

以下函数在协程环境中不能正常工作：

sleep() / usleep() → 使用 Coroutine::sleep()
file_get_contents() → 使用 Swoole\Coroutine\System::readFile()
stream_socket_client() → 使用 Swoole\Coroutine\Client
PDO/MySQLi → 使用 Swoole\Coroutine\MySQL
curl_* → 使用 Swoole\Coroutine\Http\Client

静态变量问题

php

<?php
declare(strict_types=1);

use Swoole\Coroutine;

// 协程间共享静态变量 - 需要特别注意
function counter(): int
{
    static $count = 0;
    return ++$count;
}

Coroutine\run(function (): void {
    Coroutine::create(function (): void {
        for ($i = 0; $i < 5; $i++) {
            echo "协程1: " . counter() . PHP_EOL;
            Coroutine::yield();
        }
    });

    Coroutine::create(function (): void {
        for ($i = 0; $i < 5; $i++) {
            echo "协程2: " . counter() . PHP_EOL;
            Coroutine::yield();
        }
    });
});

最佳实践

1. 合理设置 Worker 数量

php

$server->set([
    'worker_num' => swoole_cpu_num(), // 通常等于 CPU 核心数
    // I/O 密集型可以适当增加
    'worker_num' => swoole_cpu_num() * 2,
]);

2. 防止内存泄漏

php

$server->set([
    'max_request' => 5000, // 每个 worker 处理 5000 个请求后重启
    'max_request_grace' => 100, // 到达上限后平滑过渡
]);

3. 优雅重启

bash

# 发送 USR2 信号进行平滑重启
kill -USR1 $(cat /var/run/swoole.pid)

# 或使用 swoole CLI
php server.php reload

继续学习：ReactPHP 事件驱动

Swoole 协程框架 ​

概述 ​

基础概念 ​

协程（Coroutine） ​

协程 vs 线程 vs 进程 ​

Swoole 架构 ​

安装与配置 ​

详细说明 ​

协程基础 ​

创建协程 ​

协程调度 ​

HTTP 服务器 ​

路由系统 ​

WebSocket 服务器 ​

异步 MySQL 客户端 ​

定时器 ​

进程管理 ​

实战示例 ​

协程并发 HTTP 客户端 ​

生产级 HTTP 服务器 ​

注意事项 ​

协程安全 ​

静态变量问题 ​

最佳实践 ​

1. 合理设置 Worker 数量 ​

2. 防止内存泄漏 ​

3. 优雅重启 ​

参考链接 ​