7.14 实时通信工程化

一句话：实时通信的难点不在『怎么建立连接』，而在『连接断了怎么办』——所以核心是「重连退避 + 心跳探活 + 断线补偿（用序号补发缺失消息）+ 多标签页只留一个连接」这四件事，而选型上「SSE 常被低估、WebSocket 常被过度使用」。

一、四种方案：先选对

⭐⭐⭐ 选型的三个判据（⭐ 顺序很重要）：

① 【「需不需要客户端高频发消息」】⭐ 最重要
   不需要（只接收推送）→ ⭐ **优先 SSE**：
     · 它就是一个普通 HTTP 长连接（网关/代理/CDN 天然支持，不需要 Upgrade）
     · ⭐ 浏览器**内建**自动重连（默认 3 秒）
     · ⭐⭐ 而且**重连时会自动带上 `Last-Event-ID`**（服务端能据此「补发缺失的消息」）
       —— 这正是「断线补偿」最难做的部分，SSE **白送**
     · 实现简单（服务端就是 `res.write('data: ...\n\n')`）
   · ⚠️ 但 SSE 是**单向**的：客户端要发数据得「另开一个 POST 接口」
   · ⚠️ 且 `EventSource` **不能自定义请求头**（认证只能靠 cookie 或 query）
   → ⭐ 「聊天」也可以「SSE 收 + POST 发」（⭐ 很多聊天应用就这么做）

② 【「延迟要求是毫秒级吗」】
   是（游戏/交易/协作光标）→ **WebSocket**（SSE 的「事件流」语义在毫秒级场景下不够直接）
   否（秒级/亚秒级）→ SSE 或长轮询

③ 【「兼容性/基础设施约束」】
   · 有「不支持长连接的代理/网关」→ 长轮询（⭐ 但现代基础设施基本都支持 SSE）
   · 需要「跨域 + 自定义头」→ WebSocket 也**不能自定义头**（只能 cookie/subprotocol），
     SSE 也不能 → ⭐ 两者在「认证」上的限制是一样的

⭐ 一句话：
   · **只推送 → SSE（⭐ 默认选择）**
   · **双向高频 → WebSocket**
   · **实时性要求低 → 轮询（最省事）**

⭐⭐⭐ 「WebSocket 常被过度使用」的具体表现：

   ❌ 「消息通知/进度更新/AI 流式输出」用了 WebSocket
      → ⭐ 这些场景**只需要「服务端推送」**，用 SSE 能省掉：
        「重连逻辑 + 心跳 + 消息协议 + 网关 Upgrade 支持 + 连接状态管理」
      （⭐ 而且「双向」会让「谁在什么时候发什么」变得难以推理）

   ❌ 「用 WebSocket 做『客户端命令通道』」（客户端发命令、服务端执行）
      → ⚠️ 这等于「自己造了一个 RPC 协议」（没有超时、没有重试、没有幂等、没有可观测性）
      → ✅ 用「POST 接口」+ 服务端「通过 SSE 推结果」

   ⭐ 判据：「**如果『客户端 → 服务端』的消息频率很低（每分钟几次），那就用 POST**」

二、连接管理：五件事缺一不可

⭐⭐⭐⭐ 实时连接的五个「必备件」（⭐ 用 WebSocket 时要全做；用 SSE 时前三个白送）：

① 【重连 + 指数退避 + 抖动】
   ❌ 「`onclose` 里立刻重连」→ 服务端挂了时「所有客户端每秒重连一次」（⭐ 把服务端打得更死）
   ✅ **指数退避**（1s → 2s → 4s → … 上限 30s）+ **抖动**（避免「同一秒集中重连」）
   ✅ 还要有「上限」（如「退避到 30s 后就不再增加」）与「成功连接后重置」

② 【心跳（探活）】⭐
   为什么必须做：
     · ⚠️ **「半开连接」**：网络断了但 TCP 没感知（如「笔记本换了网络」「中间设备静默丢弃」）→
       连接看似「OPEN」，实际「发不出去也收不到」
     · ⚠️ **中间设备会断「空闲连接」**（Nginx 默认 60s、负载均衡 300s、云厂商可能更短）
   ✅ 做法：**定时发心跳**（如每 25 秒）+ **超时未收到「任何消息」（含服务端心跳）就主动重连**
   ⚠️ 间隔选择：太短费电（移动端尤其）→ 一般「25~30 秒」，
     并「小于中间设备的最短空闲超时」

③ 【断线补偿（⭐ 最难）】
   ⭐ 问题：「断线 30 秒」期间的消息全部丢了 → 重连后 UI 与真实状态不一致
   ✅ 两种做法：
     · ⭐ **「带序号 + 补发」**：服务端给每条消息一个「单调递增的序号」，
       客户端记录「最后收到的序号」，重连时带上它 → 服务端**补发缺失的消息**
     · ⭐ **「重连后全量刷新」**：更简单但「贵」（⭐ 适合「状态小、变化频繁」的场景）
   ⚠️ 边界：「缺失太多」（离线几小时）→ 应该返回「**需要全量刷新**」的信号
     （而不是「补发十万条消息」）

④ 【多标签页只留一个连接】⭐
   ❌ 每个标签页都开连接 → 「连接数 ×N」+「同一条消息被处理 N 次」
     （⚠️ 典型症状：Toast 弹 N 次、埋点重复上报）
   ✅ **Leader 选举**（Web Locks API，见 3.14）+ **BroadcastChannel 广播**给其它标签页
   ⭐ 这样「一个连接 + 多标签页共享」

⑤ 【后台标签页的处理】⭐
   ⚠️ 浏览器会「节流/挂起」后台标签页：WebSocket 可能被断开、定时器被降频
   ✅ 恢复可见（`visibilitychange`）时：
     · **检查连接状态**（`readyState`）
     · **主动重连**（如果已断）
     · ⭐ **做一次「补偿/全量刷新」**（因为「后台期间可能错过了消息」）

/** ============ 一个「消息序号」的设计（⭐ 断线补偿的基础） ============ */
export interface ServerMessage {
  /** ⭐⭐ 单调递增的序号（每个「频道/房间」独立递增） */
  seq: number;
  /** 消息类型（用「判别联合」表达，便于穷举处理） */
  type: string;
  /** 消息体 */
  payload: unknown;
  /** 服务端时间（用于展示/调试，⭐ 不要用它做「顺序判断」） */
  serverTime: number;
}

/** ⭐ 客户端记录「各频道最后收到的序号」 */
export class SequenceTracker {
  private lastSeq = new Map<string, number>();

  /** ⭐ 收到消息（返回「是否按序」——用来发现「服务端漏发」） */
  accept(channel: string, seq: number): { inOrder: boolean; gap: number } {
    const prev = this.lastSeq.get(channel) ?? 0;
    this.lastSeq.set(channel, Math.max(prev, seq));
    return { inOrder: seq === prev + 1, gap: Math.max(0, seq - prev - 1) };
  }

  get(channel: string): number { return this.lastSeq.get(channel) ?? 0; }

  /** ⭐ 重连后「重置」（因为要「从服务端要补发」，不能再假设连续） */
  reset(channel: string, seq: number) { this.lastSeq.set(channel, seq); }
}

三、完整实现：一个实时连接层（250 行）

// realtime.ts —— 一个「可直接用」的实时层（心跳 + 退避重连 + 补偿 + Leader 选举）
// 支持两种传输：WebSocket（双向）与 SSE（单向）—— ⭐ 统一在同一套「连接接口」后
import { createLeaderElection } from './leader';       // 见 3.14

/** ============ 1. 传输抽象（⭐ 让「上层」不关心 WS 还是 SSE） ============ */
export interface RealtimeTransport {
  readonly kind: 'websocket' | 'sse';
  connect(url: string, options: {
    /** ⭐ 重连时带上「各频道最后收到的序号」（用于补偿） */
    lastSeqByChannel: Record<string, number>;
    onMessage: (msg: ServerMessage) => void;
    /** ⭐ 「连接就绪」（可以开始接收） */
    onOpen: () => void;
    /** ⭐ 「需要全量刷新」（服务端无法补发时） */
    onResyncRequired: (reason: string) => void;
    onError: (err: Event) => void;
    onClose: (info: { code?: number; reason?: string; wasClean: boolean }) => void;
  }): void;
  send(data: unknown): boolean;
  close(): void;
  readonly state: 'connecting' | 'open' | 'closed';
}

/** ============ 2. WebSocket 传输（⭐ 心跳 + 半开检测） ============ */
export function createWebSocketTransport(options: {
  heartbeatIntervalMs?: number;
  /** ⭐ 多久没收到「任何消息」（含服务端心跳）就认为「连接已死」 */
  pongTimeoutMs?: number;
} = {}): RealtimeTransport {
  const { heartbeatIntervalMs = 25_000, pongTimeoutMs = 60_000 } = options;

  let ws: WebSocket | null = null;
  let state: RealtimeTransport['state'] = 'closed';
  let heartbeatTimer: ReturnType<typeof setInterval> | null = null;
  let lastActivityAt = 0;

  const clearHeartbeat = () => {
    if (heartbeatTimer) { clearInterval(heartbeatTimer); heartbeatTimer = null; }
  };

  const transport: RealtimeTransport = {
    kind: 'websocket',
    get state() { return state; },

    connect(url, io) {
      state = 'connecting';
      lastActivityAt = Date.now();

      // ⭐ 认证：WebSocket 握手时「浏览器不能自定义请求头」→
      //    ① 同源时 cookie 会自动带上（⭐ 推荐）
      //    ② 跨域或不方便用 cookie 时，用 subprotocol 或 query 参数（⚠️ query 会进日志）
      ws = new WebSocket(url, ['app.v1']);

      ws.onopen = () => {
        state = 'open';
        lastActivityAt = Date.now();

        // ⭐⭐ 连接就绪后「立刻上报『我有哪些频道、最后收到第几号』」→ 让服务端补发
        ws?.send(JSON.stringify({
          type: '__subscribe',
          lastSeqByChannel: io.lastSeqByChannel,
        }));

        io.onOpen();

        // ⭐ 心跳（⭐ 用「应用层心跳」而不是「协议层 ping 帧」——
        //    因为浏览器 WebSocket API 不能发 ping 帧，也没法读 pong）
        clearHeartbeat();
        heartbeatTimer = setInterval(() => {
          if (!ws || ws.readyState !== WebSocket.OPEN) return;

          // ⭐ 半开检测：超过 pongTimeout 没收到任何消息 → 认为「已死」→ 主动关闭以触发重连
          if (Date.now() - lastActivityAt > pongTimeoutMs) {
            console.warn('[realtime] 心跳超时，主动断开以触发重连');
            ws.close(4000, 'heartbeat timeout');
            return;
          }
          ws.send(JSON.stringify({ type: '__ping', at: Date.now() }));
        }, heartbeatIntervalMs);
      };

      ws.onmessage = (event) => {
        lastActivityAt = Date.now();                  // ⭐ 任何消息都算「活着」
        try {
          const data = JSON.parse(String(event.data)) as ServerMessage | { type: '__pong' } | { type: '__resync'; reason: string };

          if (data.type === '__pong') return;         // ⭐ 心跳响应，不往上层传
          if (data.type === '__resync') {
            io.onResyncRequired((data as { reason: string }).reason);
            return;
          }
          io.onMessage(data as ServerMessage);
        } catch (err) {
          console.warn('[realtime] 消息解析失败', err);
        }
      };

      ws.onerror = (e) => io.onError(e);
      ws.onclose = (e) => {
        state = 'closed';
        clearHeartbeat();
        io.onClose({ code: e.code, reason: e.reason, wasClean: e.wasClean });
      };
    },

    send(data) {
      if (!ws || ws.readyState !== WebSocket.OPEN) return false;   // ⭐ 返回 false 让上层「入队」
      ws.send(JSON.stringify(data));
      return true;
    },

    close() {
      clearHeartbeat();
      ws?.close(1000, 'client close');
      ws = null;
      state = 'closed';
    },
  };

  return transport;
}

/** ============ 3. SSE 传输（⭐ 单向：只接收，发送走另一个 POST 接口） ============ */
export function createSseTransport(options: {
  /** ⭐ 客户端「发消息」的接口（SSE 是单向的） */
  sendEndpoint: string;
}): RealtimeTransport {
  let es: EventSource | null = null;
  let state: RealtimeTransport['state'] = 'closed';

  return {
    kind: 'sse',
    get state() { return state; },

    connect(url, io) {
      state = 'connecting';

      // ⭐ EventSource 会「自动重连」，但我们要控制「带上 lastSeq」→
      //    所以把「序号」放进 URL query（⚠️ 会进服务端日志，不要放敏感信息）
      const params = new URLSearchParams();
      params.set('lastSeq', JSON.stringify(io.lastSeqByChannel));
      es = new EventSource(`${url}?${params}`, { withCredentials: true });   // ⭐ 同源 cookie 认证

      es.onopen = () => { state = 'open'; io.onOpen(); };

      es.onmessage = (event) => {
        // ⭐ SSE 的 `id:` 字段会自动变成「重连时的 `Last-Event-ID` 请求头」
        //    —— 所以「序号」可以交给浏览器带（⭐ 这是 SSE 白送的补偿能力）
        try {
          const msg = JSON.parse(event.data) as ServerMessage | { type: '__resync'; reason: string };
          if (msg.type === '__resync') {
            // ⭐ 服务端说「无法补发」→ 让上层做「全量刷新」
            io.onResyncRequired((msg as { reason: string }).reason);
            return;
          }
          io.onMessage(msg as ServerMessage);
        } catch (err) {
          console.warn('[realtime] SSE 消息解析失败', err);
        }
      };

      es.onerror = () => {
        // ⚠️ SSE 的 onerror「很含糊」：可能是「网络断」、「服务端关了」、也可能「正在重连」
        //    它**没有**「连接已死的明确信号」→ 所以我们用「自己的心跳超时」兜底
        state = 'connecting';                        // ⭐ EventSource 会自动重连，这里标「连接中」
        io.onError(new Event('sse-error'));
      };
    },

    send(data) {
      // ⭐ SSE 单向：发送走「普通 POST」（⭐ 这样「发送」获得「HTTP 的全部能力」：
      //    重试、幂等键、超时、可观测性 —— 见 7.1/7.12）
      void fetch(options.sendEndpoint, {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify(data),
        keepalive: true,
      }).catch(() => { /* 上层可以通过「队列」重试，见 7.12 */ });
      return true;
    },

    close() { es?.close(); es = null; state = 'closed'; },
  };
}

/** ============ 4. 上层：统一的 RealtimeClient（重连 + 广播 + 多标签页） ============ */
export interface RealtimeOptions {
  url: string;
  transport: 'websocket' | 'sse';
  sseSendEndpoint?: string;
  /** ⭐ 心跳参数（仅 WS 用；SSE 靠浏览器自动重连） */
  heartbeatIntervalMs?: number;
  /** ⭐ 重连退避参数 */
  reconnectBaseMs?: number;
  reconnectMaxMs?: number;
  /** ⭐ 需要「全量刷新」时的回调（由业务层实现，通常是「重新拉一次关键数据」） */
  onResyncRequired?: (reason: string) => void;
}

type Listener = (msg: ServerMessage) => void;

export function createRealtimeClient(options: RealtimeOptions) {
  const {
    url, transport: kind, sseSendEndpoint,
    heartbeatIntervalMs = 25_000,
    reconnectBaseMs = 1_000,
    reconnectMaxMs = 30_000,
    onResyncRequired,
  } = options;

  let transport: RealtimeTransport;
  let state: 'idle' | 'connecting' | 'open' | 'closed' = 'idle';
  let attempts = 0;
  let reconnectTimer: ReturnType<typeof setTimeout> | null = null;
  let manuallyClosed = false;

  const listeners = new Set<Listener>();
  const seqTracker = new SequenceTracker();
  /** ⭐ 频道订阅（业务层按频道注册） */
  const channels = new Set<string>();
  /** ⭐ 离线期间的「待发消息」（连接恢复后按序发送） */
  const outbox: unknown[] = [];

  const stateListeners = new Set<() => void>();
  const notifyState = () => stateListeners.forEach((fn) => fn());

  // ⭐⭐ 多标签页：只让 Leader 维护连接，其它标签页「通过 BroadcastChannel 接收」
  const channel = typeof BroadcastChannel !== 'undefined' ? new BroadcastChannel(`realtime:${url}`) : null;

  function buildTransport(): RealtimeTransport {
    return kind === 'websocket'
      ? createWebSocketTransport({ heartbeatIntervalMs })
      : createSseTransport({ sendEndpoint: sseSendEndpoint ?? '/api/realtime/send' });
  }

  function scheduleReconnect() {
    if (manuallyClosed) return;
    if (reconnectTimer) return;                        // ⭐ 防重（一次只排一个）

    // ⭐ 指数退避 + 抖动
    const base = Math.min(reconnectMaxMs, reconnectBaseMs * 2 ** attempts);
    const jitter = Math.random() * 0.3 * base;
    const delay = Math.round(base + jitter);
    attempts += 1;

    console.info(`[realtime] ${Math.round(delay / 1000)}s 后重连（第 ${attempts} 次）`);
    reconnectTimer = setTimeout(() => {
      reconnectTimer = null;
      connect();
    }, delay);
  }

  function connect() {
    if (manuallyClosed) return;
    state = 'connecting';
    notifyState();

    transport = buildTransport();
    transport.connect(url, {
      // ⭐ 把「各频道最后收到的序号」带给服务端 → 请求补偿
      lastSeqByChannel: Object.fromEntries([...channels].map((c) => [c, seqTracker.get(c)])),

      onOpen: () => {
        state = 'open';
        attempts = 0;                                  // ⭐ 成功后重置退避
        notifyState();

        // ⭐⭐ 把「离线期间攒的消息」按序发出去
        while (outbox.length > 0) {
          const msg = outbox[0];
          if (!transport.send(msg)) break;             // ⭐ 还是发不出去 → 留在队列
          outbox.shift();
        }
      },

      onMessage: (msg) => {
        // ⭐ ① 序号校验（发现「服务端漏发」时「主动要补偿」）
        if (typeof msg.seq === 'number' && msg.type !== '__pong') {
          const channelName = (msg as unknown as { channel?: string }).channel ?? 'default';
          const { gap } = seqTracker.accept(channelName, msg.seq);
          if (gap > 0) {
            console.warn(`[realtime] 频道 ${channelName} 缺了 ${gap} 条消息 → 触发一次手动刷新`);
            // ⭐ 兜底：缺口太大或无法补发时，「全量刷新」比「等补发」更可靠
            onResyncRequired?.(`sequence gap: ${gap}`);
          }
        }

        // ⭐② 广播给「其它标签页」（多标签页只一个连接）
        channel?.postMessage(msg);
        // ⭐③ 通知本标签页的监听器
        listeners.forEach((fn) => fn(msg));
      },

      onResyncRequired: (reason) => {
        console.warn('[realtime] 服务端要求全量刷新：', reason);
        seqTracker.reset('default', 0);
        onResyncRequired?.(reason);
      },

      onError: () => { /* ⭐ 错误统一在 onClose 里处理重连（避免「错误+关闭」重连两次） */ },

      onClose: () => {
        if (state === 'open') state = 'closed';
        notifyState();
        scheduleReconnect();                           // ⭐ 断线 → 退避重连
      },
    });
  }

  // ⭐ 其它标签页：接收「Leader 广播的消息」，转给自己的监听器
  channel?.addEventListener('message', (e) => {
    listeners.forEach((fn) => fn(e.data as ServerMessage));
  });

  const leader = createLeaderElection(
    `realtime:${url}`,
    () => { connect(); },                              // ⭐ 成为 Leader → 建立连接
    () => { manuallyClosed = true; transport?.close(); } // ⭐ 失去 Leader → 断开
  );

  return {
    /** ⭐ 启动：抢到 Leader 锁后由 `onBecomeLeader` 触发 connect（见 3.14 的 createLeaderElection） */
    start() { attempts = 0; manuallyClosed = false; },
    stop() { manuallyClosed = true; leader.release(); transport?.close(); if (reconnectTimer) clearTimeout(reconnectTimer); },
    get state() { return state; },

    /** ⭐ 订阅频道（会「带上 lastSeq」让服务端补发） */
    subscribe(channelName: string) {
      channels.add(channelName);
      transport.send({ type: '__subscribe', channel: channelName, lastSeq: seqTracker.get(channelName) });
    },

    /** ⭐ 发消息（连不上时「入队」—— 与 7.12 的队列思想一致） */
    send(msg: unknown): boolean {
      const ok = transport.send(msg);
      if (!ok) {
        outbox.push(msg);
        // ⚠️ 队列要有上限（避免「离线很久」时内存爆）
        if (outbox.length > 200) outbox.shift();
      }
      return ok;
    },

    /** ⭐ 订阅消息 */
    onMessage(fn: Listener) { listeners.add(fn); return () => { listeners.delete(fn); }; },

    /** ⭐ 订阅「连接状态」（UI 显示「连接中/已断开」） */
    onStateChange(fn: () => void) { stateListeners.add(fn); return () => { stateListeners.delete(fn); }; },

    /** ⭐⭐ 「恢复可见」时主动检查（后台标签页可能被挂起） */
    handleVisibilityChange() {
      if (document.visibilityState !== 'visible') return;
      if (state !== 'open') {
        console.info('[realtime] 恢复可见但连接已断 → 立即重连（不走退避）');
        attempts = 0;
        connect();
      } else {
        // ⭐ 连接还在，但「后台期间可能错过消息」→ 主动补偿一次
        transport.send({ type: '__resync_request', lastSeqByChannel: Object.fromEntries([...channels].map((c) => [c, seqTracker.get(c)])) });
      }
    },
  };
}

/** ============ 5. React 集成 ============ */
import { useEffect, useRef, useSyncExternalStore, useState, useCallback } from 'react';

export function useRealtime(options: RealtimeOptions) {
  const clientRef = useRef<ReturnType<typeof createRealtimeClient> | null>(null);
  if (clientRef.current === null) clientRef.current = createRealtimeClient(options);
  const client = clientRef.current;

  const [state, setState] = useState(client.state);

  useEffect(() => {
    client.start();

    const offState = client.onStateChange(() => setState(client.state));

    // ⭐⭐ 关键：恢复可见时「检查连接 + 补偿」（后台标签页会被节流/挂起）
    const onVisible = () => client.handleVisibilityChange();
    document.addEventListener('visibilitychange', onVisible);

    // ⭐ 网络恢复时「立即重连」（不走退避）
    const onOnline = () => { if (client.state !== 'open') client.handleVisibilityChange(); };
    window.addEventListener('online', onOnline);

    return () => {
      document.removeEventListener('visibilitychange', onVisible);
      window.removeEventListener('online', onOnline);
      offState();
      client.stop();
    };
  }, [client]);

  return { client, state };
}

/** ⭐ 订阅「某类消息」（业务组件用） */
export function useRealtimeMessage<T extends ServerMessage['payload']>(
  client: ReturnType<typeof createRealtimeClient>,
  type: string,
  handler: (payload: T, msg: ServerMessage) => void
) {
  const handlerRef = useRef(handler);
  handlerRef.current = handler;

  useEffect(() => {
    return client.onMessage((msg) => {
      if (msg.type !== type) return;
      handlerRef.current(msg.payload as T, msg);
    });
  }, [client, type]);
}

/** ============ 6. 使用示例：一个「实时订单列表」 ============ */
export function RealtimeOrders() {
  const { client, state } = useRealtime({ url: '/api/realtime', transport: 'websocket' });
  const queryClient = useQueryClient();

  // ⭐ 订阅频道（会让服务端「补发」断线期间的变化）
  useEffect(() => { client.subscribe('orders'); }, [client]);

  // ⭐ 收到「订单变更」→ 更新缓存（⭐ 与 7.4 的「推送后失效」是同一思路）
  useRealtimeMessage(client, 'order.updated', (payload: { id: string }) => {
    void queryClient.invalidateQueries({ queryKey: ['order', payload.id] });
    void queryClient.invalidateQueries({ queryKey: ['orders'] });
  });

  // ⭐ 断线时「全量刷新」（兜底）
  useRealtimeMessage(client, 'resync', () => {
    void queryClient.invalidateQueries();
  });

  return (
    <div>
      {/* ⭐⭐ 连接状态必须可见（否则用户以为「没消息」而不是「断线了」） */}
      {state !== 'open' && (
        <div role="status" aria-live="polite" style={{ background: '#fff7e6', padding: 6, fontSize: 12 }}>
          {state === 'connecting' ? '正在连接实时服务…' : '实时连接已断开，正在重连…'}
        </div>
      )}
      <OrderList />
    </div>
  );
}

⭐⭐⭐ 这个实现里的六个「工程关键点」：

① 【把「心跳超时」当作「连接已死」的唯一判据】⭐
   `onclose`/`onerror` **不可靠**：
     · 「半开连接」时它们**不会触发**（连接看起来还是 OPEN）
     · 「中间设备静默断连」时也**不会立刻触发**
   ⭐ 所以「定时心跳 + 超时未收到任何消息 → 主动 close」才是可靠的探活手段。
   ⚠️ 而且「任何消息」都算活着（包括业务消息）→ 不用只看 `__pong`。

② 【「序号 + 补发」是断线补偿的核心】⭐
   客户端记「各频道最后收到的 seq」→ 重连时带给服务端 → 服务端补发缺失部分。
   ⭐ 并加一层兜底：「发现序号缺口」→ 触发「全量刷新」（`onResyncRequired`）——
     因为「补发」在「缺口很大」或「服务端没实现」时都不可靠。

③ 【SSE 白送「重连 + `Last-Event-ID`」】⭐
   `EventSource` 会自动重连，并把「最后收到的 `id:` 字段」放进 **`Last-Event-ID` 请求头**。
   ⭐ 于是「服务端只要『按 id 找到断点并补发』」就能实现补偿——**不用自己写重连逻辑**。
   ⚠️ 但它的 `onerror` 语义含糊（分不清「网络上抖动」还是「服务端关了」）→
     仍需要「应用层心跳」兜底。

④ 【多标签页：Leader 选举 + BroadcastChannel】⭐
   只有 Leader 建连接；收到消息后「广播给其它标签页」。
   ⚠️ 不做的后果：「Toast 弹 N 次」「埋点重复上报」「连接数 ×N」。

⑤ 【「发消息」在连不上时入队】⭐
   `send()` 返回 `false` → 上层入队 → 连接恢复后「按序补发」。
   ⭐ 这与 7.12 的「离线队列」是同一个思想（只是「传输通道」不同）。
   ⚠️ 队列要有**上限**（离线很久时不能无限攒）。

⑥ 【恢复可见/网络恢复时「立刻重连 + 补偿」】⭐
   因为后台标签页会被「节流/挂起」（WS 可能被断、定时器被降频）。
   ⭐ 所以 `visibilitychange` 时：连接断了就「立即重连（不走退避）」；
     连接还在就「主动要一次补偿」（因为后台期间可能错过消息）。

四、服务端侧：不谈实现，谈四个约束

⭐⭐⭐ 服务端要做实时的四个「架构约束」（面试常问）：

① 【连接是有状态的 → 水平扩展变难】⭐
   「5000 个客户端分布在 7 个实例上」→
   要给「用户 A」推消息时，怎么知道「A 连在哪个实例上」？
   ✅ 方案：
     · **广播**（每个实例都收到，只给「自己持有的连接」推）—— 简单，但**消息量 ×实例数**
     · ⭐ **路由表**（用 Redis 记录「userId → instanceId」）—— 精确，但要多一次查询
     · ⭐ **消息队列/发布订阅**（Redis Pub/Sub / Kafka / NATS）—— 实例订阅「频道」，
       只处理「自己关心的」→ 这是主流做法
   ⚠️ 注意：「单机内存里的连接表」在「多实例」下必然不够（⭐ 这是「本地能跑、上线就错乱」的原因）。

② 【「房间/频道」的粒度设计】
   · 按「实体」分房间（`order:123`）→ 精准，但「房间数爆炸」
   · 按「用户」分房间（`user:42`）→ 简单（⭐ 推荐：每个用户一个「个人频道」）
   · ⭐ **实践**：「**每用户一个频道 + 服务端按权限决定『推给他什么』**」——
     这样「客户端不需要声明『我要订哪些』」，也就「没有越权订阅的风险」（见 6.12）

③ 【「推送」可能失败/重复】
   ⭐ 实时通道**不能当作可靠传输**：
     · 「发出去」不等于「客户端收到」（可能正好在断线窗口）
     · ⚠️ 也可能「重复」（重连补发 + 原来的消息）
   ✅ 做法（与 7.12 相同）：
     · 消息带 **序号**（客户端去重 + 发现缺口）
     · ⭐ **「推送」只是「提示有变化」，真正的数据「从 REST/GraphQL 拉」**——
       这样「丢了推送」最坏只是「晚一点知道」，**不会「数据错」**（⭐ 强烈推荐这种设计）
     · 写操作走「有幂等键的 POST/DELETE」（见 7.12）

④ 【背压与限流】
   ⚠️ 「10 万客户端 × 每秒 10 条」= 100 万消息/秒（⭐ 服务端直接崩）
   ✅ 做法：
     · ⭐ **「合并/降采样」**：同一实体 100ms 内的多次变更「只推最后一次」（⭐ 极有效）
     · 「按用户限流」（超出就丢，客户端靠「全量刷新」兜底）
     · 「大房间不做全量广播」（改成「定向推送」）

⭐ 一句话：「**Service 端的实时实现方式有千万种，但这四个约束是绕不过的**」。

五、本篇特有的坑

// ① `onclose` 里「立刻重连」（服务端挂了 → 所有客户端每秒重连，把它打得更死）
ws.onclose = () => connect();                                  // ❌
// ✅ 指数退避 + 抖动 + 上限

// ② 没有心跳（半开连接「看起来是 OPEN」，实际收发都断）
// ⚠️ 症状：「界面长时间不更新，但连接状态显示正常」
// ✅ 定时心跳 + 「超时未收到任何消息 → 主动重连」

// ③ 心跳间隔太长（被中间设备断掉空闲连接）
// ⚠️ Nginx 默认 60s、负载均衡可能 300s
// ✅ 25~30 秒（并小于中间设备的最短超时）

// ④ 心跳太频繁（移动端耗电 + 服务端压力）
setInterval(ping, 1000);                                       // ❌
// ✅ 25~30 秒

// ⑤ 「断线期间的消息全丢」但没有补偿
// ⚠️ 症状：重连后 UI 与真实状态不一致（用户看到「旧数据」）
// ✅ 「序号 + 补发」或「重连后全量刷新」

// ⑥ 「补发十万条消息」（离线太久）
// ✅ 服务端返回「需要全量刷新」（`__resync`）

// ⑦ 多个标签页各开一个连接（连接数 ×N + 消息重复处理）
// ⚠️ 典型症状：Toast 弹 N 次、埋点重复
// ✅ Leader 选举 + BroadcastChannel（见 3.14）

// ⑧ 后台标签页「连接被挂起」但「没有在恢复时检查」
// ✅ `visibilitychange` → 检查状态 + 重连/补偿

// ⑨ 「用 WebSocket 做客户端→服务端的命令通道」（自己造了个不可靠的 RPC）
ws.send({ type: 'refund', orderId });                          // ⚠️
// ✅ 走 POST（有超时/重试/幂等/可观测性，见 7.1/7.13）

// ⑩ 「推送即数据」（把推送当「可靠传输」）
// ⚠️ 推送可能丢/重复 → 「数据错了」而不是「晚一点知道」
// ✅ ⭐ **推送只作「有变化」的提示，数据从 REST/GraphQL 拉**

// ⑪ 消息「没有去重」（重连补发 + 原消息 → 同一条处理两次）
// ✅ 带「消息 id/seq」+ 客户端 LRU 去重（见 3.14）

// ⑫ 消息「没有版本/序号」（无法判断「乱序」「缺口」）
// ✅ 单调递增 seq

// ⑬ `send()` 失败「直接丢弃」（用户的操作消失）
// ✅ 入队 + 重连后补发（并给「队列上限」与「UI 提示」）

// ⑭ 「认证」用 query 参数（⭐ 会进服务端日志、浏览器历史、Referer）
new WebSocket(`/ws?token=${token}`);                            // ⚠️
// ✅ 同源用 cookie；跨域用「连接后第一条消息发 token」（⚠️ 且有「未认证前的超时」）

// ⑮ 「认证 token 过期」时连接还在（能收发但会越权）
// ✅ 服务端「定期校验 token」+ 「过期就发一条 `__auth_expired` 让客户端刷新并重连」

// ⑯ 「跨域/代理」不支持 WebSocket（Upgrade 被拦）
// ⚠️ 症状：握手 400/101 失败
// ✅ 用 SSE 兜底（⭐ 它是普通 HTTP）；或让运维放行 Upgrade

// ⑰ SSE 在「HTTP/1.1 下占满同域连接数」（6 个限制 → 其它请求被排队）⭐
// ✅ 用 HTTP/2（多路复用）；⭐ 或「减少 EventSource 数量」（用一个连接 + 多频道）

// ⑱ SSE 的 `EventSource` 不能自定义头（认证受限）⭐
// ✅ cookie（同源）+ `withCredentials: true`；⭐ 或「用 fetch + ReadableStream 自己实现 SSE」
//    （⭐ 这样就能带自定义头了 —— 代价是「要自己实现重连」）

// ⑲ 「服务端没有『连接数』的监控」（连接泄漏导致打满文件描述符）
// ✅ 监控「当前连接数 / 每秒新建连接数 / 异常关闭率」

// ⑳ 「消息体积大」（把「完整对象」塞进推送）
// ⚠️ 浪费带宽（尤其移动网络）
// ✅ ⭐ **推送只带「id + 变更类型」**（客户端据此去拉详情，见第 ⑩ 条）

// ㉑ 「本地开发能跑，多实例上线后错乱」（连接表在单机内存里）
// ✅ 用 Redis Pub/Sub 或消息队列做「跨实例广播/路由」

// ㉒ 「消息顺序」被当作「全局有序」
// ⚠️ WebSocket 保证「单连接内有顺序」，但「多连接/多实例」下没有全局顺序
// ✅ 用「按实体的序号」而不是「全局序号」

// ㉓ 「重连风暴」（服务端重启 → 十万客户端同时重连）
// ✅ 客户端「抖动 + 退避」；⭐ 服务端「灰度重启」（分批，让客户端错开重连）

// ㉔ 「没有『连接状态』的 UI」（用户不知道「断线了」）
// ✅ 状态条（「连接中/已断开/正在重连」）+ ⭐ 「断线超过 N 秒」时「明确提示数据可能陈旧」

// ⑱ 的完整处理（用 fetch + ReadableStream 自己实现 SSE —— 可以带自定义头）
export async function fetchSse(
  url: string,
  options: { headers?: Record<string, string>; signal?: AbortSignal; onEvent: (e: { id?: string; event?: string; data: string }) => void }
) {
  const res = await fetch(url, {
    headers: { Accept: 'text/event-stream', ...options.headers },   // ⭐ 可以带 Authorization
    signal: options.signal,
    credentials: 'same-origin',
  });
  if (!res.ok || !res.body) throw new Error(`SSE 连接失败：${res.status}`);

  const reader = res.body.pipeThrough(new TextDecoderStream()).getReader();
  let buffer = '';

  // ⭐ 按 SSE 规范解析：事件之间用「空行」分隔
  for (;;) {
    const { value, done } = await reader.read();
    if (done) break;
    buffer += value;

    let sepIndex: number;
    while ((sepIndex = buffer.indexOf('\n\n')) >= 0) {
      const raw = buffer.slice(0, sepIndex);
      buffer = buffer.slice(sepIndex + 2);

      let id: string | undefined;
      let event: string | undefined;
      const dataLines: string[] = [];

      for (const line of raw.split('\n')) {
        if (line.startsWith(':')) continue;                    // ⭐ 注释/心跳
        const colon = line.indexOf(':');
        const field = colon === -1 ? line : line.slice(0, colon);
        const val = colon === -1 ? '' : line.slice(colon + 1).replace(/^ /, '');
        if (field === 'id') id = val;
        else if (field === 'event') event = val;
        else if (field === 'data') dataLines.push(val);
      }
      if (dataLines.length) options.onEvent({ id, event, data: dataLines.join('\n') });
    }
  }
}
// ⭐ 代价：要自己实现「重连 + Last-Event-ID」（⭐ 但这就能「带 Authorization 头」了）

// ① 的完整处理（退避 + 抖动的完整实现）
function createReconnector(options: { baseMs?: number; maxMs?: number; onReconnect: () => void }) {
  const { baseMs = 1000, maxMs = 30_000, onReconnect } = options;
  let attempts = 0;
  let timer: ReturnType<typeof setTimeout> | null = null;

  return {
    schedule() {
      if (timer) return;
      const exp = Math.min(maxMs, baseMs * 2 ** attempts);
      const jitter = Math.random() * 0.3 * exp;               // ⭐ 抖动：避免「重连风暴」
      attempts += 1;
      timer = setTimeout(() => { timer = null; onReconnect(); }, Math.round(exp + jitter));
    },
    reset() { attempts = 0; },                                // ⭐ 连接成功后重置
    cancel() { if (timer) { clearTimeout(timer); timer = null; } },
    get attempts() { return attempts; },
  };
}

六、面试延伸

1. 「SSE 和 WebSocket 怎么选？」
   ⭐ 先问「需不需要客户端高频发消息」：① ⭐ 不需要（只接收推送）→ 优先 SSE——它是一个普通 HTTP 长连接（网关/代理/CDN 天然支持，不需要 Upgrade），⭐ 浏览器内建自动重连，⭐ 而且重连时会自动带 Last-Event-ID 请求头（服务端据此补发缺失消息）——这正是「断线补偿」最难的部分，SSE 白送；实现也简单（服务端 res.write('data: ...\n\n')）；⚠️ 代价是单向（客户端要发数据得另开 POST 接口，⭐ 这反而是好事：发送获得 HTTP 的全部能力——超时/重试/幂等/可观测性）与 EventSource 不能自定义请求头（认证只能 cookie 或 query）。② ⭐ 需要双向高频（毫秒级）→ WebSocket（协作编辑、游戏、交易）；⚠️ 代价是「重连/心跳/协议/网关 Upgrade 全要自己实现」。③ 实时性要求低 → 轮询（最省事）。⭐ 一句话：「只推送 → SSE（默认选择）；双向高频 → WebSocket；要求低 → 轮询」——⭐ WebSocket 在「只推送」场景里是被过度使用的。

2. 「WebSocket 的心跳为什么必须做？」
   因为 onclose/onerror 不可靠——有两类「连接已死但代码不知道」的情况：① ⭐ 「半开连接」——网络切换（笔记本换 WiFi）、中间设备静默丢弃，TCP 层没有感知，连接看起来还是 OPEN，但收发都断（⚠️ 症状：「界面长时间不更新，但连接状态显示正常」）；② ⭐ 中间设备会断「空闲连接」（Nginx 默认 60s、负载均衡可能 300s）。做法：定时发心跳（如每 25~30 秒，⭐ 要小于中间设备的最短空闲超时）+ 「超过 N 秒没收到『任何消息』（包括服务端心跳/业务消息）就主动 close 并重连」。两个取舍：间隔太短 → 移动端耗电 + 服务端压力；太长 → 检测慢（用户等很久才发现要重连）。⚠️ 另注意：浏览器 WebSocket API 不能发 ping 帧、也读不到 pong → 所以要用「应用层心跳」（发一条 {'type':'__ping'}）。

3. 「断线期间的消息丢了怎么办？」
   两条路线：① ⭐ 「序号 + 补发」——服务端给每条消息一个单调递增的序号，客户端记「各频道最后收到的 seq」，重连时带上它 → 服务端补发缺失的消息（⭐ SSE 下这件事是白送的：Last-Event-ID）；② ⭐ 「重连后全量刷新」——更简单但「贵」（适合「状态小、变化频繁」）。三个必须处理的边界：① ⭐ 「缺口太大」（离线几小时）→ 服务端应返回「需要全量刷新」的信号（__resync），而不是「补发十万条」；② ⭐ 「推送可能重复」（重连补发 + 原消息）→ 客户端要按消息 id/seq 去重；③ ⭐ 更根本的设计：「推送只作『有变化』的提示，真正的数据从 REST/GraphQL 拉」——这样「丢了推送」最坏只是「晚一点知道」，不会「数据错」。

4. 「多标签页同时打开会怎样？」
   ⚠️ 两个问题：① 连接数 ×N（每个标签页都建一个连接 → 服务端成本 ×N）；② ⭐ 同一条消息被处理 N 次（典型症状：Toast 弹 N 次、埋点重复上报、动画重复播放）。解法：⭐ Leader 选举（Web Locks API，见 3.14）+ BroadcastChannel——只有一个标签页（Leader）维护连接，收到消息后通过 BroadcastChannel 广播给其它标签页（其它标签页「只监听广播，不建连接」）。⚠️ 两个细节：① Leader 切换时（原 Leader 关闭）要「新 Leader 立刻建连 + 补偿」；② 非 Leader 标签页的「发送」也要走「BroadcastChannel 转给 Leader」或「直接走 POST」（⭐ 后者更简单，因为「发送」本来就是 HTTP 更合适）。

5. 「后台标签页有什么问题？」
   浏览器会对后台标签页做节能节流：① 定时器降频（心跳/重连定时器不准）；② ⭐ WebSocket 可能被挂起或断开；③ requestAnimationFrame 停止。后果：⚠️ 「用户切回标签页时，看到的是旧数据」（因为「断线期间的消息都错过了」）。解法：⭐ visibilitychange 恢复可见时做三件事：① 检查连接状态（readyState）；② 如果已断 → 立即重连（⭐ 不走退避，因为用户就在眼前等着）；③ ⭐ 如果连接还在 → 主动要一次「补偿/全量刷新」（因为「后台期间可能错过消息」，⭐ 这也是 3.14 里「恢复可见时拉一次」的同一条纪律）。

6. 「服务端做实时有什么架构约束？」
   四个：① ⭐ 连接是有状态的 → 水平扩展变难——「5000 个客户端分布在 7 个实例上」时，「给用户 A 推消息」怎么知道 A 连在哪个实例？✅ 三种做法：广播（简单但消息量 ×实例数）、⭐ 路由表（Redis 记 userId → instanceId）、⭐ 消息队列/PubSub（Redis Pub/Sub、Kafka、NATS——主流做法）；⚠️ 「单机内存里的连接表」正是「本地能跑、上线就错乱」的原因。② ⭐ 「房间/频道」的粒度——推荐「每用户一个频道 + 服务端按权限决定推什么」（⭐ 客户端不需要「声明订阅什么」，也就没有越权订阅的风险）。③ ⭐ 推送可能失败/重复——实时通道不是可靠传输，所以要让「推送只作提示，数据从 REST/GraphQL 拉」。④ ⭐ 背压与限流——「10 万客户端 × 每秒 10 条 = 100 万消息/秒」→ ✅ 合并/降采样（同一实体 100ms 内多次变更只推最后一次，⭐ 极有效）+ 按用户限流 + 「大房间改定向推送」。

7. 「为什么说『推送即数据』是坏设计？」
   因为它把「不可靠的通道」当成了「可靠的数据源」：① ⭐ 可能丢——「发送成功」不等于「客户端收到」（客户端可能正好在断线窗口，即使有补偿「也可能有缺口」）；② ⭐ 可能重复（重连补发 + 原消息）；③ 可能乱序（多连接/多实例下没有全局顺序）。后果：⚠️ 「数据错了」而不是「晚一点知道」——而「数据错」的排查成本远高于「慢」。✅ 正确设计：⭐ 「推送只带『哪个实体变了』（甚至只有一个 id），真正的数据『从 REST/GraphQL 拉』」——于是「丢了推送」最坏只是「晚一点知道」（⭐ 而且「重连时的补偿」也简化成「拉一次」）。顺带的好处：① 消息体积小（⭐ 移动网络友好）；② 权限校验只有一处（拉数据时校验，不用在推送里做）；③ 推送协议简单（不需要「可靠投递」的复杂机制）。

8. 「实时功能怎么测？」
   五层：① 单元测试——「退避算法」「消息解析（SSE 规范）」「序号缺口检测」「消息去重」都是纯逻辑，⭐ 最容易测也最该测；② ⭐ 连接生命周期的模拟——用「假 WebSocket/假 EventSource」模拟「断开 → 退避 → 重连 → 补偿」，断言「重连时带上了 lastSeq」；③ ⭐ 消息补偿的测试——「故意丢几条」→ 断言「客户端发现缺口并触发全量刷新」；④ 端到端——Playwright 里「开两个页面」验证「多标签页只一个连接 + 消息不重复处理」；⑤ ⭐ 混沌/故障演练——「服务端重启 30 秒」→ 观察「客户端是否能自动恢复且数据不错」（⭐ 这类「断网/重启/超时」的演练比「正常路径的测试」更有价值）。另外：监控「连接数/建连速率/异常关闭率/消息延迟」——⭐ 「连接泄漏」是实时服务的经典隐患（打满文件描述符）。

一句话速记

选型：⭐ 只推送 → SSE（默认）（普通 HTTP 长连接、内建重连、⭐ 重连自动带 Last-Event-ID → 断线补偿白送；⚠️ 单向 + EventSource 不能自定义头）、双向高频 → WebSocket（⚠️ 重连/心跳/协议/Upgrade 全自己写）、要求低 → 轮询；⭐ 五个必备件：① 重连用「指数退避 + 抖动 + 上限」（❌ 立刻重连会「重连风暴」）② 心跳（⭐ onclose 不可靠——「半开连接」与「中间设备断空闲连接」都不会触发它；间隔 25~30s 且小于设备超时；用「应用层心跳」因为浏览器 WS API 不能发 ping 帧）且「超时未收到任何消息 → 主动重连」③ 断线补偿（⭐ **序号 + 补发，缺口太大就 __resync 全量刷新，并按消息 id 去重）④ 多标签页用 Leader 选举（Web Locks）+ BroadcastChannel（否则「Toast 弹 N 次、埋点重复、连接数 ×N」）⑤ 后台标签页会被挂起 → visibilitychange 时「检查连接 + 立即重连（不走退避）+ 主动补偿」（网络恢复同理）；⭐ **「推送只作提示、数据从 REST/GraphQL 拉」（推送会丢/重复/乱序，不能当可靠传输——否则「数据错」而不是「晚一点知道」）；发送侧「连不上就入队 + 重连补发」（同 7.12），且客户端→服务端的命令走 POST 而不是 WS（能获得超时/重试/幂等/可观测性）；服务端四个约束：有状态连接的水平扩展（广播 / 路由表 / ⭐ PubSub——⚠️ 单机内存连接表 = 本地能跑上线就错乱）、房间粒度推荐「每用户一频道 + 服务端按权限推」、推送非可靠传输、背压与合并降采样（同实体 100ms 只推最后一次）；认证用 cookie（同源）或「连接后首发 token」（⚠️ 别放 query，会进日志）；SSE 在 HTTP/1.1 下占同域连接数（用 HTTP/2，或减少连接数）；测试重点：退避/解析/缺口检测等纯逻辑 + 「断开→重连→补偿」的模拟 + 混沌演练 + 连接数监控。