Организация двунаправленного и потокового обмена данными между клиентом и сервером требует перехода от классической парадигмы "запрос-ответ" (Request-Response) к долгоживущим постоянным соединениям (Persistent Connections). В экосистеме Node.js для этих задач применяются два фундаментальных стандарта: WebSocket (RFC 6455) для полнодуплексного обмена и Server-Sent Events (SSE, W3C Recommendation) для однонаправленного реактивного стриминга.
Сравнение сетевого стека WebSocket и Server-Sent Events (SSE):
1. WebSocket (RFC 6455) - Полнодуплексный бинарный протокол:
Client ──[HTTP GET Upgrade: websocket]──> Server (Порт 80/443)
Client <──[HTTP 101 Switching Protocols]── Server
==================================================================
Двунаправленный обмен фреймами (TCP Stream):
Client ──[Opcode: Binary/Text, Masked Payload]──> Server
Client <──[Opcode: Binary/Text, Unmasked Payload]── Server
Client <──[Opcode: 0x9 Ping] / ──[Opcode: 0xA Pong]── Server
2. Server-Sent Events (SSE) - Однонаправленный текстовый поток:
Client ──[HTTP GET (Accept: text/event-stream)]──> Server (Порт 80/443)
Client <──[HTTP 200 OK (Transfer-Encoding: chunked)]── Server
==================================================================
Однонаправленный сброс событий (Server -> Client):
Server ──[id: 101\nevent: delta\ndata: {"chunk":"Hello"}\n\n]──> ClientWebSocket стартует как стандартный HTTP/1.1 запрос на обновление протокола (HTTP Upgrade Handshake):
Connection: Upgrade, Upgrade: websocket, Sec-WebSocket-Version: 13 и сгенерированную случайную строку Sec-WebSocket-Key (16 байт в кодировке Base64).258EAFA5-E914-47DA-95CA-C5AB0DC85B11), вычисляет криптографический SHA-1 дайджест, кодирует его в Base64 и возвращает в заголовке Sec-WebSocket-Accept со статусом HTTP 101 Switching Protocols.net.Socket переключается на бинарное фреймирование WebSocket.Каждый фрейм WebSocket содержит заголовок переменной длины от 2 до 14 байт:
permessage-deflate).Server-Sent Events (SSE) функционирует поверх стандартного протокола HTTP (HTTP/1.1 или HTTP/2) без переключения на альтернативные бинарные протоколы.
Content-Type: text/event-stream, Cache-Control: no-cache, no-transform, Connection: keep-alive.\n\n. Поддерживаются служебные поля: id: (идентификатор события), event: (имя кастомного события для EventListener), retry: (таймаут автопереподключения в мс) и data: (полезная нагрузка).EventSource автоматически повторяет подключение, передавая заголовок Last-Event-ID, что позволяет серверу дослать потерянные события из буфера без усложнения клиентской логики.Разработка отказоустойчивых систем реального времени строится вокруг контроля жизненного цикла сокетов, преодоления сетевых задержек мобильных сетей и управления буферами памяти процесса Node.js.
HTTP Upgrade Half-Open Socket (Мертвый сокет) Heartbeat Mechanism Socket Backpressure (bufferedAmount) Distributed Pub/Sub AdapterГоризонтальное масштабирование WebSocket-кластера за L7 балансировщиком
[ Балансировщик нагрузки (AWS ALB / NGINX) ]
│ (Round-Robin / IP-Hash)
┌────────────────┴────────────────┐
▼ ▼
[Node.js Pod 1 (ws)] [Node.js Pod 2 (ws)]
│ │ │ │
Клиент A Клиент B Клиент C Клиент D
│ ▲
│ (Отправка сообщения клиенту D) │
▼ │
[Pod 1 Publisher] [Pod 2 Subscriber]
│ ▲
└──────────► [ Redis Cluster ] ───┘
(Pub/Sub Канал: "chat:room_1")Клиент отправляет заголовок Upgrade: websocket со случайным ключом Sec-WebSocket-Key. Сервер валидирует JWT-токен.
Сервер возвращает SHA-1 хэш ключа в Sec-WebSocket-Accept. Сетевой сокет переключается на бинарный протокол.
Двунаправленный обмен типизированными фреймами. Клиентские фреймы маскируются 4-байтовым XOR-ключом.
Сервер каждые 30 секунд рассылает Ping-кадры. Сокеты, не вернувшие Pong в течение интервала, уничтожаются через terminate().
При закрытии стороны обмениваются кадрами с Opcode 0x8 (Close Frame) со статус-кодом закрытия и закрывают TCP-сокет.
В каком сценарии архитектурно более оправдан выбор Server-Sent Events (SSE) вместо полнодуплексного WebSocket?
ws для передачи десятков тысяч бинарных тиков в секунду с использованием кастомных бинарных Protobuf или MessagePack сериализаторов с нулевыми накладными расходами на HTTP-заголовки.Библиотека ws является стандартом де-факто для Node.js, обеспечивая максимальную производительность парсинга фреймов на чистом C/C++ через нативный модуль bufferutil.
Для предотвращения утечек памяти и исчерпания файловых дескрипторов из-за "мертвых" соединений (Half-Open Sockets) сервер обязан реализовать активный Heartbeat-конвейер:
import { createServer, IncomingMessage } from "node:http";
import { WebSocketServer, WebSocket, RawData } from "ws";
// Расширяем интерфейс WebSocket для трекинга состояния соединения
interface TrackedWebSocket extends WebSocket {
isAlive: boolean;
userId?: string;
subscribedChannels: Set<string>;
}
const server = createServer();
const wss = new WebSocketServer({
noServer: true, // Ручная обработка HTTP Upgrade для безопасной аутентификации
maxPayload: 1024 * 1024, // Жесткий лимит полезной нагрузки: 1 MB
});
// 1. Безопасная обработка HTTP Upgrade Handshake
server.on("upgrade", (request: IncomingMessage, socket, head) => {
const url = new URL(request.url ?? "", `http://${request.headers.host}`);
const token = url.searchParams.get("token");
// Валидация токена аутентификации до переключения протокола
if (!token || token !== "valid-production-token") {
socket.write("HTTP/1.1 401 Unauthorized\r\n\r\n");
socket.destroy();
return;
}
wss.handleUpgrade(request, socket, head, (ws) => {
const trackedWs = ws as TrackedWebSocket;
trackedWs.isAlive = true;
trackedWs.userId = "usr_" + Math.random().toString(36).substring(2, 9);
trackedWs.subscribedChannels = new Set();
wss.emit("connection", trackedWs, request);
});
});
// 2. Обработка установленного соединения
wss.on("connection", (ws: TrackedWebSocket) => {
console.log(`[WS] Клиент подключен: ${ws.userId}`);
// Сброс флага при получении Pong от клиента
ws.on("pong", () => {
ws.isAlive = true;
});
ws.on("message", (data: RawData, isBinary: boolean) => {
try {
if (isBinary) {
// Обработка бинарного фрейма (Buffer)
console.log(`[WS Binary] Получено ${data.length} байт`);
return;
}
const message = JSON.parse(data.toString());
handleClientMessage(ws, message);
} catch (err) {
console.error("[WS] Некорректный формат сообщения:", err);
ws.send(JSON.stringify({ error: "Invalid JSON format" }));
}
});
ws.on("close", (code, reason) => {
console.log(`[WS] Клиент ${ws.userId} отключен (Код: ${code}, Причина: ${reason.toString()})`);
});
ws.on("error", (error) => {
console.error(`[WS Error ${ws.userId}]:`, error);
});
});
function handleClientMessage(ws: TrackedWebSocket, payload: Record<string, unknown>) {
if (payload.action === "subscribe" && typeof payload.channel === "string") {
ws.subscribedChannels.add(payload.channel);
ws.send(JSON.stringify({ status: "subscribed", channel: payload.channel }));
}
}
// 3. Heartbeat-конвейер для детекции Half-Open сокетов
// Каждые 30 секунд проверяем пиры, не ответившие на предыдущий Ping
const HEARTBEAT_INTERVAL = 30000;
const interval = setInterval(() => {
for (const client of wss.clients) {
const ws = client as TrackedWebSocket;
if (!ws.isAlive) {
console.warn(`[WS Heartbeat] Сокет ${ws.userId} не ответил на Ping. Принудительное закрытие.`);
// terminate() закрывает сокет немедленно на уровне TCP RST, не ожидая хэндшейка закрытия
ws.terminate();
continue;
}
// Сбрасываем флаг в false и отправляем пинг-фрейм (Opcode: 0x9)
ws.isAlive = false;
ws.ping();
}
}, HEARTBEAT_INTERVAL);
wss.on("close", () => {
clearInterval(interval);
});
server.listen(8080, () => {
console.log("[WS Server] Сервер запущен на порту 8080");
});Одна из главных причин падения WebSocket серверов под нагрузкой — накопление буфера отправки (socket.bufferedAmount). Если клиент находится на медленном мобильном канале (3G), а сервер непрерывно отправляет ему сообщения, операционная система не может мгновенно сбросить данные в сетевую карту. Буфер Node.js начинает лавинообразно потреблять память процесса, приводя к Out Of Memory (OOM).
import { WebSocket } from "ws";
// Порог безопасности буфера сокета (например, 512 КБ)
const HIGH_WATERMARK = 512 * 1024;
/**
* Безопасная отправка данных с учетом Backpressure клиента
*/
export async function sendWithBackpressure(
ws: WebSocket,
data: string | Buffer
): Promise<boolean> {
if (ws.readyState !== WebSocket.OPEN) {
return false;
}
// Если буфер сокета переполнен, ждем его освобождения
if (ws.bufferedAmount >= HIGH_WATERMARK) {
console.warn(`[Backpressure Alert] Буфер сокета переполнен: ${ws.bufferedAmount} байт`);
// Даем клиенту время вычитать буфер
const isDrained = await waitForDrain(ws, 5000);
if (!isDrained) {
console.error("[Backpressure Drop] Клиент не вычитывает сокет. Завершение соединения.");
ws.terminate();
return false;
}
}
ws.send(data);
return true;
}
function waitForDrain(ws: WebSocket, timeoutMs: number): Promise<boolean> {
return new Promise((resolve) => {
const startTime = Date.now();
const checkInterval = setInterval(() => {
// Если клиент отключился
if (ws.readyState !== WebSocket.OPEN) {
clearInterval(checkInterval);
resolve(false);
return;
}
// Буфер опустился ниже половины порога безопасности
if (ws.bufferedAmount < HIGH_WATERMARK / 2) {
clearInterval(checkInterval);
resolve(true);
return;
}
// Превышен допустимый таймаут ожидания
if (Date.now() - startTime > timeoutMs) {
clearInterval(checkInterval);
resolve(false);
}
}, 50);
});
}import Fastify, { FastifyRequest, FastifyReply } from "fastify";
const app = Fastify();
interface QueryParams {
prompt: string;
}
app.get(
"/api/v1/ai/completions/stream",
async (request: FastifyRequest<{ Querystring: QueryParams }>, reply: FastifyReply) => {
const { prompt } = request.query;
if (!prompt) {
reply.status(400).send({ error: "Параметр prompt обязателен" });
return;
}
// 1. Установка обязательных заголовков для Server-Sent Events
reply.raw.writeHead(200, {
"Content-Type": "text/event-stream; charset=utf-8",
"Cache-Control": "no-cache, no-transform",
"Connection": "keep-alive",
"X-Accel-Buffering": "no", // Критично для NGINX: запрещает прокси буферизовать стрим
});
// 2. Отправка начального события с конфигурацией переподключения
reply.raw.write("retry: 3000\n\n");
let eventId = 0;
const tokens = ["Архитектура", " ", "высоконагруженных", " ", "систем", " ", "в", " ", "Node.js."];
// 3. Потоковая генерация и сброс чанков
const streamInterval = setInterval(() => {
if (eventId >= tokens.length) {
// Отправка кастомного события завершения генерации
reply.raw.write(`id: ${eventId}\nevent: done\ndata: [DONE]\n\n`);
clearInterval(streamInterval);
reply.raw.end();
return;
}
const token = tokens[eventId];
eventId++;
const payload = JSON.stringify({ token, index: eventId });
// Форматирование спецификации SSE: id, event, data
reply.raw.write(`id: ${eventId}\nevent: token\ndata: ${payload}\n\n`);
}, 150);
// 4. Очистка ресурсов при закрытии вкладки или обрыве соединения клиентом
request.raw.on("close", () => {
clearInterval(streamInterval);
console.log("[SSE] Клиент разорвал соединение со стримом");
});
}
);
app.listen({ port: 3001, host: "0.0.0.0" }, () => {
console.log("[SSE Service] Сервер стриминга запущен на порту 3001");
});При горизонтальном масштабировании (несколько подов в Kubernetes за балансировщиком нагрузки) клиенты одной и той же комнаты могут быть подключены к разным физическим серверам. Для их синхронизации используется распределенная шина сообщений.
Клиент А отправляет сообщение в локальный WebSocket-сокет на Pod 1.
Pod 1 сериализует сообщение и публикует его в Redis канал 'channel:room_42'.
Redis распределяет сообщение по всем подписчикам (Pod 1, Pod 2, Pod N), подписанным на данный канал.
Pod 2 получает сообщение от Redis и рассылает его подключенным к нему локальным сокетам (Клиенту Б).
import { WebSocketServer, WebSocket } from "ws";
import Redis from "ioredis";
export interface BroadcastMessage {
channel: string;
senderId: string;
event: string;
data: unknown;
}
export class DistributedWebSocketBroker {
private publisher: Redis;
private subscriber: Redis;
private wss: WebSocketServer;
private localSubscriptions: Map<string, Set<WebSocket>> = new Map();
constructor(wss: WebSocketServer, redisUri: string) {
this.wss = wss;
// В Redis клиент, переведенный в режим подписки (SUBSCRIBE),
// не может выполнять обычные команды. Требуется два соединения.
this.publisher = new Redis(redisUri, { lazyConnect: false });
this.subscriber = new Redis(redisUri, { lazyConnect: false });
this.initRedisSubscriber();
}
private initRedisSubscriber() {
this.subscriber.on("message", (channel: string, messageStr: string) => {
try {
const parsed: BroadcastMessage = JSON.parse(messageStr);
this.broadcastLocal(channel, parsed);
} catch (err) {
console.error("[Redis Sub Error] Ошибка парсинга сообщения:", err);
}
});
}
public subscribeLocal(channel: string, ws: WebSocket) {
if (!this.localSubscriptions.has(channel)) {
this.localSubscriptions.set(channel, new Set());
// Подписываем серверный инстанс на канал в Redis
this.subscriber.subscribe(channel);
}
this.localSubscriptions.get(channel)!.add(ws);
}
public unsubscribeLocal(channel: string, ws: WebSocket) {
const clients = this.localSubscriptions.get(channel);
if (!clients) return;
clients.delete(ws);
if (clients.size === 0) {
this.localSubscriptions.delete(channel);
// Если на этой ноде больше нет слушателей канала, отписываемся в Redis
this.subscriber.unsubscribe(channel);
}
}
public async publishToCluster(message: BroadcastMessage): Promise<void> {
const payload = JSON.stringify(message);
await this.publisher.publish(message.channel, payload);
}
private broadcastLocal(channel: string, message: BroadcastMessage) {
const clients = this.localSubscriptions.get(channel);
if (!clients) return;
const payload = JSON.stringify(message);
for (const client of clients) {
if (client.readyState === WebSocket.OPEN) {
client.send(payload);
}
}
}
public async destroy(): Promise<void> {
await this.subscriber.quit();
await this.publisher.quit();
}
}Почему при потере связи на мобильном клиенте метод ws.close() недостаточен и как гарантированно освободить ресурсы сервера?
Если полагаться только на событие сокета ws.on('close'), сервер никогда не узнает об обрыве связи при потере сети на мобильном устройстве (например, при входе в лифт или туннель). Без явного TCP FIN сокет остается в памяти ядра и процесса вечно. Регулярная отправка ws.ping() с последующим вызовом ws.terminate() для неответивших сокетов обязательна.
Массовая рассылка сообщений (broadcast) через цикл for (const client of clients) client.send(data) без проверки client.bufferedAmount. Если один из клиентов подключен через узкий канал связи, память процесса Node.js заполняется неотправленными буферами до тех пор, пока процесс не будет принудительно завершен операционной системой из-за OOM (Out Of Memory).
Использование расширения permessage-deflate в библиотеке ws сжимает каждый отдельный фрейм алгоритмом Deflate (Zlib). Это накладывает огромную вычислительную нагрузку на CPU: при 10 000 сообщений в секунду однопоточный цикл событий Node.js полностью парализуется синхронным или микрозадачным сжатием коротких строк, увеличивая задержку в сотни раз.
При развертывании SSE-серверов за обратным прокси NGINX без отключения буферизации (proxy_buffering off; или заголовка X-Accel-Buffering: no), NGINX накапливает чанки стрима в системном буфере (обычно 4–16 КБ) и сбрасывает их клиенту только после заполнения. В результате пользователь видит не плавный реалтайм-стрим, а задержки по несколько секунд и отдачу информации крупными блоками.