websocket-realtime

WebSocket 与实时通信

WebSocket 服务端

import { WebSocketServer, WebSocket } from "ws";

const wss = new WebSocketServer({ port: 8080 });

const rooms = new Map<string, Set<WebSocket>>();

wss.on("connection", (ws, req) => {
  const userId = authenticateFromUrl(req.url);
  if (!userId) {
    ws.close(4001, "Unauthorized");
    return;
  }

  ws.on("message", (data) => {
    const message = JSON.parse(data.toString());

    switch (message.type) {
      case "join":
        joinRoom(message.room, ws);
        break;
      case "leave":
        leaveRoom(message.room, ws);
        break;
      case "broadcast":
        broadcastToRoom(message.room, message.payload, ws);
        break;
    }
  });

  ws.on("close", () => {
    rooms.forEach((members) => members.delete(ws));
  });

  ws.send(JSON.stringify({ type: "connected", userId }));
});

function joinRoom(room: string, ws: WebSocket) {
  if (!rooms.has(room)) rooms.set(room, new Set());
  rooms.get(room)!.add(ws);
}

function broadcastToRoom(room: string, payload: unknown, sender: WebSocket) {
  const members = rooms.get(room);
  if (!members) return;
  const message = JSON.stringify({ type: "message", room, payload });
  members.forEach((client) => {
    if (client !== sender && client.readyState === WebSocket.OPEN) {
      client.send(message);
    }
  });
}

使用房间的 Socket.io

import { Server } from "socket.io";
import { createAdapter } from "@socket.io/redis-adapter";
import { createClient } from "redis";

const io = new Server(httpServer, {
  cors: { origin: "https://app.example.com" },
  pingTimeout: 20000,
  pingInterval: 25000,
});

const pubClient = createClient({ url: "redis://localhost:6379" });
const subClient = pubClient.duplicate();
await Promise.all([pubClient.connect(), subClient.connect()]);
io.adapter(createAdapter(pubClient, subClient));

io.use(async (socket, next) => {
  const token = socket.handshake.auth.token;
  try {
    socket.data.user = verifyToken(token);
    next();
  } catch {
    next(new Error("Authentication failed"));
  }
});

io.on("connection", (socket) => {
  socket.join(`user:${socket.data.user.id}`);

  socket.on("chat:join", (roomId) => {
    socket.join(`chat:${roomId}`);
    socket.to(`chat:${roomId}`).emit("chat:userJoined", socket.data.user);
  });

  socket.on("chat:message", async ({ roomId, text }) => {
    const message = await saveMessage(roomId, socket.data.user.id, text);
    io.to(`chat:${roomId}`).emit("chat:message", message);
  });

  socket.on("disconnect", () => {
    console.log(`User ${socket.data.user.id} disconnected`);
  });
});

服务端发送事件（SSE）

app.get("/events/:userId", authenticate, (req, res) => {
  res.writeHead(200, {
    "Content-Type": "text/event-stream",
    "Cache-Control": "no-cache",
    Connection: "keep-alive",
  });

  const sendEvent = (event: string, data: unknown) => {
    res.write(`event: ${event}\n`);
    res.write(`data: ${JSON.stringify(data)}\n\n`);
  };

  sendEvent("connected", { userId: req.params.userId });

  const interval = setInterval(() => {
    res.write(":heartbeat\n\n");
  }, 30000);

  const listener = (message: string) => {
    const event = JSON.parse(message);
    sendEvent(event.type, event.data);
  };

  redis.subscribe(`user:${req.params.userId}`, listener);

  req.on("close", () => {
    clearInterval(interval);
    redis.unsubscribe(`user:${req.params.userId}`, listener);
  });
});

对于服务端到客户端的单向流式传输，SSE 比 WebSocket 更简单。无需特殊配置即可通过 HTTP 代理和负载均衡器工作。

客户端重新连接

class ReconnectingWebSocket {
  private ws: WebSocket | null = null;
  private retryCount = 0;
  private maxRetries = 10;

  constructor(private url: string) {
    this.connect();
  }

  private connect() {
    this.ws = new WebSocket(this.url);
    this.ws.onopen = () => { this.retryCount = 0; };
    this.ws.onclose = () => { this.scheduleReconnect(); };
    this.ws.onerror = () => { this.ws?.close(); };
  }

  private scheduleReconnect() {
    if (this.retryCount >= this.maxRetries) return;
    const delay = Math.min(1000 * 2 ** this.retryCount, 30000);
    this.retryCount++;
    setTimeout(() => this.connect(), delay);
  }

  send(data: string) {
    if (this.ws?.readyState === WebSocket.OPEN) {
      this.ws.send(data);
    }
  }
}

反模式

• 在握手期间不对 WebSocket 连接进行身份验证
• 发送不受限制的载荷，且不限制消息大小
• 缺少用于检测失效连接的心跳／ping-pong 机制
• 在 SSE 足以满足需求时仍使用 WebSocket（仅服务端到客户端）
• 使用 Socket.io 进行水平扩展时未使用 Redis 适配器
• 通过同步处理消息阻塞事件循环

检查清单

• 在握手期间对 WebSocket 连接进行身份验证
• 对传入数据强制执行消息大小限制
• 心跳机制能够检测并关闭失效连接
• 客户端实现指数退避重新连接
• 多服务端部署使用 Redis 发布／订阅适配器
• 仅需服务端到客户端通信时使用 SSE
• 断开连接时清理房间／频道成员关系
• 应用速率限制以防止消息泛滥