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

PHP项目WebSocket与Ratchet

wen PHP项目 2

本文目录导读：

PHP项目WebSocket与Ratchet

  1. Ratchet简介
  2. 安装
  3. 基础WebSocket服务器实现
  4. 高级功能实现
  5. JavaScript客户端
  6. 性能优化建议
  7. Nginx反向代理配置
  8. Docker部署

我来详细介绍一下在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的场景：

  • 需要长连接实时通信
  • 聊天应用、在线游戏
  • 实时数据推送
  • 协作编辑工具

上一篇PHP项目流式HTTP与回调

下一篇PHP项目SOAP与WebService

相关文章

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