WebSocket vs SSE vs Long Polling: выбираем реалтайм в 2025

# ❌ Polling каждые 2 секунды с фронтенда
@app.get("/messages")
async def get_messages(last_id: int = 0):
    return await db.query(
        "SELECT * FROM messages WHERE id > ?", last_id
    )

# Клиент
ws = new WebSocket('ws://localhost:8000/ws')
ws.send('Hello')              // Клиент → Сервер
ws.onmessage = (msg) => {...} // Сервер → Клиент
 
# Сервер (FastAPI)
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    await websocket.send_text("Welcome!")  # Сервер → Клиент
    data = await websocket.receive_text()  # Клиент → Сервер

# Клиент
const eventSource = new EventSource('/sse')
eventSource.onmessage = (event) => {
  console.log(event.data) // Только получение
}
 
# Сервер (FastAPI)
@app.get("/sse")
async def sse_endpoint(request: Request):
    async def event_stream():
        while True:
            yield f"data: {datetime.now()}\n\n"
            await asyncio.sleep(1)
 
    return StreamingResponse(
        event_stream(),
        media_type="text/event-stream"
    )

# Клиент
async function poll() {
  const response = await fetch('/poll?last_id=123')
  const data = await response.json()
  processData(data)
  poll() // Сразу следующий запрос
}
 
# Сервер (FastAPI)
@app.get("/poll")
async def long_poll(last_id: int = 0):
    # Ждём до 30 секунд новых данных
    for _ in range(30):
        new_data = await get_new_messages(last_id)
        if new_data:
            return new_data
        await asyncio.sleep(1)
 
    return []  # Timeout, клиент повторит

location /ws {
    proxy_pass http://backend;
    proxy_http_version 1.1;
    proxy_set_header Upgrade $http_upgrade;
    proxy_set_header Connection "upgrade";
 
    # Увеличили до 1 часа
    proxy_read_timeout 3600s;
    proxy_send_timeout 3600s;
}

# Сервер шлёт ping каждые 30 секунд
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
 
    async def send_ping():
        while True:
            try:
                await websocket.send_text("ping")
                await asyncio.sleep(30)
            except:
                break
 
    asyncio.create_task(send_ping())
 
    # ... обработка сообщений

ws.onmessage = (event) => {
  if (event.data === "ping") {
    ws.send("pong"); // Ответ на heartbeat
    return;
  }
  // обработка реальных сообщений
};

location /sse {
    proxy_pass http://backend;
 
    # Отключаем буферизацию для SSE
    proxy_buffering off;
    proxy_cache off;
 
    # Обязательно для SSE
    proxy_set_header Connection '';
    proxy_http_version 1.1;
    chunked_transfer_encoding off;
 
    # HTTP/1.1 keep-alive
    proxy_read_timeout 86400s;
}

from starlette.responses import StreamingResponse
 
@app.get("/sse")
async def sse(request: Request):
    async def event_generator():
        while True:
            if await request.is_disconnected():
                break
 
            # Формат SSE: "data: <payload>\n\n"
            yield f"data: {json.dumps({'time': time.time()})}\n\n"
            await asyncio.sleep(1)
 
    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",  # Для Nginx
        }
    )

# ❌ Плохо: Sync workers
gunicorn app:app --workers 4 --worker-class sync
 
# ✅ Хорошо: Async workers (gevent или uvicorn)
gunicorn app:app --workers 4 --worker-class uvicorn.workers.UvicornWorker

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import List
import json
 
app = FastAPI()
 
# Менеджер подключений
class ConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []
 
    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)
 
    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)
 
    async def broadcast(self, message: str):
        """Отправка всем клиентам"""
        for connection in self.active_connections:
            try:
                await connection.send_text(message)
            except:
                # Если отправка упала — удалить соединение
                self.disconnect(connection)
 
manager = ConnectionManager()
 
@app.websocket("/ws/chat")
async def websocket_chat(websocket: WebSocket):
    await manager.connect(websocket)
 
    try:
        while True:
            # Получаем сообщение от клиента
            data = await websocket.receive_text()
 
            message = json.loads(data)
            username = message.get("username", "Anonymous")
            text = message.get("text", "")
 
            # Broadcast всем
            await manager.broadcast(json.dumps({
                "username": username,
                "text": text,
                "timestamp": time.time()
            }))
 
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(json.dumps({
            "system": f"User left the chat"
        }))

const ws = new WebSocket("ws://localhost:8000/ws/chat");
 
ws.onopen = () => console.log("Connected");
 
ws.onmessage = (event) => {
  const message = JSON.parse(event.data);
  addMessageToUI(message);
};
 
// Отправка сообщения
function sendMessage(text) {
  ws.send(
    JSON.stringify({
      username: "User123",
      text: text,
    })
  );
}
 
// Reconnect при обрыве
ws.onclose = () => {
  console.log("Disconnected, reconnecting...");
  setTimeout(() => location.reload(), 1000);
};

from fastapi import FastAPI, Request
from sse_starlette.sse import EventSourceResponse
import asyncio
import psutil
import json
 
app = FastAPI()
 
async def generate_metrics():
    """Генератор метрик каждую секунду"""
    while True:
        metrics = {
            "cpu_percent": psutil.cpu_percent(interval=1),
            "memory_percent": psutil.virtual_memory().percent,
            "timestamp": time.time()
        }
 
        # Формат SSE
        yield {
            "event": "metrics",
            "data": json.dumps(metrics)
        }
 
        await asyncio.sleep(1)
 
@app.get("/sse/metrics")
async def sse_metrics(request: Request):
    """SSE endpoint для метрик"""
 
    async def event_stream():
        try:
            async for event in generate_metrics():
                # Проверяем, жив ли клиент
                if await request.is_disconnected():
                    break
 
                # Формат SSE: event: <type>\ndata: <payload>\n\n
                yield f"event: {event['event']}\n"
                yield f"data: {event['data']}\n\n"
 
        except asyncio.CancelledError:
            # Клиент отключился
            pass
 
    return EventSourceResponse(
        event_stream(),
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",
        }
    )

const eventSource = new EventSource("/sse/metrics");
 
eventSource.addEventListener("metrics", (event) => {
  const data = JSON.parse(event.data);
 
  updateChart("cpu", data.cpu_percent);
  updateChart("memory", data.memory_percent);
});
 
eventSource.onerror = () => {
  console.error("SSE connection failed");
  eventSource.close();
 
  // Reconnect через 5 секунд
  setTimeout(() => location.reload(), 5000);
};

// Получаем метрики через SSE
const eventSource = new EventSource("/sse/metrics");
 
// Отправляем команды через fetch
async function restartService() {
  await fetch("/api/restart", { method: "POST" });
}

from fastapi import FastAPI
import asyncio
 
app = FastAPI()
 
# In-memory очередь уведомлений (в проде — Redis/DB)
notifications_queue = []
 
@app.get("/poll/notifications")
async def poll_notifications(last_id: int = 0):
    """
    Long polling: ждём до 30 секунд новых уведомлений
    """
    timeout_seconds = 30
 
    for _ in range(timeout_seconds):
        # Проверяем новые уведомления
        new_notifications = [
            n for n in notifications_queue
            if n["id"] > last_id
        ]
 
        if new_notifications:
            return {"notifications": new_notifications}
 
        # Ждём 1 секунду перед следующей проверкой
        await asyncio.sleep(1)
 
    # Timeout — возвращаем пустой ответ
    return {"notifications": []}
 
@app.post("/notifications")
async def create_notification(text: str):
    """Создать уведомление (для теста)"""
    notification = {
        "id": len(notifications_queue) + 1,
        "text": text,
        "timestamp": time.time()
    }
    notifications_queue.append(notification)
    return notification

let lastNotificationId = 0;
 
async function pollNotifications() {
  try {
    const response = await fetch(
      `/poll/notifications?last_id=${lastNotificationId}`
    );
    const data = await response.json();
 
    if (data.notifications.length > 0) {
      data.notifications.forEach(showNotification);
      lastNotificationId = Math.max(...data.notifications.map((n) => n.id));
    }
  } catch (error) {
    console.error("Polling failed:", error);
    // Backoff при ошибке
    await new Promise((resolve) => setTimeout(resolve, 5000));
  }
 
  // Сразу следующий запрос
  pollNotifications();
}
 
// Старт
pollNotifications();

import aioredis
 
redis = aioredis.from_url("redis://localhost")
 
@app.get("/poll/notifications")
async def poll_notifications(user_id: int, last_id: int = 0):
    """Long polling через Redis Pub/Sub"""
 
    # Подписываемся на канал пользователя
    pubsub = redis.pubsub()
    await pubsub.subscribe(f"user:{user_id}:notifications")
 
    # Ждём событие или timeout
    try:
        async with asyncio.timeout(30):
            async for message in pubsub.listen():
                if message["type"] == "message":
                    notification = json.loads(message["data"])
                    return {"notifications": [notification]}
 
    except asyncio.TimeoutError:
        return {"notifications": []}
 
    finally:
        await pubsub.unsubscribe()
 
# При создании уведомления — publish в Redis
@app.post("/notifications")
async def create_notification(user_id: int, text: str):
    notification = {"id": generate_id(), "text": text}
 
    # Публикуем в канал пользователя
    await redis.publish(
        f"user:{user_id}:notifications",
        json.dumps(notification)
    )
 
    return notification

from fastapi import FastAPI
 
app = FastAPI()
 
# WebSocket для чата support
@app.websocket("/ws/support")
async def support_chat(websocket: WebSocket):
    # Двусторонний чат с операторами
    ...
 
# SSE для real-time метрик
@app.get("/sse/metrics")
async def metrics_stream(request: Request):
    # Обновление графиков каждые 5 секунд
    ...
 
# Long Polling для редких уведомлений
@app.get("/poll/notifications")
async def poll_notifications(last_id: int):
    # Уведомления о завершении задач
    ...
 
# Обычный REST для всего остального
@app.get("/api/users")
async def get_users():
    ...

from prometheus_client import Counter, Histogram
 
ws_connections = Counter('ws_connections_total', 'WebSocket connections')
ws_messages = Counter('ws_messages_total', 'WebSocket messages sent')
ws_latency = Histogram('ws_message_latency_seconds', 'Message latency')
 
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    ws_connections.inc()
 
    start = time.time()
    await websocket.send_text("hello")
    ws_latency.observe(time.time() - start)
 
    ws_messages.inc()

import logging
 
logger = logging.getLogger(__name__)
 
@app.get("/sse")
async def sse(request: Request):
    client_id = request.client.host
    logger.info(f"SSE connected: {client_id}")
 
    async def event_stream():
        try:
            while True:
                yield f"data: {time.time()}\n\n"
                await asyncio.sleep(1)
        except Exception as e:
            logger.error(f"SSE error for {client_id}: {e}")
        finally:
            logger.info(f"SSE disconnected: {client_id}")
 
    return StreamingResponse(event_stream(), media_type="text/event-stream")