实时 React Hooks

使用 SSE、WebSocket 和 SWR 在 React 应用中实现实时数据的生产级模式。

安装

OpenClaw / Moltbot / Clawbot

npx clawhub@latest install realtime-react-hooks

何时使用

• 需要实时数据更新的 React 应用
• 显示实时指标的仪表盘
• 聊天界面、通知系统
• 任何需要无需刷新即可更新的 UI

模式 1：SSE Hook

import { useEffect, useRef, useState, useCallback } from 'react';

interface UseSSEOptions<T> {
  url: string;
  onMessage?: (data: T) => void;
  onError?: (error: Event) => void;
  enabled?: boolean;
}

export function useSSE<T>({
  url,
  onMessage,
  onError,
  enabled = true,
}: UseSSEOptions<T>) {
  const [data, setData] = useState<T | null>(null);
  const [isConnected, setIsConnected] = useState(false);
  const eventSourceRef = useRef<EventSource | null>(null);

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

    const eventSource = new EventSource(url);
    eventSourceRef.current = eventSource;

    eventSource.onopen = () => {
      setIsConnected(true);
    };

    eventSource.onmessage = (event) => {
      try {
        const parsed = JSON.parse(event.data) as T;
        setData(parsed);
        onMessage?.(parsed);
      } catch (e) {
        console.error('SSE 解析错误:', e);
      }
    };

    eventSource.onerror = (error) => {
      setIsConnected(false);
      onError?.(error);
    };

    return () => {
      eventSource.close();
      eventSourceRef.current = null;
    };
  }, [url, enabled]);

  const close = useCallback(() => {
    eventSourceRef.current?.close();
    setIsConnected(false);
  }, []);

  return { data, isConnected, close };
}

模式 2：带重连机制的 WebSocket Hook

interface UseWebSocketOptions {
  url: string;
  onMessage?: (data: unknown) => void;
  reconnect?: boolean;
  maxRetries?: number;
}

export function useWebSocket({
  url,
  onMessage,
  reconnect = true,
  maxRetries = 5,
}: UseWebSocketOptions) {
  const [isConnected, setIsConnected] = useState(false);
  const wsRef = useRef<WebSocket | null>(null);
  const retriesRef = useRef(0);

  const connect = useCallback(() => {
    const ws = new WebSocket(url);
    wsRef.current = ws;

    ws.onopen = () => {
      setIsConnected(true);
      retriesRef.current = 0;
    };

    ws.onmessage = (event) => {
      try {
        const data = JSON.parse(event.data);
        onMessage?.(data);
      } catch {
        onMessage?.(event.data);
      }
    };

    ws.onclose = () => {
      setIsConnected(false);
      if (reconnect && retriesRef.current < maxRetries) {
        retriesRef.current++;
        const delay = Math.min(1000 * 2 ** retriesRef.current, 30000);
        setTimeout(connect, delay);
      }
    };

    ws.onerror = () => {
      ws.close();
    };
  }, [url, onMessage, reconnect, maxRetries]);

  useEffect(() => {
    connect();
    return () => wsRef.current?.close();
  }, [connect]);

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

  return { isConnected, send };
}

模式 3：支持实时更新的 SWR

import useSWR from 'swr';
import { useEffect } from 'react';

export function useRealtimeData<T>(
  key: string,
  fetcher: () => Promise<T>
) {
  const { data, mutate, ...rest } = useSWR(key, fetcher);

  // 订阅实时更新
  useEffect(() => {
    const eventSource = new EventSource(`/api/events/${key}`);

    eventSource.onmessage = (event) => {
      const update = JSON.parse(event.data);

      // 乐观更新缓存
      mutate((current) => {
        if (!current) return update;
        return { ...current, ...update };
      }, false); // false = 不重新验证
    };

    return () => eventSource.close();
  }, [key, mutate]);

  return { data, mutate, ...rest };
}

模式 4：订阅 Hook

interface UseSubscriptionOptions {
  channels: string[];
  onEvent: (channel: string, data: unknown) => void;
}

export function useSubscription({ channels, onEvent }: UseSubscriptionOptions) {
  const { send, isConnected } = useWebSocket({
    url: '/api/ws',
    onMessage: (msg: any) => {
      if (msg.type === 'event') {
        onEvent(msg.channel, msg.data);
      }
    },
  });

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

    // 订阅频道
    channels.forEach((channel) => {
      send({ type: 'subscribe', channel });
    });

    return () => {
      channels.forEach((channel) => {
        send({ type: 'unsubscribe', channel });
      });
    };
  }, [channels, isConnected, send]);

  return { isConnected };
}

模式 5：连接状态指示器

export function ConnectionStatus({ isConnected }: { isConnected: boolean }) {
  return (
    <div className="flex items-center gap-2">
      <span
        className={cn(
          'size-2 rounded-full',
          isConnected ? 'bg-success animate-pulse' : 'bg-destructive'
        )}
      />
      <span className="text-xs text-muted-foreground">
        {isConnected ? '已连接' : '已断开'}
      </span>
    </div>
  );
}

相关技能

• 元技能: [ai/skills/meta/realtime-dashboard/](../../meta/realtime-dashboard/) — 完整的实时仪表盘指南
• [resilient-connections](../resilient-connections/) — 重试逻辑
• [design-systems/animated-financial-display](../../design-systems/animated-financial-display/) — 数字动画效果

绝对不要这样做

• 永远记得清理资源 — 在组件卸载时始终关闭连接
• 永远不要无限重连 — 使用最大重试次数配合指数退避策略
• 永远不要不加 try/catch 解析数据 — 服务器可能发送格式错误的数据
• 永远不要直接修改并触发重新验证 — 使用 mutate(data, false) 实现乐观更新
• 永远不要忽略连接状态 — 向用户展示当前是否已断开连接

快速参考

// SSE
const { data, isConnected } = useSSE({ url: '/api/events' });

// WebSocket
const { isConnected, send } = useWebSocket({
  url: 'wss://api.example.com/ws',
  onMessage: (data) => console.log(data),
});

// SWR + 实时数据
const { data } = useRealtimeData('metrics', fetchMetrics);

// 订阅
useSubscription({
  channels: ['user:123', 'global'],
  onEvent: (channel, data) => updateState(channel, data),
});