PHP WebSocket 实时通信实战：从 Ratchet 到 Laravel Reverb 的完整实现指南

在现代 Web 应用中，实时通信已经成为许多产品的核心需求——在线客服、协作编辑、实时通知、股票行情推送、多人游戏等场景都离不开 WebSocket 技术。传统的 HTTP 轮询方式效率低下且延迟高，而 WebSocket 协议通过一条持久化的全双工连接解决了这些问题。本文将系统讲解在 PHP 生态中实现 WebSocket 通信的完整方案，从最底层的 Ratchet 到现代框架 Laravel Reverb，再到 Swoole 高性能方案，覆盖架构设计、代码实现、认证授权和部署运维的每一个关键环节。

一、WebSocket 协议基础与 PHP 的适配挑战

1.1 为什么是 WebSocket 而不是轮询

HTTP 协议是请求-响应模型，客户端必须主动发起请求才能获取数据。在实时场景下，常见的 Hack 方案是长轮询（Long Polling）和服务器发送事件（SSE），但它们都存在本质缺陷：

• 长轮询：服务器 hold 住请求直到有新数据才返回，每次返回后客户端需要立即重新建立连接，产生大量空请求和握手开销。
• SSE：单向通信，只能服务器推送到客户端，客户端发送数据仍需 HTTP 请求，无法实现真正的双向交互。
• WebSocket：一次 HTTP Upgrade 握手后升级为全双工 TCP 通道，双方可随时发送数据帧，开销极低，延迟通常在毫秒级。

1.2 PHP 的传统模型与 WebSocket 的冲突

PHP 的传统运行模式是请求驱动的：每个 HTTP 请求由一个 PHP 进程处理，处理完即释放。这种”即来即走”的模型与 WebSocket 的长连接常驻需求天然矛盾。要让 PHP 处理 WebSocket，本质上需要让 PHP 进程常驻内存，持续监听连接。这催生了三种主要方案：

二、Ratchet 实战：从零搭建 WebSocket 服务

2.1 项目初始化

Ratchet 是 PHP WebSocket 领域最经典的库，基于 ReactPHP 事件循环实现。下面我们从零搭建一个完整的聊天室服务。

1
2
3
4
5
6
7
8
9
10
11
12
mkdir php-websocket-demo && cd php-websocket-demo

composer require cboden/ratchet



# 项目结构

# ├── composer.json

# ├── src/

# │   ├── Chat.php          # 消息处理逻辑

# │   └── AuthMiddleware.php # 认证中间件

# ├── bin/

# │   └── server.php         # 启动脚本

# └── public/

#     └── index.html         # 前端页面

2.2 核心消息处理类

Ratchet 的核心接口是

，我们需要实现四个方法：

、

、

和

。下面是一个功能完整的聊天室实现：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
<?php

// src/Chat.php



namespace App;



use Ratchet\MessageComponentInterface;

use Ratchet\ConnectionInterface;



class Chat implements MessageComponentInterface

{

    /** @var \SplObjectStorage<ConnectionInterface> 已连接的客户端 */

    private \SplObjectStorage $clients;



    /** @var array<int, array> 在线用户信息 */

    private array $users = [];



    /** @var array<string, string> 房间ID => 房间名 */

    private array $rooms = [

        'general'  => '综合大厅',

        'tech'     => '技术交流',

        'random'   => '闲聊水区',

    ];



    public function __construct()

    {

        $this->clients = new \SplObjectStorage();

    }



    public function onOpen(ConnectionInterface $conn): void

    {

        $this->clients->attach($conn);

        $conn->resourceId = $conn->resourceId;



        echo "新连接: #{$conn->resourceId}

";

    }



    public function onMessage(ConnectionInterface $from, $msg): void

    {

        $data = json_decode($msg, true);



        if (json_last_error() !== JSON_ERROR_NONE) {

            $from->send(json_encode([

                'type'  => 'error',

                'msg'   => '无效的 JSON 格式',

            ]));

            return;

        }



        $type = $data['type'] ?? '';



        switch ($type) {

            case 'join':

                $this->handleJoin($from, $data);

                break;

            case 'message':

                $this->handleMessage($from, $data);

                break;

            case 'typing':

                $this->handleTyping($from, $data);

                break;

            default:

                $from->send(json_encode([

                    'type' => 'error',

                    'msg'  => "未知消息类型: {$type}",

                ]));

        }

    }



    private function handleJoin(ConnectionInterface $conn, array $data): void

    {

        $username = trim($data['username'] ?? '');

        $room     = $data['room'] ?? 'general';



        if ($username === '') {

            $conn->send(json_encode([

                'type' => 'error',

                'msg'  => '用户名不能为空',

            ]));

            return;

        }



        if (!isset($this->rooms[$room])) {

            $conn->send(json_encode([

                'type' => 'error',

                'msg'  => "房间 {$room} 不存在",

            ]));

            return;

        }



        // 记录用户信息

        $this->users[$conn->resourceId] = [

            'username' => $username,

            'room'     => $room,

            'joined_at' => time(),

        ];



        // 发送欢迎消息

        $conn->send(json_encode([

            'type'    => 'joined',

            'room'    => $room,

            'msg'     => "欢迎 {$username} 加入 {$this->rooms[$room]}",

            'rooms'   => array_keys($this->rooms),

        ]));



        // 广播给房间内其他用户

        $this->broadcastToRoom($room, [

            'type' => 'system',

            'msg'  => "{$username} 加入了房间",

            'timestamp' => date('Y-m-d H:i:s'),

        ], $conn->resourceId);

    }



    private function handleMessage(ConnectionInterface $from, array $data): void

    {

        $user = $this->users[$from->resourceId] ?? null;

        if ($user === null) {

            $from->send(json_encode([

                'type' => 'error',

                'msg'  => '请先加入房间',

            ]));

            return;

        }



        $text = trim($data['text'] ?? '');

        if ($text === '') {

            return;

        }



        // 简单的 XSS 过滤

        $text = htmlspecialchars($text, ENT_QUOTES, 'UTF-8');

        // 限制消息长度

        $text = mb_substr($text, 0, 500);



        $this->broadcastToRoom($user['room'], [

            'type'      => 'message',

            'username'  => $user['username'],

            'text'      => $text,

            'timestamp' => date('Y-m-d H:i:s'),

        ]);

    }



    private function handleTyping(ConnectionInterface $from, array $data): void

    {

        $user = $this->users[$from->resourceId] ?? null;

        if ($user === null) return;



        $this->broadcastToRoom($user['room'], [

            'type'     => 'typing',

            'username' => $user['username'],

            'isTyping' => (bool)($data['isTyping'] ?? false),

        ], $from->resourceId);

    }



    /**

     * 向指定房间的所有用户广播消息

     */

    private function broadcastToRoom(string $room, array $message, ?int $excludeId = null): void

    {

        $payload = json_encode($message, JSON_UNESCAPED_UNICODE);



        foreach ($this->clients as $client) {

            $clientId = $client->resourceId;

            if ($excludeId !== null && $clientId === $excludeId) {

                continue;

            }

            $user = $this->users[$clientId] ?? null;

            if ($user !== null && $user['room'] === $room) {

                $client->send($payload);

            }

        }

    }



    public function onClose(ConnectionInterface $conn): void

    {

        $user = $this->users[$conn->resourceId] ?? null;

        $username = $user['username'] ?? '未知用户';

        $room = $user['room'] ?? null;



        $this->clients->detach($conn);

        unset($this->users[$conn->resourceId]);



        if ($room !== null) {

            $this->broadcastToRoom($room, [

                'type' => 'system',

                'msg'  => "{$username} 离开了房间",

            ]);

        }



        echo "连接关闭: #{$conn->resourceId} ({$username})

";

    }



    public function onError(ConnectionInterface $conn, \Exception $e): void

    {

        echo "错误: {$e->getMessage()}

";

        $conn->close();

    }

}

2.3 服务启动脚本

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
<?php

// bin/server.php



use Ratchet\Server\IoServer;

use Ratchet\Http\HttpServer;

use Ratchet\WebSocket\WsServer;

use App\Chat;



require __DIR__ . '/../vendor/autoload.php';



$port = 8080;



$server = IoServer::factory(

    new HttpServer(

        new WsServer(

            new Chat()

        )

    ),

    $port

);



echo "WebSocket 服务器启动在 0.0.0.0:{$port}

";

echo "按 Ctrl+C 停止

";



$server->run();

启动服务：

，一个功能完整的聊天室后端就运行起来了。前端通过 JavaScript 的

API 即可连接：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
// 前端连接示例

const ws = new WebSocket('ws://localhost:8080');



ws.onopen = () => {

    ws.send(JSON.stringify({

        type: 'join',

        username: '张三',

        room: 'general'

    }));

};



ws.onmessage = (event) => {

    const data = JSON.parse(event.data);

    if (data.type === 'message') {

        appendMessage(data.username, data.text);

    } else if (data.type === 'system') {

        appendSystem(data.msg);

    }

};



// 发送消息

function sendMessage(text) {

    ws.send(JSON.stringify({

        type: 'message',

        text: text

    }));

}

三、认证与安全：WebSocket 连接的身份验证

裸 WebSocket 连接默认不携带 HTTP Cookie，传统的 Session 认证无法直接使用。在 WebSocket 场景下，认证通常有三种方案：

3.1 查询参数 Token 方案（最简单）

1
2
3
// 前端：连接时附带 token

const token = localStorage.getItem('auth_token');

const ws = new WebSocket(`ws://localhost:8080?token=${token}`);

服务端在

时解析 query string：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
public function onOpen(ConnectionInterface $conn): void

{

    $query = $conn->httpRequest->getUri()->getQuery();

    parse_str($query, $params);

    $token = $params['token'] ?? '';



    $userId = $this->verifyToken($token);

    if ($userId === null) {

        $conn->close();

        return;

    }



    $conn->userId = $userId;

    $this->clients->attach($conn);

}



private function verifyToken(string $token): ?int

{

    // 查 Redis 或数据库验证 JWT / API Token

    $payload = json_decode(

        base64_decode(str_replace('_', '/', $token)),

        true

    );

    if (!isset($payload['user_id']) || $payload['exp'] < time()) {

        return null;

    }

    return $payload['user_id'];

}

3.2 HTTP Header Token 方案（更安全）

浏览器原生 WebSocket API 不支持自定义 Header，但可以使用

子协议头来传递 token：

1
2
// 前端

const ws = new WebSocket('ws://localhost:8080', ['token', authToken]);

1
2
3
4
5
6
7
8
9
10
// 服务端在 onOpen 前通过中间件拦截

use Ratchet\RFC6455\Messaging\MessageInterface;



$server = new WsServer($chat);



// 自定义握手，检查子协议

$server->enableKeepAlive = true;

$server->setEnableKeepAlive(false);



// 实际上需要在 HttpServer 层拦截，更推荐使用 Ratchet 的 HttpServer 配合自定义 HttpHandler

3.3 使用 Ratchet 的 WAMP 协议实现更安全的认证

WAMP（WebSocket Application Messaging Protocol）在握手阶段就走标准 HTTP 头，可以直接利用已有的 Cookie 和 Session 机制。对于使用 Symfony 的项目，可以结合

中间件将 HTTP Session 注入到 WebSocket 连接中：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
use Ratchet\Session\SessionProvider;

use Symfony\Component\HttpFoundation\Session\Storage\Handler\PdoSessionHandler;



$pdo = new PDO('mysql:host=127.0.0.1;dbname=app', 'user', 'pass');



$sessionHandler = new PdoSessionHandler($pdo, [

    'db_table'    => 'sessions',

    'db_id_col'   => 'sess_id',

    'db_data_col' => 'sess_data',

    'db_time_col' => 'sess_time',

]);



$wsServer = new WsServer($chat);

$wsServer = new SessionProvider($wsServer, $sessionHandler);



$server = IoServer::factory(

    new HttpServer($wsServer),

    8080

);

3.4 安全防护清单

• 来源校验（Origin Check）：在

  1

  onOpen

  中检查

  1	$conn->httpRequest->getHeader('Origin')

  ，拒绝非白名单域名的连接，防止 CSWSH（跨站 WebSocket 劫持）攻击。

• 消息速率限制：对每个连接设置消息频率上限，防止恶意刷屏或洪水攻击。
• 消息大小限制：拒绝超过设定大小的消息帧，防止内存耗尽。
• 输入过滤：所有用户输入的消息必须经过

  1	htmlspecialchars

  过滤，防止 XSS 注入。

• TLS 加密（wss://）：生产环境必须使用 wss 协议，配合 Nginx 反向代理做 TLS 终端。

四、Laravel Reverb：Laravel 官方的现代方案

Laravel Reverb 是 Laravel 11 引入的官方 WebSocket 服务器，基于 ReactPHP 构建，与 Laravel 的广播系统深度集成。如果你在 Laravel 项目中需要实时通信，Reverb 是首选方案。

4.1 安装与配置

1
2
3
4
5
6
7
8
# 安装 Reverb

composer require laravel/reverb



# 发布配置文件

php artisan vendor:publish --provider="Laravel\Reverb\Providers\ReverbServiceProvider"



# 运行数据库迁移（如果使用数据库频道）

php artisan migrate

配置文件

的关键配置：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
// config/reverb.php

return [

    'apps' => [

        [

            'key'    => env('REVERB_APP_KEY', 'local-key'),

            'secret' => env('REVERB_APP_SECRET', 'local-secret'),

            'app_id' => env('REVERB_APP_ID', 'local-id'),

            'options' => [

                'host'     => env('REVERB_HOST', '0.0.0.0'),

                'port'     => env('REVERB_PORT', 8080),

                'scheme'   => env('REVERB_SCHEME', 'http'),

                'use_tls'  => env('REVERB_USE_TLS', false),

            ],

            'allowed_origins' => ['*'],

            // 限制每个连接的速率

            'ping_interval'   => 60,

            'max_request_size' => 10240,

        ],

    ],

    'server' => [

        'host'       => env('REVERB_SERVER_HOST', '0.0.0.0'),

        'port'       => env('REVERB_SERVER_PORT', 8080),

        'max_connections' => 10000,

    ],

];

4.2 启动 Reverb 服务

1
2
3
4
5
6
7
8
9
10
11
12
# 启动 WebSocket 服务器

php artisan reverb:start --debug



# 生产环境使用 Supervisor 守护

[program:reverb]

command=php /var/www/app/artisan reverb:start

autostart=true

autorestart=true

user=www-data

numprocs=1

redirect_stderr=true

stdout_logfile=/var/log/reverb.log

4.3 后端事件广播

Reverb 与 Laravel 的事件广播系统无缝衔接。定义一个广播事件：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
<?php

// app/Events/OrderStatusUpdated.php



namespace App\Events;



use Illuminate\Broadcasting\Channel;

use Illuminate\Broadcasting\InteractsWithSockets;

use Illuminate\Broadcasting\PrivateChannel;

use Illuminate\Contracts\Broadcasting\ShouldBroadcast;

use Illuminate\Foundation\Events\Dispatchable;

use Illuminate\Queue\SerializesModels;



class OrderStatusUpdated implements ShouldBroadcast

{

    use Dispatchable, InteractsWithSockets, SerializesModels;



    public function __construct(

        public int $orderId,

        public string $status,

        public float $amount,

    ) {}



    /**

     * 广播到指定频道

     */

    public function broadcastOn(): array

    {

        return [

            new PrivateChannel("orders.{$this->orderId}"),

            new Channel('orders'),

        ];

    }



    /**

     * 广播的数据负载

     */

    public function broadcastWith(): array

    {

        return [

            'order_id' => $this->orderId,

            'status'   => $this->status,

            'amount'   => $this->amount,

            'updated_at' => now()->toDateTimeString(),

        ];

    }



    /**

     * 事件名称

     */

    public function broadcastAs(): string

    {

        return 'order.updated';

    }

}



// 触发广播

event(new OrderStatusUpdated(12345, 'shipped', 299.99));

4.4 前端订阅（Laravel Echo + Pusher 兼容协议）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
// 安装前端依赖

// npm install laravel-echo pusher-js



import Echo from 'laravel-echo';

import Pusher from 'pusher-js';



window.Echo = new Echo({

    broadcaster: 'reverb',

    key: import.meta.env.VITE_REVERB_APP_KEY,

    wsHost: window.location.hostname,

    wsPort: 8080,

    forceTLS: false,

    disableStats: true,

    authorizer: (channel) => {

        return {

            authorize: (socketId, callback) => {

                fetch('/broadcasting/auth', {

                    method: 'POST',

                    headers: {

                        'Content-Type': 'application/json',

                        'X-CSRF-TOKEN': document.querySelector('meta[name="csrf-token"]').content,

                    },

                    body: JSON.stringify({

                        socket_id: socketId,

                        channel_name: channel.name,

                    }),

                })

                .then(res => res.json())

                .then(data => callback(false, data))

                .catch(err => callback(true, err));

            },

        };

    },

});



// 订阅公共频道

Echo.channel('orders')

    .listen('order.updated', (e) => {

        console.log(`订单 ${e.order_id} 状态更新: ${e.status}`);

    });



// 订阅私有频道（需要认证）

Echo.private(`orders.${orderId}`)

    .listen('order.updated', (e) => {

        updateOrderUI(e);

    })

    .listen('.order.cancelled', (e) => {

        showCancellation(e);

    });



// 在线状态通知

Echo.join('chat-room')

    .here((users) => console.log('在线用户', users))

    .joining((user) => console.log(`${user.name} 加入了`))

    .leaving((user) => console.log(`${user.name} 离开了`))

    .listen('MessageSent', (e) => {

        appendMessage(e.user, e.message);

    });

后端的频道认证路由自动注册在

中：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
// routes/channels.php

use Illuminate\Support\Facades\Broadcast;



// 私有频道认证

Broadcast::channel('orders.{orderId}', function ($user, $orderId) {

    return $user->orders()->where('id', $orderId)->exists();

});



// 在线状态频道

Broadcast::channel('chat-room', function ($user) {

    if (auth()->check()) {

        return ['id' => $user->id, 'name' => $user->name];

    }

    return false;

});

五、Swoole 高性能方案与水平扩展

5.1 Swoole WebSocket 服务器

当并发连接数达到万级以上时，纯 PHP 的 ReactPHP 方案可能成为瓶颈。Swoole 通过 C 扩展提供协程和原生异步 I/O，性能远超纯 PHP 实现：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
<?php

use Swoole\WebSocket\Server;

use Swoole\Table;



$server = new Server('0.0.0.0', 9502, SWOOLE_PROCESS, SWOOLE_SOCK_TCP | SWOOLE_SSL);



// 使用 Swoole Table 做跨进程共享内存

$userTable = new Table(10240);

$userTable->column('fd', Table::TYPE_INT);

$userTable->column('username', Table::TYPE_STRING, 64);

$userTable->column('room', Table::TYPE_STRING, 32);

$userTable->create();



$server->on('start', function (Server $server) {

    echo "Swoole WebSocket 启动在 0.0.0.0:9502

";

    echo "Worker 进程数: {$server->setting['worker_num']}

";

});



$server->on('open', function (Server $server, $request) use ($userTable) {

    echo "新连接 fd={$request->fd}

";



    // 认证：从 query string 获取 token

    $token = $request->get['token'] ?? '';

    $userInfo = verifyJwtToken($token);

    if (!$userInfo) {

        $server->push($request->fd, json_encode([

            'type' => 'error',

            'msg'  => '认证失败',

        ]));

        $server->disconnect($request->fd);

        return;

    }



    $userTable->set($request->fd, [

        'fd'       => $request->fd,

        'username' => $userInfo['name'],

        'room'     => 'general',

    ]);



    $server->push($request->fd, json_encode([

        'type' => 'connected',

        'msg'  => "欢迎 {$userInfo['name']}",

    ]));

});



$server->on('message', function (Server $server, $frame) use ($userTable) {

    $data = json_decode($frame->data, true);

    $user = $userTable->get($frame->fd);



    if (!$user) return;



    switch ($data['type'] ?? '') {

        case 'message':

            $msg = [

                'type'      => 'message',

                'username'  => $user['username'],

                'text'      => htmlspecialchars($data['text'] ?? '', ENT_QUOTES),

                'timestamp' => date('Y-m-d H:i:s'),

            ];

            // 向所有在线用户广播

            foreach ($userTable as $fd => $row) {

                $server->push($fd, json_encode($msg, JSON_UNESCAPED_UNICODE));

            }

            break;



        case 'room_join':

            $userTable->set($frame->fd, ['room' => $data['room'] ?? 'general']);

            break;

    }

});



$server->on('close', function (Server $server, $fd) use ($userTable) {

    $user = $userTable->get($fd);

    if ($user) {

        echo "{$user['username']} 断开连接

";

        $userTable->del($fd);

    }

});



$server->start();

5.2 多进程水平扩展与 Redis Pub/Sub

单个 WebSocket 进程能承载的连接数有限。在多 Worker 进程甚至多服务器架构下，进程之间的消息传递需要借助 Redis Pub/Sub 或其他消息中间件来实现广播：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
<?php

// Swoole + Redis Pub/Sub 实现跨进程广播



use Swoole\Coroutine\Redis;



// 在每个 Worker 中订阅 Redis 广播频道

$server->on('workerStart', function ($server, $workerId) {

    if ($workerId === 0) {  // 只在第一个 Worker 中订阅

        go(function () use ($server) {

            $redis = new Redis();

            $redis->connect('127.0.0.1', 6379);

            $redis->subscribe(['ws_broadcast'], function ($redis, $result) use ($server) {

                $message = $result[2]; // 频道消息内容

                $data = json_decode($message, true);



                // 向当前进程内的所有连接推送

                foreach ($server->connections as $fd) {

                    if (shouldReceive($fd, $data)) {

                        $server->push($fd, $message);

                    }

                }

            });

        });

    }

});



// 发送广播消息时，通过 Redis 发布

function broadcastToAll(Server $server, array $message): void

{

    go(function () use ($message) {

        $redis = new Redis();

        $redis->connect('127.0.0.1', 6379);

        $redis->publish('ws_broadcast', json_encode($message, JSON_UNESCAPED_UNICODE));

    });

}

六、Nginx 反向代理与生产部署

生产环境中，WebSocket 服务通常需要通过 Nginx 反向代理暴露给用户，Nginx 负责 TLS 终端、负载均衡和连接超时管理：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
# Nginx 配置：WebSocket 反向代理

upstream websocket_backend {

    server 127.0.0.1:8080;

    # 多实例负载均衡

    # server 127.0.0.1:8081;

    # server 10.0.0.2:8080;

}



server {

    listen 443 ssl http2;

    server_name ws.example.com;



    ssl_certificate     /etc/ssl/certs/example.pem;

    ssl_certificate_key /etc/ssl/private/example.key;

    ssl_protocols       TLSv1.2 TLSv1.3;



    # WebSocket 升级配置

    location /ws {

        proxy_pass http://websocket_backend;

        proxy_http_version 1.1;



        # 关键：HTTP 升级头

        proxy_set_header Upgrade $http_upgrade;

        proxy_set_header Connection "upgrade";



        # 传递真实客户端信息

        proxy_set_header Host $host;

        proxy_set_header X-Real-IP $remote_addr;

        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;

        proxy_set_header X-Forwarded-Proto $scheme;



        # WebSocket 长连接超时设置（默认 60s 太短）

        proxy_read_timeout 86400s;

        proxy_send_timeout 86400s;



        # 缓冲区设置

        proxy_buffering off;

        proxy_buffer_size 4k;

    }

}

6.1 健康检查与自动重连

WebSocket 连接可能因网络波动断开，前端必须实现自动重连机制：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
class ReconnectingWebSocket {

    private ws = null;

    private reconnectAttempts = 0;

    private maxReconnectAttempts = 10;

    private reconnectDelay = 1000;

    private heartbeatInterval = null;



    connect(url) {

        this.ws = new WebSocket(url);



        this.ws.onopen = () => {

            console.log('WebSocket 连接成功');

            this.reconnectAttempts = 0;

            this.startHeartbeat();

        };



        this.ws.onclose = () => {

            console.log('连接断开，准备重连');

            this.stopHeartbeat();

            this.scheduleReconnect(url);

        };



        this.ws.onerror = (error) => {

            console.error('WebSocket 错误', error);

        };

    }



    scheduleReconnect(url) {

        if (this.reconnectAttempts >= this.maxReconnectAttempts) {

            console.error('达到最大重连次数，停止重连');

            return;

        }



        // 指数退避策略

        const delay = this.reconnectDelay * Math.pow(2, this.reconnectAttempts);

        setTimeout(() => {

            this.reconnectAttempts++;

            console.log(`第 ${this.reconnectAttempts} 次重连，${delay}ms 后执行`);

            this.connect(url);

        }, delay);

    }



    startHeartbeat() {

        this.heartbeatInterval = setInterval(() => {

            if (this.ws && this.ws.readyState === WebSocket.OPEN) {

                this.ws.send(JSON.stringify({ type: 'ping' }));

            }

        }, 30000);

    }



    stopHeartbeat() {

        if (this.heartbeatInterval) {

            clearInterval(this.heartbeatInterval);

        }

    }

}

七、性能对比与选型建议

三种方案在不同并发场景下的表现差异显著，以下是在 4 核 8GB 内存的测试环境下的参考数据：

选型决策参考

• 小型项目 / 原型验证：选择 Ratchet，无需额外扩展，Composer 一键安装即可运行。适合几百到几千连接的场景。
• Laravel 项目：直接使用 Reverb。它与你现有的认证系统、队列、事件广播完美融合，几乎零额外学习成本。
• 高并发 / 超大规模：选择 Swoole。万级以上连接、需要协程并发的场景下，Swoole 的性能优势碾压纯 PHP 方案。配合 Swoole Table 共享内存和 Redis Pub/Sub，可以实现真正意义上的横向扩展。
• 已有 PHP-FPM 项目想增量添加 WebSocket：推荐独立部署一个 Ratchet 或 Swoole 服务，通过 Redis 或消息队列与主应用通信，不侵入现有架构。

八、运维监控与故障排查

8.1 连接数监控

实时监控 WebSocket 连接数是运维的基础。可以通过自定义 HTTP 端点暴露监控指标：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
<?php

// 在 Ratchet 服务中添加监控端点

use Ratchet\Http\HttpServerInterface;

use Ratchet\Http\HttpServer;

use Ratchet\WebSocket\WsServer;



class StatsHandler implements HttpServerInterface

{

    private Chat $chat;



    public function __construct(Chat $chat)

    {

        $this->chat = $chat;

    }



    public function onOpen(ConnectionInterface $conn): void

    {

        $stats = [

            'connected_clients' => $this->chat->getClientCount(),

            'total_users'       => $this->chat->getUserCount(),

            'rooms'             => $this->chat->getRoomStats(),

            'uptime'            => time() - START_TIME,

            'memory_usage'      => memory_get_usage(true),

            'peak_memory'       => memory_get_peak_usage(true),

        ];



        $body = json_encode($stats, JSON_PRETTY_PRINT);

        $conn->send("HTTP/1.1 200 OK

Content-Type: application/json



{$body}");

        $conn->close();

    }



    public function onMessage(ConnectionInterface $from, $msg): void {}

    public function onClose(ConnectionInterface $conn): void {}

    public function onError(ConnectionInterface $conn, \Exception $e): void {}

}



// 在 server.php 中将监控路由挂载

$router = new \Ratchet\Http\Router();

$router->addRoute('/stats', ['GET'], new StatsHandler($chat));

$router->addRoute('/ws', ['GET'], new WsServer($chat));

8.2 常见故障与排查

8.3 Supervisor 进程守护配置

无论是 Ratchet 还是 Swoole，生产环境都需要使用 Supervisor 或 systemd 做进程守护，确保服务崩溃后自动重启：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# /etc/supervisor/conf.d/websocket.conf

[program:websocket-chat]

command=php /var/www/chat/bin/server.php

directory=/var/www/chat

autostart=true

autorestart=true

startretries=3

user=www-data

redirect_stderr=true

stdout_logfile=/var/log/websocket-chat.log

stdout_logfile_maxbytes=10MB

stdout_logfile_backups=5

stopwaitsecs=10

killasgroup=true

stopsignal=TERM



# 控制命令

# supervisorctl start websocket-chat

# supervisorctl restart websocket-chat

# supervisorctl status websocket-chat

总结

PHP 生态在 WebSocket 实时通信领域已经相当成熟。Ratchet 适合轻量级项目和快速原型，Laravel Reverb 为 Laravel 开发者提供了一等公民级的开箱即用体验，而 Swoole 则在需要极致性能的场景下展现出不可替代的优势。无论选择哪种方案，核心的架构挑战是一致的：认证授权、消息路由、跨进程广播、连接管理和运维监控。理解这些底层原理后，你可以根据项目规模、团队能力和性能需求灵活选型。希望本文的完整代码示例和架构指南能帮助你快速在自己的 PHP 项目中落地实时通信功能。