Організація двонаправленого та потокового обміну даними між клієнтом і бекендом вимагає переходу від класичної моделі «запит-відповідь» (Request-Response) до постійних довгоживучих з'єднань (Persistent Connections). В екосистемі Node.js для цих цілей застосовуються два стандарти: WebSocket (RFC 6455) для повнодуплексного обміну та Server-Sent Events (SSE, рекомендація W3C) для односпрямованого реактивного стрімінгу.
Порівняння мережевого стеку 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]──> ClientСесія WebSocket починається як стандартне 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: (ім'я кастомної події для обробника в браузері), retry: (таймаут перепідключення у мс) та data: (корисне навантаження).EventSource підтримує автоматичний реконнект із передачею заголовка Last-Event-ID для донадсилання втрачених подій.Проектування систем реального часу вимагає контролю життєвого циклу сокетів, виявлення обривів у мобільних мережах та керування буферами пам'яті процесу.
HTTP Upgrade Half-Open сокет Механізм Heartbeat Зворотний тиск сокета (Socket Backpressure) Розподілений адаптер Pub/SubГоризонтальне масштабування кластера 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")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
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 секунд
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 та надсилаємо кадр Ping (Opcode: 0x9)
ws.isAlive = false;
ws.ping();
}
}, HEARTBEAT_INTERVAL);
wss.on("close", () => {
clearInterval(interval);
});
server.listen(8080, () => {
console.log("[WS Server] Сервер слухає порт 8080");
});Неконтрольоване зростання буфера сокета (socket.bufferedAmount) є частою причиною збоїв WebSocket-серверів. Якщо клієнт підключений через повільну мобільну мережу, а сервер надсилає потокові дані, ядро ОС не встигає скидати пакети. Буфер пам'яті Node.js переповнюється, викликаючи аварійне падіння процесу через вичерпання пам'яті (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 за балансувальником клієнти однієї кімнати можуть підключатися до різних фізичних інстансів. Для їхньої синхронізації застосовується спільна шина повідомлень.
Клієнт А надсилає повідомлення у локальний сокет на 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.on('close') залишає сервер сліпим до раптових обривів мобільного зв'язку. Без сигналу TCP FIN сокет залишається відкритим у пам'яті процесу та ядра. Регулярна відправка ws.ping() із примусовим викликом ws.terminate() для мовчазних сокетів є обов'язковою.
Масова розсилка через for (const client of clients) client.send(data) без перевірки client.bufferedAmount. Якщо один клієнт опиняється на повільному каналі зв'язку, буфер пам'яті Node.js переповнюється невідправленими пакетами, доки операційна система не завершить процес через Out Of Memory (OOM).
Використання розширення permessage-deflate у бібліотеці ws змушує стискати кожен малий фрейм алгоритмом zlib. При 10 000+ повідомлень на секунду однопоточний Event Loop Node.js повністю блокується синхронним стисненням, збільшуючи затримки у сотні разів.
Розгортання SSE-сервісів за NGINX без вимкнення буферизації (proxy_buffering off; або заголовка X-Accel-Buffering: no). NGINX накопичує фрагменти потоку у власному буфері (4–16 КБ) і відправляє їх лише після заповнення, перетворюючи плавний стрім на затримані пачки токенів.
Клієнт надсилає заголовок Upgrade: websocket із випадковим ключем Sec-WebSocket-Key. Сервер верифікує токен.
Сервер повертає хеш у Sec-WebSocket-Accept. Мережевий сокет перемикається на бінарний протокол.
Двонаправлений обмін фреймами. Клієнтські фрейми маскуються 4-байтовим XOR-ключем.
Сервер що 30 секунд надсилає Ping. Сокети без відповіді Pong знищуються через terminate().
Сторони обмінюються кадрами закриття Opcode 0x8 з кодом статусу та коректно завершують з'єднання.