全栈实时通信工程实践：技术选型、连接治理与架构取舍

// server.ts
import { WebSocketServer, WebSocket } from 'ws';

interface ExtWebSocket extends WebSocket {
  isAlive: boolean;
  userId?: string;
}

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

// 30 秒一次的心跳检查
const interval = setInterval(() => {
  wss.clients.forEach((ws) => {
    const extWs = ws as ExtWebSocket;
    if (!extWs.isAlive) {
      extWs.terminate();
      return;
    }
    extWs.isAlive = false;
    extWs.ping();
  });
}, 30000);

wss.on('connection', (ws: ExtWebSocket, req) => {
  ws.isAlive = true;

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

  ws.on('message', (data: string) => {
    try {
      const payload = JSON.parse(data.toString());
      handleMessage(ws, payload);
    } catch (err) {
      console.error('Payload parse error:', err);
    }
  });

  ws.on('close', () => {
    // 处理清理逻辑，比如通知协同房间其他人
  });
});

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

function handleMessage(ws: ExtWebSocket, payload: any) {
  // 简易的消息分发逻辑
  if (payload.type === 'chat') {
    // 广播给其他在线客户端
    wss.clients.forEach((client) => {
      if (client !== ws && client.readyState === WebSocket.OPEN) {
        client.send(JSON.stringify(payload));
      }
    });
  }
}

// client.ts
type MessageHandler = (data: any) => void;

export class RealtimeClient {
  private url: string;
  private ws: WebSocket | null = null;
  private reconnectAttempts = 0;
  private maxReconnectAttempts = 5;
  private baseDelay = 1000;
  private handlers: Set<MessageHandler> = new Set();
  private messageQueue: string[] = [];

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

  private connect() {
    this.ws = new WebSocket(this.url);

    this.ws.onopen = () => {
      this.reconnectAttempts = 0;
      this.flushQueue();
    };

    this.ws.onmessage = (event) => {
      try {
        const data = JSON.parse(event.data);
        this.handlers.forEach((handler) => handler(data));
      } catch (e) {
        console.error('Failed to parse message', e);
      }
    };

    this.ws.onclose = () => {
      this.handleReconnect();
    };

    this.ws.onerror = (err) => {
      console.error('WebSocket error:', err);
      this.ws?.close();
    };
  }

  private handleReconnect() {
    if (this.reconnectAttempts >= this.maxReconnectAttempts) {
      console.warn('Max reconnect attempts reached.');
      return;
    }

    // 指数退避算法 + 随机抖动，防止服务端雪崩
    const delay = Math.min(
      this.baseDelay * Math.pow(2, this.reconnectAttempts) + Math.random() * 500,
      10000
    );
    this.reconnectAttempts++;

    setTimeout(() => {
      this.connect();
    }, delay);
  }

  public send(payload: any) {
    const raw = JSON.stringify(payload);
    if (this.ws && this.ws.readyState === WebSocket.OPEN) {
      this.ws.send(raw);
    } else {
      this.messageQueue.push(raw);
    }
  }

  private flushQueue() {
    while (this.messageQueue.length > 0 && this.ws?.readyState === WebSocket.OPEN) {
      const msg = this.messageQueue.shift();
      if (msg) this.ws.send(msg);
    }
  }

  public subscribe(handler: MessageHandler) {
    this.handlers.add(handler);
    return () => this.handlers.delete(handler);
  }
}