本文目录导读：
我来详细介绍一下在PHP项目中使用Ratchet实现WebSocket。
Ratchet简介
Ratchet是一个PHP实现的WebSocket库,基于ReactPHP事件循环，支持：
- WebSocket服务器
- HTTP长轮询
- WebSocket客户端
安装
composer require cboden/ratchet
基础WebSocket服务器实现
1 创建消息处理器
<?php
// src/Chat.php
namespace MyApp;
use Ratchet\MessageComponentInterface;
use Ratchet\ConnectionInterface;
class Chat implements MessageComponentInterface {
protected $clients;
protected $users = [];
public function __construct() {
$this->clients = new \SplObjectStorage;
echo "Chat server initialized\n";
}
public function onOpen(ConnectionInterface $conn) {
// 存储新的连接
$this->clients->attach($conn);
echo "New connection: {$conn->resourceId}\n";
}
public function onMessage(ConnectionInterface $from, $msg) {
$data = json_decode($msg, true);
switch($data['type']) {
case 'login':
$this->handleLogin($from, $data);
break;
case 'message':
$this->handleMessage($from, $data);
break;
case 'private':
$this->handlePrivateMessage($from, $data);
break;
case 'typing':
$this->handleTyping($from, $data);
break;
}
}
public function onClose(ConnectionInterface $conn) {
$this->clients->detach($conn);
// 处理用户断开
if(isset($this->users[$conn->resourceId])) {
$username = $this->users[$conn->resourceId];
unset($this->users[$conn->resourceId]);
$this->broadcast([
'type' => 'user_offline',
'username' => $username
]);
}
echo "Connection {$conn->resourceId} has disconnected\n";
}
public function onError(ConnectionInterface $conn, \Exception $e) {
echo "Error: {$e->getMessage()}\n";
$conn->close();
}
private function handleLogin($conn, $data) {
$this->users[$conn->resourceId] = $data['username'];
// 广播用户上线
$this->broadcast([
'type' => 'user_online',
'username' => $data['username'],
'users' => array_values($this->users)
], $conn);
// 发送当前在线用户列表给新用户
$conn->send(json_encode([
'type' => 'login_success',
'username' => $data['username'],
'users' => array_values($this->users)
]));
}
private function handleMessage($from, $data) {
$response = [
'type' => 'message',
'from' => $this->users[$from->resourceId] ?? 'Unknown',
'message' => $data['message'],
'timestamp' => date('Y-m-d H:i:s')
];
$this->broadcast($response);
}
private function handlePrivateMessage($from, $data) {
$response = [
'type' => 'private',
'from' => $this->users[$from->resourceId],
'message' => $data['message'],
'timestamp' => date('Y-m-d H:i:s')
];
// 发送给指定用户
foreach ($this->clients as $client) {
if (isset($this->users[$client->resourceId]) &&
$this->users[$client->resourceId] === $data['to']) {
$client->send(json_encode($response));
break;
}
}
}
private function handleTyping($from, $data) {
// 广播正在输入状态
$this->broadcast([
'type' => 'typing',
'username' => $this->users[$from->resourceId],
'is_typing' => $data['is_typing']
], $from);
}
private function broadcast($data, $exclude = null) {
$json = json_encode($data);
foreach ($this->clients as $client) {
if ($client !== $exclude) {
$client->send($json);
}
}
}
}
2 创建启动脚本
<?php
// bin/chat-server.php
require __DIR__ . '/../vendor/autoload.php';
use Ratchet\Server\IoServer;
use Ratchet\Http\HttpServer;
use Ratchet\WebSocket\WsServer;
use MyApp\Chat;
$port = $argv[1] ?? 8080;
$server = IoServer::factory(
new HttpServer(
new WsServer(
new Chat()
)
),
$port
);
echo "WebSocket server started on port {$port}\n";
$server->run();
高级功能实现
1 房间/频道系统
<?php
// src/RoomServer.php
namespace MyApp;
use Ratchet\MessageComponentInterface;
use Ratchet\ConnectionInterface;
class RoomServer implements MessageComponentInterface {
private $clients;
private $rooms = [];
private $users = [];
public function __construct() {
$this->clients = new \SplObjectStorage;
}
public function onOpen(ConnectionInterface $conn) {
$this->clients->attach($conn);
$this->users[$conn->resourceId] = [
'conn' => $conn,
'rooms' => []
];
}
public function onMessage(ConnectionInterface $from, $msg) {
$data = json_decode($msg, true);
switch($data['action']) {
case 'join_room':
$this->joinRoom($from, $data['room']);
break;
case 'leave_room':
$this->leaveRoom($from, $data['room']);
break;
case 'room_message':
$this->roomMessage($from, $data);
break;
case 'create_room':
$this->createRoom($from, $data['room_name']);
break;
case 'list_rooms':
$this->listRooms($from);
break;
}
}
private function joinRoom($conn, $roomName) {
if (!isset($this->rooms[$roomName])) {
$this->rooms[$roomName] = new \SplObjectStorage;
}
$this->rooms[$roomName]->attach($conn);
$this->users[$conn->resourceId]['rooms'][] = $roomName;
// 通知房间成员
$this->broadcastToRoom($roomName, [
'action' => 'user_joined',
'user_id' => $conn->resourceId,
'room' => $roomName,
'users_count' => $this->rooms[$roomName]->count()
], $conn);
$conn->send(json_encode([
'action' => 'joined_room',
'room' => $roomName,
'success' => true
]));
}
private function leaveRoom($conn, $roomName) {
if (isset($this->rooms[$roomName])) {
$this->rooms[$roomName]->detach($conn);
// 通知房间成员
$this->broadcastToRoom($roomName, [
'action' => 'user_left',
'user_id' => $conn->resourceId,
'room' => $roomName
]);
// 清理空房间
if ($this->rooms[$roomName]->count() === 0) {
unset($this->rooms[$roomName]);
}
}
}
private function roomMessage($from, $data) {
$response = [
'action' => 'room_message',
'room' => $data['room'],
'from' => $from->resourceId,
'message' => $data['message'],
'timestamp' => time()
];
$this->broadcastToRoom($data['room'], $response, $from);
}
private function createRoom($conn, $roomName) {
if (!isset($this->rooms[$roomName])) {
$this->rooms[$roomName] = new \SplObjectStorage;
$conn->send(json_encode([
'action' => 'room_created',
'room' => $roomName,
'success' => true
]));
} else {
$conn->send(json_encode([
'action' => 'room_created',
'room' => $roomName,
'success' => false,
'error' => 'Room already exists'
]));
}
}
private function listRooms($conn) {
$roomList = [];
foreach ($this->rooms as $name => $storage) {
$roomList[] = [
'name' => $name,
'users_count' => $storage->count()
];
}
$conn->send(json_encode([
'action' => 'room_list',
'rooms' => $roomList
]));
}
private function broadcastToRoom($roomName, $data, $exclude = null) {
if (isset($this->rooms[$roomName])) {
$json = json_encode($data);
foreach ($this->rooms[$roomName] as $client) {
if ($client !== $exclude) {
$client->send($json);
}
}
}
}
public function onClose(ConnectionInterface $conn) {
// 从所有房间移除
foreach ($this->rooms as $roomName => $storage) {
if ($storage->contains($conn)) {
$storage->detach($conn);
$this->broadcastToRoom($roomName, [
'action' => 'user_left',
'user_id' => $conn->resourceId,
'room' => $roomName
]);
}
}
$this->clients->detach($conn);
unset($this->users[$conn->resourceId]);
}
public function onError(ConnectionInterface $conn, \Exception $e) {
$conn->close();
}
}
2 认证中间件
<?php
// src/AuthProvider.php
namespace MyApp;
use Ratchet\MessageComponentInterface;
use Ratchet\ConnectionInterface;
class AuthProvider implements MessageComponentInterface {
private $decorated;
private $authTokens = [];
public function __construct(MessageComponentInterface $component) {
$this->decorated = $component;
}
public function onOpen(ConnectionInterface $conn) {
$queryString = $conn->httpRequest->getUri()->getQuery();
parse_str($queryString, $params);
$token = $params['token'] ?? '';
if ($this->validateToken($token)) {
$this->authTokens[$conn->resourceId] = $token;
$this->decorated->onOpen($conn);
} else {
$conn->send(json_encode([
'type' => 'error',
'message' => 'Authentication failed'
]));
$conn->close();
}
}
public function onMessage(ConnectionInterface $from, $msg) {
if (isset($this->authTokens[$from->resourceId])) {
$this->decorated->onMessage($from, $msg);
}
}
public function onClose(ConnectionInterface $conn) {
if (isset($this->authTokens[$conn->resourceId])) {
unset($this->authTokens[$conn->resourceId]);
}
$this->decorated->onClose($conn);
}
public function onError(ConnectionInterface $conn, \Exception $e) {
$this->decorated->onError($conn, $e);
}
private function validateToken($token) {
// 实现token验证逻辑
return !empty($token) && strlen($token) > 10;
}
}
JavaScript客户端
// public/chat.js
class WebSocketClient {
constructor(url) {
this.url = url;
this.ws = null;
this.reconnectAttempts = 0;
this.maxReconnectAttempts = 5;
this.listeners = {};
}
connect() {
this.ws = new WebSocket(this.url);
this.ws.onopen = () => {
console.log('Connected to WebSocket server');
this.reconnectAttempts = 0;
this.emit('connected');
};
this.ws.onmessage = (event) => {
const data = JSON.parse(event.data);
this.emit('message', data);
this.emit(data.type || data.action, data);
};
this.ws.onclose = () => {
console.log('Disconnected from WebSocket server');
this.emit('disconnected');
this.reconnect();
};
this.ws.onerror = (error) => {
console.error('WebSocket error:', error);
this.emit('error', error);
};
}
reconnect() {
if (this.reconnectAttempts < this.maxReconnectAttempts) {
this.reconnectAttempts++;
const delay = Math.min(1000 * Math.pow(2, this.reconnectAttempts), 30000);
console.log(`Reconnecting in ${delay}ms...`);
setTimeout(() => this.connect(), delay);
} else {
console.error('Max reconnection attempts reached');
this.emit('reconnect_failed');
}
}
send(data) {
if (this.ws && this.ws.readyState === WebSocket.OPEN) {
this.ws.send(JSON.stringify(data));
} else {
console.error('WebSocket is not connected');
}
}
on(event, callback) {
if (!this.listeners[event]) {
this.listeners[event] = [];
}
this.listeners[event].push(callback);
}
off(event, callback) {
if (this.listeners[event]) {
this.listeners[event] = this.listeners[event]
.filter(cb => cb !== callback);
}
}
emit(event, data) {
if (this.listeners[event]) {
this.listeners[event].forEach(callback => callback(data));
}
}
disconnect() {
if (this.ws) {
this.ws.close();
}
}
}
// 使用示例
const client = new WebSocketClient('ws://localhost:8080');
client.on('connected', () => {
console.log('Ready to send messages');
});
client.on('message', (data) => {
console.log('Received:', data);
});
client.on('error', (error) => {
console.error('Error:', error);
});
client.connect();
// 发送消息
client.send({
type: 'login',
username: 'User1'
});
性能优化建议
// 6.1 使用连接池
class ConnectionPool {
private $connections = [];
public function add($id, $connection) {
$this->connections[$id] = [
'connection' => $connection,
'last_active' => time()
];
}
public function remove($id) {
unset($this->connections[$id]);
}
// 清理超时连接
public function cleanup($timeout = 300) {
$now = time();
foreach ($this->connections as $id => $data) {
if ($now - $data['last_active'] > $timeout) {
$data['connection']->close();
$this->remove($id);
}
}
}
}
// 6.2 消息队列处理
class MessageQueue {
private $queue = [];
public function push($message) {
$this->queue[] = $message;
}
public function processBatch($callback) {
while (!empty($this->queue)) {
$batch = array_splice($this->queue, 0, 100);
$callback($batch);
}
}
}
// 6.3 使用内存缓存（Redis）
class RedisBackend {
private $redis;
public function __construct() {
$this->redis = new \Redis();
$this->redis->connect('127.0.0.1', 6379);
}
public function saveUserState($userId, $data) {
$this->redis->set("user:{$userId}", json_encode($data));
$this->redis->expire("user:{$userId}", 3600);
}
public function getUserState($userId) {
$data = $this->redis->get("user:{$userId}");
return $data ? json_decode($data, true) : null;
}
}
Nginx反向代理配置
# /etc/nginx/sites-available/chat.example.com
upstream websocket {
server 127.0.0.1:8080;
}
server {
listen 80;
server_name chat.example.com;
location /ws {
proxy_pass http://websocket;
proxy_http_version 1.1;
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;
proxy_read_timeout 86400;
proxy_send_timeout 86400;
}
location / {
proxy_pass http://127.0.0.1:8000; # PHP应用
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
}
}
Docker部署
# Dockerfile
FROM php:8.1-cli
RUN apt-get update && apt-get install -y \
git \
unzip \
libevent-dev \
&& pecl install event \
&& docker-php-ext-enable event
COPY --from=composer:latest /usr/bin/composer /usr/bin/composer
WORKDIR /app
COPY composer.json composer.lock ./
RUN composer install --no-interaction --optimize-autoloader
COPY . .
EXPOSE 8080
CMD ["php", "bin/chat-server.php"]
# docker-compose.yml
version: '3.8'
services:
websocket:
build: .
ports:
- "8080:8080"
volumes:
- .:/app
environment:
- REDIS_HOST=redis
- DB_HOST=mysql
restart: unless-stopped
redis:
image: redis:alpine
ports:
- "6379:6379"
restart: unless-stopped
Ratchet提供了完整的WebSocket实现方案,关键点：
- 使用消息组件接口处理连接生命周期
- 实现心跳检测和重连机制
- 支持房间/频道系统
- 集成认证中间件
- 配合Nginx实现负载均衡
- 使用Docker简化部署
选择Ratchet的场景：
- 需要长连接实时通信
- 聊天应用、在线游戏
- 实时数据推送
- 协作编辑工具
版权声明：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直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
