7.8 轮询与实时数据

一句话：实时数据有四种方案——轮询（简单但浪费）、SSE（服务端单向推送，浏览器自动重连）、WebSocket（双向、需要自己做心跳与重连）、长轮询（兼容性最好但服务器压力大）；选择依据是「是否需要双向 + 实时性要求 + 基础设施成本」。

一、四种方案对比

选择的三个问题：
① 需要「客户端主动推消息给服务端」吗？→ 需要 → WebSocket
② 只需要「服务端推给客户端」吗？      → 是   → SSE（更简单）
③ 数据变化频率很低（分钟级）吗？      → 是   → 轮询就够

⚠️ 常见误区：
· 「实时」就上 WebSocket —— 很多场景 SSE 或轮询更合适（更简单、更省）
· 用 WebSocket 做「请求-响应」（应该用 HTTP）
· 忽略「断线重连」「心跳」「消息去重」这些必须处理的细节

二、轮询

// ✅ 方案一：TanStack Query 的 refetchInterval（推荐）
function Notifications() {
  const { data } = useQuery({
    queryKey: ['notifications'],
    queryFn: fetchNotifications,
    refetchInterval: 30_000,                 // 每 30 秒重新请求
    refetchIntervalInBackground: false,      // ⭐ 标签页不可见时暂停（省资源）
    staleTime: 0,                             // 每次都取最新
  });

  return <Bell count={data?.length ?? 0} />;
}

// ✅ 方案二：setInterval + 清理（手写）
function usePolling<T>(fetcher: () => Promise<T>, interval: number, enabled = true) {
  const [data, setData] = useState<T | null>(null);
  const fetcherRef = useRef(fetcher);
  useEffect(() => { fetcherRef.current = fetcher; });          // ⭐ 保存最新回调

  useEffect(() => {
    if (!enabled) return;

    let cancelled = false;
    let timer: number;

    const tick = async () => {
      try {
        const result = await fetcherRef.current();
        if (!cancelled) setData(result);
      } catch {
        // 轮询失败静默处理（下次再试）
      }
      if (!cancelled) timer = window.setTimeout(tick, interval);   // ⭐ 完成后才排下一次
    };

    tick();
    return () => {
      cancelled = true;
      clearTimeout(timer);                      // ⭐ 必须清理
    };
  }, [interval, enabled]);

  return data;
}

⚠️ 轮询的三个要点：
① 用 setTimeout 递归而不是 setInterval
   —— 避免「上一个请求还没回来就发下一个」（请求堆积）
② 组件卸载/依赖变化时必须清理（否则内存泄漏 + 在已卸载组件里 setState）
③ 标签页不可见时暂停（可见性 API 或 refetchIntervalInBackground: false）
④ 轮询失败要「静默重试」而不是弹错误 toast（否则用户会看到大量提示）

⑤ 优化：用「增量」轮询（带 since 参数只取变化）或「ETag/If-None-Match」
   → 服务端返回 304 时几乎不消耗带宽

// ④ 的写法：失败静默，只有「连续失败 N 次」才提示
const failCountRef = useRef(0);

const tick = async () => {
  try {
    setData(await fetcherRef.current());
    failCountRef.current = 0;
  } catch {
    failCountRef.current++;
    if (failCountRef.current >= 3) setError('连接异常');
  }
};

三、SSE（Server-Sent Events）

// ✅ 服务端：设置 text/event-stream，持续写入
// Express 示例
app.get('/api/events', (req, res) => {
  res.writeHead(200, {
    'Content-Type': 'text/event-stream',
    'Cache-Control': 'no-cache',
    Connection: 'keep-alive',
  });

  const send = (event: string, data: unknown) => {
    res.write(`event: ${event}\n`);
    res.write(`data: ${JSON.stringify(data)}\n\n`);      // ⚠️ 必须以两个换行结尾
  };

  const timer = setInterval(() => send('tick', { time: Date.now() }), 1000);
  req.on('close', () => clearInterval(timer));
});

// ✅ 客户端：用原生 EventSource
function useSSE<T>(url: string) {
  const [data, setData] = useState<T | null>(null);
  const [status, setStatus] = useState<'connecting' | 'open' | 'error'>('connecting');

  useEffect(() => {
    const es = new EventSource(url, { withCredentials: true });   // ⭐ 跨域带 cookie

    es.onopen = () => setStatus('open');
    es.onerror = () => setStatus('error');           // ⚠️ 浏览器会自动重连（无需手动）
    es.onmessage = (e) => setData(JSON.parse(e.data));            // 默认事件

    // 自定义事件（对应服务端的 event: 名字）
    es.addEventListener('notification', (e) => {
      const payload = JSON.parse((e as MessageEvent).data);
      addNotification(payload);
    });

    return () => es.close();                          // ⭐ 必须关闭
  }, [url]);

  return { data, status };
}

SSE 的四个特点：
① 【单向】服务端 → 客户端（客户端要发消息仍走普通 HTTP）
② 【自动重连】浏览器内置重连（默认约 3 秒），可以用 `retry:` 字段调整
③ 【文本协议】基于 HTTP，无需特殊服务端/代理支持（但 Nginx 需要关缓冲）
④ 【断点续传】支持 `Last-Event-ID`（服务端可以据此补发丢失的消息）

⚠️ 三个坑：
① HTTP/1.1 下单域名最多 6 个并发连接（SSE 会长期占用一个）→ HTTP/2 可避免
② 代理/网关可能「缓冲」响应 → 要通过 `X-Accel-Buffering: no` 等方式关闭
③ 需要「心跳」防止中间设备断开空闲连接（服务端定期发注释行 `: ping`）

四、WebSocket

// ✅ 封装一个「带心跳与自动重连」的 WebSocket Hook
export function useWebSocket<T>(url: string, onMessage?: (msg: T) => void) {
  const [status, setStatus] = useState<'connecting' | 'open' | 'closed'>('connecting');
  const wsRef = useRef<WebSocket | null>(null);
  const retryRef = useRef(0);
  const heartbeatRef = useRef<number>();
  const reconnectTimerRef = useRef<number>();
  const onMessageRef = useRef(onMessage);
  useEffect(() => { onMessageRef.current = onMessage; });   // ⭐ 稳定回调

  useEffect(() => {
    let closedByUs = false;                                 // ⭐ 区分「主动关闭」与「异常断开」

    const connect = () => {
      const ws = new WebSocket(url);
      wsRef.current = ws;
      setStatus('connecting');

      ws.onopen = () => {
        setStatus('open');
        retryRef.current = 0;                               // 重置重试计数

        // 心跳：每 25 秒发一次 ping（防中间设备断开空闲连接）
        heartbeatRef.current = window.setInterval(() => {
          if (ws.readyState === WebSocket.OPEN) ws.send(JSON.stringify({ type: 'ping' }));
        }, 25_000);
      };

      ws.onmessage = (e) => {
        try {
          const msg = JSON.parse(e.data) as T;
          if ((msg as { type?: string }).type === 'pong') return;   // 心跳响应忽略
          onMessageRef.current?.(msg);
        } catch {
          // 忽略非 JSON 消息
        }
      };

      ws.onclose = () => {
        setStatus('closed');
        clearInterval(heartbeatRef.current);
        if (closedByUs) return;

        // ⭐ 指数退避重连（上限 30 秒）+ 抖动
        const delay = Math.min(1000 * 2 ** retryRef.current, 30_000) + Math.random() * 500;
        retryRef.current++;
        reconnectTimerRef.current = window.setTimeout(connect, delay);
      };

      ws.onerror = () => {
        ws.close();                                          // 交给 onclose 处理重连
      };
    };

    connect();

    return () => {
      closedByUs = true;                                     // ⭐ 避免卸载后仍重连
      clearInterval(heartbeatRef.current);
      clearTimeout(reconnectTimerRef.current);
      wsRef.current?.close();
    };
  }, [url]);

  const send = useCallback((msg: unknown) => {
    if (wsRef.current?.readyState === WebSocket.OPEN) {
      wsRef.current.send(JSON.stringify(msg));
    }
  }, []);

  return { status, send };
}

// 使用
function Chat() {
  const { status, send } = useWebSocket<ChatMessage>(`wss://api.example.com/chat?token=${token}`, (msg) => {
    appendMessage(msg);
  });

  return (
    <>
      {status !== 'open' && <span className="badge">连接中…</span>}
      <input onKeyDown={(e) => e.key === 'Enter' && send({ type: 'chat', text: e.currentTarget.value })} />
    </>
  );
}

⚠️ WebSocket 必须自己处理的五件事：

① 心跳（ping/pong）——否则空闲连接会被代理/负载均衡断开
② 自动重连——断线后要重连（**指数退避 + 抖动**，避免服务端重连风暴）
③ 区分「主动关闭」与「异常断开」——组件卸载时不要再重连
④ 消息格式与协议约定——`{ type, payload }` 或自定义协议（含消息 ID 用于去重/确认）
⑤ 鉴权——WebSocket 握手可以带 cookie/query token，但要注意「长连接的 token 过期」（需要刷新或重连）

进阶（视需求）：
⑥ 消息确认与重发（至少一次投递）
⑦ 消息去重（用消息 ID）
⑧ 断线期间的消息补偿（重连后用「最后收到的 ID」拉取遗漏）
⑨ 多标签页共享一个连接（用 SharedWorker / BroadcastChannel）

五、把实时数据接入缓存

// ✅ 推荐：WebSocket 收到消息后「写进 Query 缓存」，让 UI 自动更新
function useOrderRealtime(orderId: string) {
  const queryClient = useQueryClient();

  useWebSocket<OrderUpdate>(`wss://api/orders/${orderId}`, (msg) => {
    // 方式一：直接更新缓存（知道确切的新数据）
    queryClient.setQueryData<Order>(['order', orderId], (old) =>
      old ? { ...old, ...msg.patch } : old
    );

    // 方式二：让缓存失效（让库去重新拉，保证完整）
    queryClient.invalidateQueries({ queryKey: ['order', orderId] });
  });
}

// ✅ 轮询与实时结合：列表轮询 + 详情实时
function OrderList() {
  // 列表用轮询（低频）
  const { data } = useQuery({
    queryKey: ['orders'],
    queryFn: fetchOrders,
    refetchInterval: 60_000,
  });
  // 详情用 WebSocket 推送
}

⭐ 三种「实时 → 缓存」的策略：

① 消息里带「完整新数据」→ setQueryData 直接覆盖（最省请求）
② 消息里带「增量/patch」→ setQueryData 合并（注意深合并的字段范围）
③ 消息只是「通知有变化」→ invalidateQueries 让库重新拉（最简单可靠）

⚠️ 「通知式」最稳（不会出现「本地合并与服务端不一致」），
   但会多一次请求；「完整数据式」最快但要求服务端推送完整对象。

六、常见坑

// ① 用 setInterval 轮询但不清理 → 内存泄漏 + 卸载后 setState
// ② 用 setInterval 而不是「setTimeout 递归」→ 请求堆积（接口慢时）
// ③ 轮询失败就弹 toast → 用户看到大量错误提示（应该静默重试）
// ④ 标签页不可见时继续轮询（浪费流量与电量）
// ⑤ EventSource 忘记 close（组件卸载后仍在接收消息）
// ⑥ WebSocket 忘记心跳（空闲连接被代理断开，表现为「无规律掉线」）
// ⑦ WebSocket 断线后「立刻无限重连」（打垮服务端）→ 必须指数退避 + 抖动
// ⑧ 组件卸载后「仍在重连」（因为没区分「主动关闭」）
// ⑨ onmessage 里用了「闭包捕获的旧 state」（stale closure，见 2.13）
// ⑩ 用 WebSocket 做「请求-响应」（应该用 HTTP，或加消息 ID 实现 RPC）
// ⑪ 没有「消息去重」（重连后重复收到消息 → 列表出现重复项）
// ⑫ 没有「断线提示」（用户以为数据是最新的）
// ⑬ Nginx/网关缓冲导致 SSE「消息不实时到达」
// ⑭ 在 StrictMode 下「建立两个 WebSocket 连接」（因为 effect 双执行 → 必须写清理）
// ⑮ 把 WebSocket 实例放在「组件 state」里（每次渲染都变）→ 应该用 ref
// ⑯ 多标签页各自开一个连接（浪费；可用 SharedWorker 共享）
// ⑰ token 过期后 WebSocket 仍连着（不会自动带新 token）→ 需要「过期后重连」
// ⑱ 忘记处理「服务端主动断开」（如踢下线）→ 应提示并跳登录

// ⑭ 的说明：StrictMode 下 effect 双执行 → 会建立两个连接、但第一个会被清理关闭
useEffect(() => {
  const ws = new WebSocket(url);
  ws.onmessage = handle;
  return () => ws.close();          // ⭐ 有清理 → 第二个才是「活的」
}, [url]);
// 现象：开发环境会看到两次连接（一次立刻断开）；生产环境只有一次

七、面试延伸

1. 「实时数据有哪几种方案？怎么选？」
   四种：① 轮询——客户端定时请求，实现最简单但「空请求多、服务端压力大」，适合「数据变化不频繁」（如每 30 秒检查通知）；② 长轮询——客户端发起请求后服务端「挂起」直到有数据或超时，实时性更好但占用连接，适合「需要兼容老环境」；③ SSE——服务端单向推送，基于 HTTP 与 EventSource，浏览器自动重连、实现简单，适合「通知、进度、日志、行情」；④ WebSocket——全双工，实时性最好，但需要自己处理「心跳、重连、协议、鉴权」，适合「聊天、协作编辑、游戏、交易」。选择标准：需要双向 → WebSocket；只需服务端推送 → SSE；变化很慢 → 轮询。

2. 「SSE 和 WebSocket 有什么区别？」
   五点：① 方向——SSE 是单向（服务端 → 客户端），WebSocket 是双向；② 协议——SSE 基于 HTTP + 文本协议（text/event-stream），WebSocket 是独立的 ws:// 协议（需要 HTTP 升级握手）；③ 重连——SSE 浏览器自动重连（可用 retry: 配置，支持 Last-Event-ID 断点续传），WebSocket 要自己实现；④ 传输内容——SSE 只能传文本（二进制要自己编码），WebSocket 支持二进制；⑤ 复杂度与基础设施——SSE 实现简单但要小心代理缓冲与 HTTP/1.1 连接数限制，WebSocket 需要额外处理心跳与协议设计。

3. 「WebSocket 为什么要心跳？」
   因为「TCP 连接空闲时，中间的代理、负载均衡、NAT 网关可能会静默断开」（很多设备有「空闲超时」，如 60 秒）。此时客户端与服务端都不知道连接已经断了（没有 FIN/RST），表现为「消息发不出去但界面看起来正常」。心跳（定期 ping/pong）能：① 保持连接活跃（防止被断开）；② 检测死连接（一段时间没收到 pong 就主动重连）。实践中常用「客户端定时发 ping（如 25 秒）+ 服务端回 pong + 超时未收到 pong 就重连」的策略。

4. 「WebSocket 断线重连要注意什么？」
   四件事：① 指数退避 + 抖动——不要「断开就立刻重连」（会造成重连风暴打垮服务端），应该 delay = min(2^n × 1000, 30000) + random()；② 区分「主动关闭」与「异常断开」——组件卸载时主动 close() 之后不应该再重连（用一个标记位）；③ 重连成功后要「补齐状态」——如重新拉取数据、或用「最后收到的消息 ID」让服务端补发遗漏；④ 给出 UI 反馈（「连接中…」「已断开，正在重连」），因为用户需要知道数据可能不是最新的。

5. 「轮询应该注意什么？」
   四点：① 用 setTimeout 递归而不是 setInterval——避免「上一个请求没回来就发下一个」导致请求堆积；② 必须清理（组件卸载/依赖变化时 clearTimeout、置取消标记）；③ 标签页不可见时暂停（用 refetchIntervalInBackground: false 或 document.visibilityState），省流量与电量；④ 失败静默重试（轮询失败是常态，弹 toast 会变成噪音；可以「连续失败 N 次」才提示）。此外可用「增量轮询」（带 since 参数）或「ETag/If-None-Match」（304 几乎无开销）来优化。

6. 「怎么把实时数据接入 React 状态？」
   三种策略：① 消息带完整新数据 → queryClient.setQueryData(key, newData) 直接覆盖（最省请求）；② 消息带增量/patch → setQueryData(key, old => ({ ...old, ...patch }))（注意合并范围，避免与服务端不一致）；③ 消息只是「通知有变化」 → queryClient.invalidateQueries({ queryKey }) 让库重新拉取（最简单可靠，代价是多一次请求）。对于「结构复杂、字段多」的数据，推荐策略③（不容易出错）；对「简单的开关/计数」可以用策略①。

7. 「SSE 有什么坑？」
   四点：① 代理/网关缓冲——Nginx 等默认会缓冲响应，导致消息「不实时到达」，需要配置 proxy_buffering off / X-Accel-Buffering: no；② HTTP/1.1 连接数限制——单域名最多 6 个并发连接，SSE 长期占用一个（用 HTTP/2 可避免）；③ 需要心跳——长时间无数据时中间设备可能断开连接（服务端可以定期发注释行 : ping）；④ 单向——客户端要发消息必须另走 HTTP（不像 WebSocket 可以双向）。

8. 「为什么不直接用 WebSocket 做所有请求？」
   因为 WebSocket 不是「更好的 HTTP」，它有明显的额外成本：① 没有内置的「请求-响应」语义（需要自己设计消息 ID、匹配响应、超时、重试）；② 没有 HTTP 的成熟生态（缓存、状态码、CORS、压缩、代理/session affinity、可观测性都要自己处理）；③ 连接状态管理复杂（心跳、重连、多标签页共享、鉴权与 token 过期）；④ 调试困难（不像 HTTP 可以在 Network 面板里直观看到）。所以正确做法是「CRUD 用 HTTP（配合 TanStack Query），实时推送用 WebSocket/SSE」，两者各司其职。

一句话速记

实时数据四方案：轮询（最简单、变化慢时够用；用 setTimeout 递归、必须清理、不可见时暂停、失败静默）、SSE（单向服务端推送、基于 EventSource、浏览器自动重连 + Last-Event-ID 断点续传、要注意代理缓冲与 HTTP/1.1 连接数）、WebSocket（双向、必须自己处理心跳、指数退避重连、区分主动关闭、消息协议/去重/鉴权）、长轮询（兼容老环境）；需要双向才上 WebSocket，只推服务端就用 SSE；实时数据接入状态优选「invalidateQueries（通知式，最稳）」，其次才是「setQueryData 覆盖/patch」；StrictMode 下会双执行 effect（所以要写清理）；CRUD 仍用 HTTP，别用 WebSocket 做请求-响应。