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

PHP项目Swoole服务器与协程

wen PHP项目 233

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服务的重要技术栈。

版权声明：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项目WorkerMan与多进程

下一篇PHP项目cURL与HTTP请求

相关文章

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

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