WebSocket на PHP: создание real-time приложений продвинутого уровня

WebSocket на PHP: от теории к промышленному real-time приложению

Протокол WebSocket произвел революцию в веб-разработке, позволив создавать полноценные двунаправленные каналы связи между клиентом и сервером. В отличие от классического AJAX (long polling, Server-Sent Events), WebSocket обеспечивает постоянное соединение с минимальными накладными расходами. Для PHP-разработчика переход от базовых примеров «эхо-сервера» к промышленному real-time приложению — это серьезный вызов, связанный с управлением состоянием, масштабированием и асинхронной обработкой.

В этой статье мы рассмотрим продвинутые техники работы с WebSocket на PHP: архитектуру многопользовательского чата, интеграцию с Redis для горизонтального масштабирования, обработку фоновых задач и оптимизацию производительности.

1. Архитектура продвинутого WebSocket-сервера на Ratchet

Библиотека Ratchet (основанная на ReactPHP) остается золотым стандартом для WebSocket на PHP. Однако для production-решений простого onMessage недостаточно. Рассмотрим компонентную архитектуру с разделением ответственности:

<?php
use Ratchet\MessageComponentInterface;
use Ratchet\ConnectionInterface;
use React\EventLoop\LoopInterface;

class RealTimeHub implements MessageComponentInterface {
    private $clients;
    private $loop;
    private $redis;
    private $userRegistry;

    public function __construct(LoopInterface $loop, \Redis $redis) {
        $this->clients = new \SplObjectStorage;
        $this->loop = $loop;
        $this->redis = $redis;
        $this->userRegistry = [];
    }

    public function onOpen(ConnectionInterface $conn) {
        $this->clients->attach($conn);
        echo "Новое подключение: {$conn->resourceId}\n";
    }

    public function onMessage(ConnectionInterface $from, $msg) {
        $data = json_decode($msg, true);

        switch ($data['type'] ?? '') {
            case 'auth':
                $this->handleAuth($from, $data);
                break;
            case 'message':
                $this->handleMessage($from, $data);
                break;
            case 'subscribe':
                $this->handleSubscribe($from, $data);
                break;
            default:
                $from->send(json_encode(['error' => 'Unknown command']));
        }
    }

    public function onClose(ConnectionInterface $conn) {
        $this->clients->detach($conn);
        // Очистка регистрации пользователя
        if (isset($conn->userId)) {
            unset($this->userRegistry[$conn->userId]);
        }
    }

    public function onError(ConnectionInterface $conn, \Exception $e) {
        $conn->close();
    }

    private function handleAuth(ConnectionInterface $conn, array $data) {
        $token = $data['token'] ?? '';
        // Проверка JWT-токена
        $userId = $this->validateToken($token);
        if ($userId) {
            $conn->userId = $userId;
            $this->userRegistry[$userId] = $conn;
            $conn->send(json_encode(['type' => 'auth_ok']));
        } else {
            $conn->send(json_encode(['type' => 'auth_fail']));
        }
    }

    private function handleMessage(ConnectionInterface $from, array $data) {
        $targetUserId = $data['to'] ?? null;
        if ($targetUserId && isset($this->userRegistry[$targetUserId])) {
            $this->userRegistry[$targetUserId]->send(json_encode([
                'type' => 'message',
                'from' => $from->userId ?? 'anonymous',
                'text' => $data['text']
            ]));
        }
    }

    private function validateToken(string $token): ?int {
        // Имитация проверки JWT
        $payload = json_decode(base64_decode(explode('.', $token)[1] ?? ''), true);
        return $payload['sub'] ?? null;
    }
}

Ключевые улучшения: разделение типов сообщений (auth, message, subscribe), привязка соединения к пользователю через JWT-токен, и централизованный реестр userRegistry для адресной отправки сообщений.

2. Масштабирование: Redis Pub/Sub и балансировка

Один PHP-процесс не выдержит тысячи одновременных подключений. Для горизонтального масштабирования необходимо, чтобы все экземпляры сервера обменивались данными. Идеальное решение — Redis Pub/Sub.

Реализуем подсистему кросс-серверного обмена:

<?php
use React\EventLoop\LoopInterface;
use Clue\React\Redis\Factory as RedisFactory;
use Clue\React\Redis\Client as RedisClient;

class RedisPubSubBridge {
    private $loop;
    private $redisClient;
    private $hub;

    public function __construct(LoopInterface $loop, RealTimeHub $hub) {
        $this->loop = $loop;
        $this->hub = $hub;
    }

    public function connect(string $redisHost = '127.0.0.1:6379') {
        $factory = new RedisFactory($this->loop);
        $factory->createClient($redisHost)->then(function (RedisClient $client) {
            $this->redisClient = $client;
            // Подписываемся на канал глобальных сообщений
            $client->subscribe('global:messages');
            $client->on('message', function ($channel, $payload) {
                $this->hub->broadcastToAll($payload);
            });
        });
    }

    public function publish(string $channel, string $message) {
        if ($this->redisClient) {
            $this->redisClient->publish($channel, $message);
        }
    }
}

Теперь, когда пользователь на сервере A отправляет сообщение, оно публикуется в Redis. Все остальные сервера (B, C, D) получают это сообщение через подписку и транслируют его своим подключенным клиентам. Это позволяет запускать неограниченное количество PHP-воркеров за балансировщиком (например, Nginx с модулем stream).

3. Асинхронные фоновые задачи и управление состоянием

Real-time приложения часто требуют выполнения тяжелых операций (сохранение в БД, генерация превью, отправка email) без блокировки основного цикла событий. Используем ReactPHP Promise и очередь на основе Redis:

<?php
use React\Promise\Deferred;
use React\Promise\Promise;

class AsyncTaskProcessor {
    private $loop;
    private $redisQueue;

    public function __construct(LoopInterface $loop, \Redis $redis) {
        $this->loop = $loop;
        $this->redisQueue = $redis;
    }

    public function enqueue(string $taskType,