websocket-engineer

WebSocket 工程师

核心工作流

1. 分析需求 — 确定连接规模、消息量和延迟要求
2. 设计架构 — 规划集群、发布/订阅、状态管理和故障转移
3. 实现 — 构建具备身份验证、房间和事件功能的 WebSocket 服务
4. 本地验证 — 在扩展规模前测试连接处理、身份验证和房间行为（例如，npx wscat -c ws://localhost:3000）；确认缺少或无效令牌时会拒绝身份验证、房间加入/离开事件正常，以及消息能够送达
5. 扩展 — 在启用适配器前验证 Redis 连接和发布/订阅往返；配置粘性会话，并通过跨多个实例的测试连接进行确认；设置负载均衡
6. 监控 — 跟踪连接数、延迟、吞吐量和错误率；针对连接数激增和错误率阈值添加告警

参考指南

根据上下文加载详细指导：

主题	参考资料	加载时机
协议	references/protocol.md	WebSocket 握手、帧、ping/pong、关闭代码
扩展	references/scaling.md	水平扩展、Redis 发布/订阅、粘性会话
模式	references/patterns.md	房间、命名空间、广播、确认
安全	references/security.md	身份验证、授权、速率限制、CORS
替代方案	references/alternatives.md	SSE、长轮询、何时选择 WebSockets

代码示例

服务端设置（使用身份验证和房间管理的 Socket.IO）

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

const httpServer = createServer();
const io = new Server(httpServer, {
  cors: { origin: process.env.ALLOWED_ORIGIN, credentials: true },
  pingTimeout: 20000,
  pingInterval: 25000,
});

// Authentication middleware — runs before connection is established
io.use((socket, next) => {
  const token = socket.handshake.auth.token;
  if (!token) return next(new Error("Authentication required"));
  try {
    socket.data.user = jwt.verify(token, process.env.JWT_SECRET);
    next();
  } catch {
    next(new Error("Invalid token"));
  }
});

// Redis adapter for horizontal scaling
const pubClient = createClient({ url: process.env.REDIS_URL });
const subClient = pubClient.duplicate();
await Promise.all([pubClient.connect(), subClient.connect()]);
io.adapter(createAdapter(pubClient, subClient));

io.on("connection", (socket) => {
  const { userId } = socket.data.user;
  console.log(`connected: ${userId} (${socket.id})`);

  // Presence: mark user online
  pubClient.hSet("presence", userId, socket.id);

  socket.on("join-room", (roomId) => {
    socket.join(roomId);
    socket.to(roomId).emit("user-joined", { userId });
  });

  socket.on("message", ({ roomId, text }) => {
    io.to(roomId).emit("message", { userId, text, ts: Date.now() });
  });

  socket.on("disconnect", () => {
    pubClient.hDel("presence", userId);
    console.log(`disconnected: ${userId}`);
  });
});

httpServer.listen(3000);

使用指数退避的客户端重连

import { io } from "socket.io-client";

const socket = io("wss://api.example.com", {
  auth: { token: getAuthToken() },
  reconnection: true,
  reconnectionAttempts: 10,
  reconnectionDelay: 1000,       // initial delay (ms)
  reconnectionDelayMax: 30000,   // cap at 30 s
  randomizationFactor: 0.5,      // jitter to avoid thundering herd
});

// Queue messages while disconnected
let messageQueue = [];

socket.on("connect", () => {
  console.log("connected:", socket.id);
  // Flush queued messages
  messageQueue.forEach((msg) => socket.emit("message", msg));
  messageQueue = [];
});

socket.on("disconnect", (reason) => {
  console.warn("disconnected:", reason);
  if (reason === "io server disconnect") socket.connect(); // manual reconnect
});

socket.on("connect_error", (err) => {
  console.error("connection error:", err.message);
});

function sendMessage(roomId, text) {
  const msg = { roomId, text };
  if (socket.connected) {
    socket.emit("message", msg);
  } else {
    messageQueue.push(msg); // buffer until reconnected
  }
}

约束

必须做到

• 使用粘性会话进行负载均衡（WebSocket 连接是有状态的——请求必须路由到同一个服务实例）
• 实现心跳/ping-pong 以检测失效连接（仅靠 TCP keepalive 不够）
• 使用房间/命名空间限定消息范围，而不是在应用逻辑中进行筛选
• 在断连期间将消息加入队列，避免数据悄无声息地丢失
• 在水平扩展前规划每个实例的连接数限制

禁止事项

• 未制定集群策略就将大量状态存储在内存中（使用 Redis 或外部存储）
• 未显式处理升级就让 WebSocket 和 HTTP 共用同一个端口
• 忘记处理连接清理（在线状态记录、房间成员关系、运行中的计时器）
• 生产环境部署前跳过负载测试——连接数激增与 HTTP 流量激增的表现不同

输出模板

实现 WebSocket 功能时，请提供：

1. 服务端设置（Socket.IO/ws 配置）
2. 事件处理程序（连接、消息、断开连接）
3. 客户端库（连接、事件、重连）
4. 扩展策略的简要说明

知识参考

Socket.IO、ws、uWebSockets.js、Redis 适配器、粘性会话、nginx WebSocket 代理、通过 WebSocket 传输 JWT、房间/命名空间、确认、二进制数据、压缩、心跳、背压、水平 Pod 自动扩缩容

由 @jeffallan 维护，他是 Synergetic Solutions 的首席顾问

文档