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

PHP项目Swoole服务器与协程

wen PHP项目 2

PHP Swoole服务器与协程详解

Swoole基础概念

Swoole是一个面向生产环境的PHP异步网络通信引擎,使PHP开发者可以编写高性能的异步并发TCP/UDP/HTTP/WebSocket服务。

PHP项目Swoole服务器与协程

核心特性

服务器类型

// HTTP服务器
$http = new Swoole\Http\Server("0.0.0.0", 9501);
$http->on('request', function ($request, $response) {
    $response->end("<h1>Hello Swoole</h1>");
});
$http->start();
// WebSocket服务器
$ws = new Swoole\WebSocket\Server("0.0.0.0", 9502);
$ws->on('open', function ($server, $request) {
    echo "连接建立: {$request->fd}\n";
});
$ws->on('message', function ($server, $frame) {
    $server->push($frame->fd, "收到消息: {$frame->data}");
});
$ws->start();
// TCP服务器
$tcp = new Swoole\Server("0.0.0.0", 9503);
$tcp->on('receive', function ($server, $fd, $reactor_id, $data) {
    $server->send($fd, "收到数据: {$data}");
});
$tcp->start();

进程模型

$server = new Swoole\Server("0.0.0.0", 9501);
// 配置参数
$server->set([
    'worker_num' => 8,        // Worker进程数
    'max_request' => 10000,   // 每个进程最大请求数
    'daemonize' => true,      // 守护进程化
    'log_file' => '/var/log/swoole.log',
    'pid_file' => '/var/run/swoole.pid'
]);
// 事件回调
$server->on('WorkerStart', function ($server, $worker_id) {
    // 每个Worker进程启动时执行
    echo "Worker #{$worker_id} 启动\n";
});

协程（Coroutine）深入

协程基础

// 创建协程
go(function () {
    // 协程1
    echo "协程1开始\n";
    Co::sleep(1);  // 协程睡眠，不阻塞进程
    echo "协程1结束\n";
});
go(function () {
    // 协程2
    echo "协程2开始\n";
    Co::sleep(0.5);
    echo "协程2结束\n";
});

协程API

// 协程通道（Channel）
$chan = new Swoole\Coroutine\Channel();
go(function () use ($chan) {
    // 生产者
    for ($i = 0; $i < 10; $i++) {
        $chan->push("数据{$i}");
        Co::sleep(0.1);
    }
    $chan->close();
});
go(function () use ($chan) {
    // 消费者
    while ($data = $chan->pop()) {
        echo "消费: {$data}\n";
    }
});
// 协程WaitGroup
$wg = new Swoole\Coroutine\WaitGroup();
for ($i = 0; $i < 5; $i++) {
    $wg->add();
    go(function () use ($wg, $i) {
        defer(function () use ($wg) {
            $wg->done();
        });
        Co::sleep(0.1 * $i);
        echo "任务{$i}完成\n";
    });
}
$wg->wait(); // 等待所有任务完成
echo "所有任务完成\n";

协程MySQL

use Swoole\Coroutine\MySQL;
go(function () {
    $mysql = new MySQL();
    $mysql->connect([
        'host' => '127.0.0.1',
        'port' => 3306,
        'user' => 'root',
        'password' => 'password',
        'database' => 'test',
    ]);
    // 查询操作（非阻塞）
    $result = $mysql->query('SELECT * FROM users WHERE id = ?', [1]);
    var_dump($result);
    // 事务
    $mysql->begin();
    try {
        $mysql->query('INSERT INTO users (name) VALUES (?)', ['张三']);
        $mysql->query('UPDATE account SET balance = balance - 100 WHERE user_id = ?', [1]);
        $mysql->commit();
    } catch (Exception $e) {
        $mysql->rollback();
    }
});

协程Redis

use Swoole\Coroutine\Redis;
go(function () {
    $redis = new Redis();
    $redis->connect('127.0.0.1', 6379);
    // 基本操作（非阻塞）
    $redis->set('key', 'value');
    $value = $redis->get('key');
    // Pipeline
    $redis->multi(Redis::PIPELINE);
    $redis->set('key1', 'val1');
    $redis->get('key2');
    $redis->del('key3');
    $result = $redis->exec();
    // 订阅发布
    go(function () use ($redis) {
        $redis->subscribe(['channel'], function ($redis, $channel, $message) {
            echo "收到消息: {$message}\n";
        });
    });
});

协程HTTP客户端

use Swoole\Coroutine\Http\Client;
go(function () {
    $cli = new Client('api.example.com', 80);
    $cli->setHeaders([
        'Host' => 'api.example.com',
        'User-Agent' => 'Swoole Coroutine Client',
    ]);
    // 并发请求
    $requests = [];
    $cli1 = new Client('httpbin.org', 80);
    $cli1->get('/get');
    $requests[] = $cli1;
    $cli2 = new Client('httpbin.org', 80);  
    $cli2->post('/post', ['name' => 'test']);
    $requests[] = $cli2;
    // 并发执行
    $results = Swoole\Coroutine::join($requests);
    foreach ($results as $i => $result) {
        echo "请求{$i}: {$result->statusCode}\n";
        echo $result->body . "\n";
    }
});

高级特性应用

协程池

class CoroutinePool
{
    private $pool;
    private $maxSize;
    public function __construct(int $maxSize = 10)
    {
        $this->maxSize = $maxSize;
        $this->pool = new Swoole\Coroutine\Channel($maxSize);
    }
    public function init(callable $factory)
    {
        for ($i = 0; $i < $this->maxSize; $i++) {
            $this->pool->push($factory($i));
        }
    }
    public function get()
    {
        return $this->pool->pop();
    }
    public function put($connection)
    {
        $this->pool->push($connection);
    }
    public function close()
    {
        $this->pool->close();
    }
}
// 使用示例
$pool = new CoroutinePool(5);
$pool->init(function ($id) {
    $mysql = new Swoole\Coroutine\MySQL();
    $mysql->connect(['host' => '127.0.0.1', 'port' => 3306, ...]);
    return $mysql;
});
go(function () use ($pool) {
    $mysql = $pool->get();
    $result = $mysql->query('SELECT * FROM users');
    $pool->put($mysql);
});

协程调度器

class Scheduler
{
    private $tasks = [];
    private $maxConcurrent;
    private $running = 0;
    public function __construct(int $maxConcurrent = 10)
    {
        $this->maxConcurrent = $maxConcurrent;
    }
    public function add(callable $task)
    {
        $this->tasks[] = $task;
        $this->schedule();
    }
    private function schedule()
    {
        while ($this->tasks && $this->running < $this->maxConcurrent) {
            $task = array_shift($this->tasks);
            $this->running++;
            go(function () use ($task) {
                try {
                    $task();
                } finally {
                    $this->running--;
                    $this->schedule();
                }
            });
        }
    }
}
// 使用
$scheduler = new Scheduler(3);
foreach ($tasks as $task) {
    $scheduler->add($task);
}

性能优化

配置优化

$server = new Swoole\Server("0.0.0.0", 9501);
$server->set([
    'worker_num' => swoole_cpu_num() * 2,  // CPU核心数的2倍
    'max_coroutine' => 30000,              // 最大协程数
    'reactor_num' => swoole_cpu_num(),     // Reactor线程数
    'backlog' => 500,                      // 连接队列大小
    'heartbeat_check_interval' => 60,      // 心跳检测间隔
    'heartbeat_idle_time' => 600,          // 最大空闲时间
    'open_tcp_nodelay' => true,            // 禁用Nagle算法
    'reload_async' => true,                // 异步重启
]);

协程安全

// 使用协程安全的数据结构
$array = new Swoole\Coroutine\Channel(100);
go(function () use ($array) {
    $array->push(['data' => 1]);
});
go(function () use ($array) {
    $data = $array->pop();
});
// 使用协程锁
$lock = new Swoole\Coroutine\Lock();
go(function () use ($lock) {
    $lock->lock();
    try {
        // 临界区代码
        $sharedResource->write($data);
    } finally {
        $lock->unlock();
    }
});
// 避免全局变量污染
class SafeCounter {
    private static $counter = 0;
    private static $lock;
    public static function increment() {
        self::$lock->lock();
        self::$counter++;
        self::$lock->unlock();
    }
}

监控与调试

// 协程状态监控
$stats = Swoole\Coroutine::stats();
echo "当前协程数: {$stats['coroutine_num']}\n";
echo "协程峰值: {$stats['coroutine_peak_num']}\n";
// 性能分析
Swoole\Coroutine::set([
    'enable_deadlock_check' => true,
    'enable_coroutine' => true,
]);
// 日志追踪
Swoole\Coroutine::defer(function () {
    // 协程结束时自动执行
    flushLog();
});

建议与最佳实践

  1. 合理使用协程：不是所有场景都适合协程，I/O密集型任务最适合
  2. 避免协程泄露：确保协程能正确结束
  3. 控制并发数：使用Channel或信号量控制并发
  4. 错误处理：使用try-catch捕获协程异常
  5. 资源管理：使用try-finally确保资源释放
  6. 测试驱动：编写协程测试用例
  7. 渐进式采用：从单个服务开始逐步引入Swoole

Swoole + 协程的组合让PHP能够处理高并发、I/O密集型的应用场景，是构建高性能PHP服务的重要技术栈。

上一篇PHP项目WorkerMan与多进程

下一篇PHP项目cURL与HTTP请求

相关文章

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