Потоковая трансляция данных с Server-Sent Events: забытое звено реального времени

Категория: Статьи • 2 July 2025 • 6 мин чтения
mermaid
graph LR
    A[Client] -->|HTTP GET| B[SSE Server]
    B -->|Event Stream| A
    B -->|Live Updates| A
    B -->|Automatic Reconnect| A
    B -->|Error Flow| A

Беспощадный запрос от клиента: "Дайте мне самые свежие данные". Но что делать, когда эти данные постоянно меняются? Многие сразу тянутся к WebSockets, забывая о существовании элегантного, специфичного подхода для однонаправленного потока данных: Server-Sent Events (SSE).

SSE - это HTTP-протокол, при котором сервер отправляет клиенту обновления в "длинном" потоке без инициализации новых запросов после начала сеанса. В отличие от сложности WebSockets, SSE остаются в экосистеме HTTP - им не нужны отдельные протоколы или порты.

Одна из главных трагедий в веб-разработке - постоянное использование неподходящего решения. Хотите отправлять уведомления, обновлять биржевые котировки или реактивно отображать прогресс бэкенд-операции? SSE - ваш инструмент.

Архитектурный срез SSE:

js
// Типичный фрагмент потока событий
event: statusUpdate
id: 456
retry: 3000
data: {"taskId":"abc", "progress":75}

event: error
data: Connection dropped

Анатомия события:

  • event: тип события (кастомное событие)
  • id: идентификатор для поддержания порядка и реконнекта
  • retry: время паузы перед попыткой восстановления
  • data: сами данные

Практикум: бэкенд на Node.js

Создадим поток событий с ограничением пропускной способности и управлением множественными подключениями.

javascript
// server.js
import express from 'express';
import http from 'http';

const app = express();
const server = http.createServer(app);

// Хранилище активных соединений
const clients = new Map();

app.get('/events/:channel', (req, res) => {
    const { channel } = req.params;
    const clientId = Date.now();

    res.set({
        'Content-Type': 'text/event-stream',
        'Cache-Control': 'no-cache',
        'Connection': 'keep-alive',
        // Критически важно для NGINX и прокси
        'X-Accel-Buffering': 'no' 
    });
    res.flushHeaders();

    // Сохраняем ссылку на ответ
    clients.set(clientId, { res, channel });

    // Принудительная отправка комментария для поддержания активности
    res.write(': keep-alive\n\n');

    const params = req.query;
    console.log(`Client ${clientId} connected to ${channel}`, params);

    req.on('close', () => {
        clients.delete(clientId);
        console.log(`Client ${clientId} disconnected`);
    });
});

// Сервис для отправки событий
export function sendEvent({ channel, event = 'message', data }) {
    for (const [id, client] of clients) {
        if (client.channel === channel) {
            try {
                client.res.write(
                    `event: ${event}\nid: ${Date.now()}\nretry: 5000\ndata: ${JSON.stringify(data)}\n\n`
                );
            } catch (e) {
                clients.delete(id);
                console.error(`Не удалось отправить событие клиенту ${id}: ${e.message}`);
            }
        }
    }
}

// Пример генератора событий с анализом загрузки канала
function generateEvents() {
    const channels = ['currency', 'stocks', 'logs'];
    channels.forEach(channel => {
        sendEvent({
            channel,
            data: { timestamp: Date.now(), value: Math.random() * 100 }
        });
    });
}

// Отправляем события каждые 2 секунды
setInterval(generateEvents, 2000);

const PORT = 3001;
server.listen(PORT, () => {
    console.log(`SSE server live on port ${PORT}`);
});

Ключевые моменты реализации:

  • Каналы событий: разделение потоков с помощью виртуальных каналов (из URL)
  • Управление памятью: автоматическое удаление закрытых соединений
  • Асинхронные ограничения: избежаны блокировки Event Loop через коллектор ресурсов
  • Прокси-совместимость: заголовки специально для корпоративных сред

Фронтенд-потребитель: больше чем addEventListener

javascript
// Сезонный обработчик с восстановлением и изоляцией ошибок
function setupEventSource(url, handlers) {
    const eventSource = new EventSource(url);
    
    // Пересоздаст соединение после таймаута
    let reconnectTimeout;
    let lastEventId = null;
    let isShutdown = false;
    
    eventSource.addEventListener('message', (e) => {
        if (e.lastEventId) lastEventId = e.lastEventId;
        try {
            handlers.onMessage(JSON.parse(e.data));
        } catch (parseError) {
            handlers.onError?.(parseError);
        }
    });
    
    // Кастомные обработчики событий
    Object.entries(handlers)
        .filter(([name]) => name.startsWith('on'))
        .forEach(([name, fn]) => {
            if (name !== 'onMessage') {
                eventSource.addEventListener(
                    name.substring(2).toLowerCase(), 
                    e => fn(JSON.parse(e.data))
                );
            }
        });

    eventSource.onerror = function () {
        handlers.onError?.(new Error('EventSource failed'));
        
        if (eventSource.readyState === EventSource.CLOSED) {
            // Zакрыто явным вызовом .close()
            if (isShutdown) return;
            
            // Ждем перед восстановлением соединения
            clearTimeout(reconnectTimeout);
            reconnectTimeout = setTimeout(
                () => {
                    if (!isShutdown) setupEventSource(
                        lastEventId ? `${url}?lastEventId=${lastEventId}` : url,
                         listeners
                    );
                },
                1000 * Math.random() * 5 + 1000 // Рандомизация интервала
            );
        }
    };
    
    // API для явного закрытия
    return function shutdown() {
        isShutdown = true;
        clearTimeout(reconnectTimeout);
        eventSource.close();
    };
}

// Инициализация потребителя событий курсов валют
const closeCurrencyStream = setupEventSource('http://localhost:3001/events/currency', {
    onMessage: data => updateUI(data.value),
    onCriticalAlert: msg => showToast(msg),
    onError: err => console.error('SSE pipeline failure:', err)
});

Таким образом, мы получаем:

  • Реконнект со случайной экспоненциальной задержкой - не заваливаем сервер после сбоя
  • Мульти-обработчики событий - как системных, так и кастомных
  • Поддержка Last-Event-ID - для синхронизации после потери данных
  • Чистое завершение соединения - избегаем утечек памяти при ресинга событий

Ограничения и обходные пути

Идеальных технологий не существует. Зная ограничения SSE, создадим стратегии для компенсации.

Предельное число подключений В отличие от WebSockets, для каждого SSE-клиента на сервере создается отдельный тред/обработчик. Стандартный Node.js glasgow pair на одном ядре под нагрузкой обслуживает примерно 6 тысяч подключений. Масштабируемые решения требуют кластеризации и распределенного хранилища для передачи событий между экземплярами.

javascript
// Муфтованный сервер с Redis для распространения событий
import { createClient } from 'redis';

const pubClient = createClient();
const subClient = pubClient.duplicate();

await Promise.all([pubClient.connect(), subClient.connect()]);

// Подписываемся на канал Redis с событиями
subClient.subscribe('global-events', (message) => {
    const { channel, event, data } = JSON.parse(message);
    sendEvent({ channel, event, data });
});

// При публикации события сначала пишем в Redis
function publishEvent(channel, event, data) {
    pubClient.publish('global-events', 
        JSON.stringify({ channel, event, data })
    );
}

Прокси и кэширование Прокси (Nginx, Cloudflare) часто разрывают "длинные" соединения до таймоута. Конфиг Nginx должен содержать:

nginx
server {
    location /events {
        proxy_pass http://backend;
        proxy_http_version 1.1;
        proxy_set_header Connection '';
        proxy_buffering off;
        proxy_cache off;
        proxy_read_timeout 24h;
    }
}

Авторизация и заголовки Поскольку SSE работают через GET-запросы, ограничьте чувствительные данные в URL. Джедайский приём - заставить SSE работать с куками:

javascript
// Защита событий через сессию и ленивую авторизацию
app.use('/events/:channel', authenticateUser);

function authenticateUser(req, res, next) {
    const sessionId = req.headers.cookie?.match(/sessionId=(\w+)/)?.[1];
    if (!sessionId) return res.status(401).end();

    const user = db.findSession(sessionId);
    if (!user || !user.canAccess(req.params.channel)) 
        return res.status(403).end();

    next();
}

Критические операции: логи и диагностика потоков

Подписываясь на поток событий, выделяйте логирование потоковой инфраструктуры:

bash
# База данных событий с водяными знаками времени
timestamp | clientID | eventType | beforeID | afterID | payloadLen
---------------------------------------------------------------
1687810200 | abx     | connect   |     -     |    -    |      0
1687810205 | abx     | message   |     10    |    11   |    142
1687810209 | abx     | resume    |     11    |    12   |      0

Такие метрики позволяют:

  • Искать причину потери данных по результатам после resume
  • Отслеживать лавинообразный прирост объема (payloadLen)
  • Детектировать MIMБ с анализом расхождения через clientID

Когда выбирать другую технологию?

Есть анти-кейсы применения для SSE - не всех проблем они решают идеально:

  1. Двухсторонняя связь, мультидиалог: сохраняемость сессии через установку состояния Решение: WebSockets

  2. Большой массив данных за 1 раз: XMLHttpRequest и multipart хуже в сравнении с... Решение: WebTransport API

  3. Один раз в миллион лет: при нормативном требовании совместимости с IE Решение: Long polling с таймаутом и серверами размером с Бегемот

  4. Строгие гарантии доставки: операционная система не простит потери одиночного пакета с SSE Решение: MQTT/WMQ

Ну что, поставляем?

Server-Sent Events обладают уникальной ценностью для решения задач, не требующих симметрии. При условии понимания ограничений, этот подход создаст вам систему обновлений с минимальным бубном. Реализация может быть компактной как на бэкенде, так и на фронтенде, а главное — не создающим технического долга из-за излишней сложности.

SSE должны вернуться в ваши уважаемые наборы инструментария. Комбинация эффективность-простота-отказоустойчивость заслуживает серьезного исследования для микрослужбных схем, высоконагруженных информационных панелей и прогресс-барчиков с характером.