构建实时 WebSocket 聊天服务器

项目概述

构建支持多房间的实时聊天：用户加入、发送消息、看到正在输入提示，并在加入或离开时收到通知。服务端负责认证、校验、限流和广播。

架构与数据模型

Express 处理升级握手，ws 管理连接，RoomManager 维护用户与房间映射，Broadcaster 负责扇出。消息带唯一 id、用户 id、时间和类型；历史按房间保存。

实施阶段

先实现安全的升级与认证，再维护房间和历史；处理消息校验、长度限制和发送速率；加入心跳检测半开连接并在断开时清理；客户端实现指数退避重连和状态恢复；最后测试房间隔离与限流。

验证与限制

两个浏览器标签页应实时互通，关闭一个后其他页面收到离开事件。内存历史重启即丢失，多实例需要 Redis 适配器；生产还需 TLS、持久化、内容审核和可观测性。

代码示例

mkdir chat-server && cd chat-server
pnpm init
pnpm add express ws nanoid
pnpm add -D typescript tsx vitest @types/node @types/express @types/ws
npx tsc --init

import http from 'node:http';
import express from 'express';
import { WebSocketServer } from 'ws';

const app = express();
app.use(express.static('public'));
const server = http.createServer(app);
const wss = new WebSocketServer({ noServer: true, maxPayload: 64 * 1024 });

server.on('upgrade', (req, socket, head) => {
  if (!isAllowedOrigin(req.headers.origin)) {
    socket.destroy();
    return;
  }
  const session = readSession(req);
  if (!session) {
    socket.write('HTTP/1.1 401 Unauthorized\r\n\r\n');
    socket.destroy();
    return;
  }
  wss.handleUpgrade(req, socket, head, (ws) => {
    wss.emit('connection', ws, req, session);
  });
});

type Room = {
  members: Map<string, User>;
  history: Message[];
  sequence: number;
};

export class RoomManager {
  private rooms = new Map<string, Room>();

  join(roomId: string, user: User): Message[] {
    const room = this.rooms.get(roomId) ?? { members: new Map(), history: [], sequence: 0 };
    this.rooms.set(roomId, room);
    room.members.set(user.id, user);
    return [...room.history];
  }

  leave(roomId: string, userId: string): void {
    const room = this.rooms.get(roomId);
    if (!room) return;
    room.members.delete(userId);
    if (room.members.size === 0) this.rooms.delete(roomId);
  }

  append(roomId: string, message: Omit<Message, 'sequence'>): Message {
    const room = this.rooms.get(roomId);
    if (!room) throw new Error('room not found');
    room.sequence += 1;
    const stored = { ...message, sequence: room.sequence };
    room.history.push(stored);
    if (room.history.length > 100) room.history.shift();
    return stored;
  }
}

type ClientEvent =
  | { type: 'chat_message'; text: string }
  | { type: 'typing'; active: boolean };

function parseEvent(raw: string): ClientEvent {
  const value = JSON.parse(raw) as unknown;
  if (typeof value !== 'object' || value === null) throw new Error('invalid event');
  const event = value as Record<string, unknown>;
  if (event.type === 'chat_message') {
    if (typeof event.text !== 'string' || event.text.trim().length === 0) throw new Error('empty message');
    if (event.text.length > 2000) throw new Error('message too long');
    return { type: 'chat_message', text: event.text };
  }
  if (event.type === 'typing' && typeof event.active === 'boolean') {
    return { type: 'typing', active: event.active };
  }
  throw new Error('unknown event');
}

const buckets = new Map<string, { tokens: number; updated: number }>();

function allow(userId: string, capacity = 5, refillPerSecond = 2): boolean {
  const now = Date.now();
  const bucket = buckets.get(userId) ?? { tokens: capacity, updated: now };
  bucket.tokens = Math.min(capacity, bucket.tokens + ((now - bucket.updated) / 1000) * refillPerSecond);
  bucket.updated = now;
  if (bucket.tokens < 1) return false;
  bucket.tokens -= 1;
  buckets.set(userId, bucket);
  return true;
}

for (const user of room.members.values()) {
  if (user.socket.readyState !== WebSocket.OPEN) continue;
  if (user.socket.bufferedAmount > 1_000_000) continue;
  user.socket.send(JSON.stringify(event));
}

const interval = setInterval(() => {
  for (const ws of wss.clients) {
    if (ws.isAlive === false) {
      ws.terminate();
      continue;
    }
    ws.isAlive = false;
    ws.ping();
  }
}, 30_000);

wss.on('connection', (ws) => {
  ws.isAlive = true;
  ws.on('pong', () => { ws.isAlive = true; });
});

wss.on('close', () => clearInterval(interval));

const scheme = location.protocol === 'https:' ? 'wss' : 'ws';
let socket;
let attempt = 0;

function connect() {
  socket = new WebSocket(`${scheme}://${location.host}/ws?room=${encodeURIComponent(roomId)}`);
  socket.addEventListener('open', () => {
    attempt = 0;
    setStatus('live');
  });
  socket.addEventListener('message', (event) => render(JSON.parse(event.data)));
  socket.addEventListener('close', () => {
    if (socket.code === 1008 || socket.code === 4401) return;
    const delay = Math.min(30_000, 500 * 2 ** attempt) * (0.5 + Math.random());
    attempt += 1;
    setStatus('reconnecting');
    setTimeout(connect, delay);
  });
}
connect();

const server = http.createServer(app);
const wss = attachWebSocket(server, rooms, broadcaster);
await new Promise<void>((resolve) => server.listen(0, resolve));
const { port } = server.address() as AddressInfo;

const alice = new WebSocket(`ws://127.0.0.1:${port}/ws?room=general`);
await new Promise((resolve) => alice.once('open', resolve));

CREATE TABLE messages (
  id          text PRIMARY KEY,
  room_id     text NOT NULL,
  sequence    bigint NOT NULL,
  sender_id   text NOT NULL,
  body        text NOT NULL,
  created_at  timestamptz NOT NULL DEFAULT now(),
  UNIQUE (room_id, sequence)
);