《TypeScript编程实战》10.3 心跳、重连与广播

socket.send(payload); // 没有抛异常
console.log(socket.readyState); // 仍然是 1（OPEN）

// Node 端（ws 库）：服务端定时给每个连接发 ping
const HEARTBEAT_INTERVAL = 30_000;

const alive = new WeakMap<WebSocket, boolean>();

const timer = setInterval(() => {
  for (const ws of wss.clients) {
    if (alive.get(ws) === false) {
      ws.terminate(); // 上一轮没回 pong，判定为死连接
      continue;
    }
    alive.set(ws, false);
    ws.ping();
  }
}, HEARTBEAT_INTERVAL);

wss.on('connection', (ws) => {
  alive.set(ws, true);
  ws.on('pong', () => alive.set(ws, true));
  ws.on('close', () => alive.delete(ws));
});

type ClientMessage = { type: 'ping'; payload: { ts: number } };
type ServerMessage = { type: 'pong'; payload: { ts: number; serverTime: number } };

// 客户端：既能探活，也能顺便估算时钟偏移
setInterval(() => {
  if (socket.readyState !== WebSocket.OPEN) return;
  socket.send(JSON.stringify({ type: 'ping', payload: { ts: Date.now() } }));
}, 25_000);

class Reconnector {
  private attempt = 0;
  private readonly base = 500;      // 首次 500ms
  private readonly cap = 30_000;    // 上限 30s
  private readonly factor = 1.8;
  private readonly jitter = 0.3;    // ±30% 抖动

  nextDelay(): number {
    const raw = Math.min(this.cap, this.base * this.factor ** this.attempt++);
    const spread = raw * this.jitter;
    return raw - spread + Math.random() * spread * 2; // 落在 [0.7, 1.3] × raw
  }

  reset(): void {
    this.attempt = 0; // 只有「连上并稳定一段时间」后才允许归零
  }
}

type ServerMessage = { type: 'event'; payload: { seq: number; body: unknown } };

// 客户端记录收到的最大序号，重连时带上
const params = new URLSearchParams({ since: String(lastSeq) });
const socket = new WebSocket(`/ws?${params}`);

// 服务端：补发 seq > since 的消息，再切到实时推送
function onConnect(ws: WebSocket, since: number) {
  for (const msg of ringBuffer.since(since)) ws.send(encode(msg));
  liveSubscribers.add(ws);
}

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

function broadcast(channel: string, msg: ServerMessage): void {
  const payload = JSON.stringify(msg);
  for (const ws of channels.get(channel) ?? []) {
    if (ws.readyState === WebSocket.OPEN) ws.send(payload);
  }
}

import { createClient } from 'redis';

const pub = createClient({ url: env.REDIS_URL });
const sub = pub.duplicate();
await Promise.all([pub.connect(), sub.connect()]);

const NODE_ID = process.env.HOSTNAME ?? 'local';

// 发布：本进程产生的事件，投递到 Redis 频道
async function publish(channel: string, msg: ServerMessage): Promise<void> {
  await pub.publish(`rt:${channel}`, JSON.stringify({ origin: NODE_ID, msg }));
}

// 订阅：每个副本都监听所有频道，但只投递给本进程的连接
await sub.pSubscribe('rt:*', (raw, redisChannel) => {
  const { msg } = JSON.parse(raw) as { origin: string; msg: ServerMessage };
  const channel = redisChannel.slice(3); // 去掉 'rt:' 前缀
  for (const ws of channels.get(channel) ?? []) {
    if (ws.readyState === WebSocket.OPEN) ws.send(JSON.stringify(msg));
  }
});

const MAX_BUFFERED = 1 << 20; // 1 MiB

function safeSend(ws: WebSocket, payload: string): void {
  if (ws.bufferedAmount > MAX_BUFFERED) {
    logger.warn({ buffered: ws.bufferedAmount }, 'slow consumer, dropping');
    ws.close(1013, 'too many pending messages'); // 1013 = Try Again Later
    return;
  }
  ws.send(payload);
}

import { describe, expect, it, vi } from 'vitest';

describe('Reconnector', () => {
  it('退避不超过上限且带抖动', () => {
    vi.useFakeTimers();
    const r = new Reconnector();
    const delays = Array.from({ length: 12 }, () => r.nextDelay());
    expect(Math.max(...delays)).toBeLessThanOrEqual(30_000 * 1.3);
    expect(new Set(delays).size).toBeGreaterThan(1); // 抖动生效
    vi.useRealTimers();
  });
});