Современные интерактивные приложения (биржевые котировки, чаты, дашборды совместной работы) требуют доставки событий с задержкой менее 20 мс. Постоянный опрос сервера (Short Polling) разрушает производительность базы данных. Индустриальный стандарт делится на два сценария: однонаправленные потоки данных через SSE и полнодуплексные двунаправленные сокеты WebSockets. Для работы под высокой нагрузкой сокеты масштабируются горизонтально через шину сообщений Redis Pub/Sub.
[ Браузер Алисы ] [ Браузер Боба ] [ Браузер Карла ]
│ │ │
▼ WebSocket ▼ WebSocket ▼ WebSocket
┌──────────────────┐ ┌──────────────────┐ ┌──────────────────┐
│ Node.js Pod #1 │ │ Node.js Pod #2 │ │ Node.js Pod #3 │
│ (Conn: Алиса) │ │ (Conn: Боб) │ │ (Conn: Карл) │
└────────┬─────────┘ └────────┬─────────┘ └────────┬─────────┘
│ │ │
│ Алиса пишет в канал: │ Принимает событие │ Принимает событие
│ redis.publish('room:42') │ и шлет в сокет Боба │ и шлет в сокет Карла
▼ ▲ ▲
┌────────────────────────────────────┴───────────────────────────┴─────────┐
│ Redis Pub/Sub Message Bus (Cluster) │
│ Канал трансляции: "chat_room_42" │
└──────────────────────────────────────────────────────────────────────────┘Создание инстанса WebSocketServer без автоматического биндинга порта (noServer: true) для интеграции с Fastify.
Анализ HTTP-запроса на этапе переключения протокола: парсинг cookie access_token и валидация JWT до рукопожатия.
Таймер интервала (30 секунд) с рассылкой фреймов ping и закрытием соединений (terminate), не приславших ответный pong.
Подключение двух независимых клиентов ioredis (Publisher и Subscriber) для межсервисной маршрутизации комнат.
Реализация HTTP GET-роута text/event-stream для легких уведомлений без расхода сокетных соединений.
Почему аутентификацию WebSocket-клиента критически важно производить в момент события HTTP 'upgrade', а не после открытия сокета через первое сообщение?
Сервер перехватывает событие upgrade, валидирует токен из cookie и связывает сессию пользователя с сокетом.
Неавторизованным клиентам возвращается HTTP 401 с немедленным сокетным сокетом socket.destroy().
Активный сокет регистрируется в Map<string, Set<WebSocket>> по идентификатору комнаты (roomId).
Один пользователь может иметь несколько вкладок, подключенных к одной комнате.
При отправке сообщения сервер публикует JSON-пейлоад в канал Redis redisPub.publish(`room:${roomId}`, payload).
Событие моментально дублируется на все остальные запущенные инстансы сервиса.
Каждый инстанс получает событие из Redis и рассылает его только своим локально подключенным клиентам.
Идеальная изоляция нагрузки: каждый узел работает только со своими сокетами.
import { IncomingMessage } from 'node:http';
import { Duplex } from 'node:stream';
import { WebSocketServer, WebSocket } from 'ws';
import { Redis } from 'ioredis';
export interface AuthenticatedSocket extends WebSocket {
isAlive: boolean;
userId: string;
roomId?: string;
}
export class RealtimeGateway {
private wss: WebSocketServer;
private redisPub: Redis;
private redisSub: Redis;
private roomSockets: Map<string, Set<AuthenticatedSocket>> = new Map();
private heartbeatInterval!: NodeJS.Timeout;
constructor(redisOptions: any) {
this.wss = new WebSocketServer({ noServer: true });
this.redisPub = new Redis(redisOptions);
this.redisSub = new Redis(redisOptions);
this.setupRedisSubscriptions();
this.setupHeartbeat();
this.setupConnectionHandlers();
}
/**
* Безопасная обработка HTTP Upgrade рукопожатия
*/
public handleUpgrade(request: IncomingMessage, socket: Duplex, head: Buffer, userId: string): void {
this.wss.handleUpgrade(request, socket, head, (ws) => {
const authWs = ws as AuthenticatedSocket;
authWs.isAlive = true;
authWs.userId = userId;
this.wss.emit('connection', authWs, request);
});
}
private setupConnectionHandlers(): void {
this.wss.on('connection', (ws: AuthenticatedSocket) => {
ws.on('pong', () => {
ws.isAlive = true;
});
ws.on('message', (data: Buffer) => {
try {
const payload = JSON.parse(data.toString('utf8'));
this.handleClientMessage(ws, payload);
} catch (err: unknown) {
ws.send(JSON.stringify({ error: 'Невалидный JSON-формат' }));
}
});
ws.on('close', () => {
this.leaveRoom(ws);
});
});
}
private handleClientMessage(ws: AuthenticatedSocket, payload: any): void {
if (payload.action === 'join_room' && typeof payload.roomId === 'string') {
this.joinRoom(ws, payload.roomId);
} else if (payload.action === 'send_message' && ws.roomId) {
// Публикуем сообщение в Redis для всех инстансов
const broadcastData = {
roomId: ws.roomId,
senderId: ws.userId,
text: String(payload.text),
timestamp: new Date().toISOString(),
};
void this.redisPub.publish(`room:${ws.roomId}`, JSON.stringify(broadcastData));
}
}
private joinRoom(ws: AuthenticatedSocket, roomId: string): void {
this.leaveRoom(ws); // Покидаем предыдущую комнату
ws.roomId = roomId;
if (!this.roomSockets.has(roomId)) {
this.roomSockets.set(roomId, new Set());
// Подписываемся на канал в Redis, если это первая подписка на данном узле
void this.redisSub.subscribe(`room:${roomId}`);
}
this.roomSockets.get(roomId)!.add(ws);
ws.send(JSON.stringify({ event: 'joined', roomId }));
}
private leaveRoom(ws: AuthenticatedSocket): void {
if (ws.roomId && this.roomSockets.has(ws.roomId)) {
const room = this.roomSockets.get(ws.roomId)!;
room.delete(ws);
if (room.size === 0) {
this.roomSockets.delete(ws.roomId);
void this.redisSub.unsubscribe(`room:${ws.roomId}`);
}
ws.roomId = undefined;
}
}
private setupRedisSubscriptions(): void {
this.redisSub.on('message', (channel: string, message: string) => {
const roomId = channel.replace('room:', '');
const localClients = this.roomSockets.get(roomId);
if (localClients) {
for (const client of localClients) {
if (client.readyState === WebSocket.OPEN) {
client.send(message);
}
}
}
});
}
/**
* Детектирование оборванных соединений (Half-Open Sockets)
*/
private setupHeartbeat(): void {
this.heartbeatInterval = setInterval(() => {
for (const client of this.wss.clients) {
const authWs = client as AuthenticatedSocket;
if (!authWs.isAlive) {
// Сокет не ответил на предыдущий ping — принудительно разрываем
authWs.terminate();
continue;
}
authWs.isAlive = false;
authWs.ping();
}
}, 30000); // Проверка каждые 30 секунд
}
public async close(): Promise<void> {
clearInterval(this.heartbeatInterval);
await Promise.all([this.redisPub.quit(), this.redisSub.quit()]);
this.wss.close();
}
}import { FastifyPluginAsync } from 'fastify';
export const sseController: FastifyPluginAsync = async (app) => {
app.get('/events', async (request, reply) => {
// Устанавливаем заголовки потокового ответа SSE
reply.raw.setHeader('Content-Type', 'text/event-stream');
reply.raw.setHeader('Cache-Control', 'no-cache, no-transform');
reply.raw.setHeader('Connection', 'keep-alive');
reply.raw.setHeader('X-Accel-Buffering', 'no'); // Отключаем буферизацию в Nginx
reply.raw.write(': connection established\n\n');
// Отправляем периодические события клиенту
const interval = setInterval(() => {
const eventData = JSON.stringify({
serverTime: new Date().toISOString(),
load: process.cpuUsage(),
});
reply.raw.write(`event: metric_update\ndata: ${eventData}\n\n`);
}, 5000);
// Очистка таймера при закрытии соединения клиентом
request.raw.on('close', () => {
clearInterval(interval);
});
});
};Клиент Redis, переведенный в режим подписки командой subscribe(), блокируется: он может принимать только команды SUBSCRIBE, UNSUBSCRIBE, PING и QUIT. Попытка выполнить redis.set() или redis.publish() на том же экземпляре вызовет фатальное падение приложения. Всегда создавайте как минимум два раздельных подключения: одно для публикации/команд, второе — исключительно для подписки.
Nginx по умолчанию накапливает исходящий поток данных в буфере размером 4–8 КБ перед отправкой клиенту. В результате браузер не получит события SSE в реальном времени, а получит сразу пачку событий с задержкой в несколько минут. Всегда передавайте заголовок X-Accel-Buffering: no.
Если на каждое входящее сообщение навешивать обработчик ws.on('message', ...) внутри другой функции без отписки, Node.js быстро выбросит предупреждение MaxListenersExceededWarning: Possible EventEmitter memory leak detected. Всегда централизуйте подписки на уровне жизненного цикла сокета.
Если сервер Node.js закрывается, не уведомив WebSocket-клиентов, браузеры попытаются мгновенно переподключиться миллионами одновременных запросов, создав шторм реконнектов (Thundering Herd). Рассылайте клиентам статус 1001 (Going Away) с небольшой рандомизированной задержкой повторного подключения (Jitter).
Зачем в протоколе WebSockets требуется отправлять служебные фреймы ping каждые 30 секунд, если TCP-соединение уже обеспечивает контроль целостности?