在前端系统里，只要业务开始涉及“实时状态”，就一定会遇到一个问题：前端到底应该怎么接收服务端的实时数据？

最常见的方案有三种：轮询、WebSocket、SSE。

轮询最简单，但它的本质是前端不断问服务端“有没有新数据”。如果数据变化不频繁，就会产生大量无效请求；如果轮询间隔太长，实时性又不够好。

WebSocket 能力最强，它是双向通信，适合聊天室、协同编辑、在线游戏、撮合系统这类需要客户端和服务端频繁双向交互的场景。但如果前端只是需要持续接收服务端推送，用 WebSocket 有时候会显得过重。

SSE，也就是 Server-Sent Events，刚好适合一种很常见的场景：服务端持续向前端推送数据，前端只负责接收和更新 UI。

比如：

行情更新
订单簿更新
近期成交更新
账户余额更新
订单状态更新
持仓状态更新
任务状态更新
通知消息推送

这些数据大多数都是“服务端产生变化，然后推给前端”。

前端并不需要通过同一条连接频繁向服务端发送消息。所以在这种场景下，SSE 通常比 WebSocket 更简单，也比轮询更实时。

这篇文章不只是讲 SSE 的概念，而是从真实前端工程角度讲：如何使用 @microsoft/fetch-event-source 把 SSE 封装成可维护、可扩展、能和 React Query / Jotai / Redux 配合的业务 Hook。

---

1. 为什么需要 SSE

在一个交易、账户或者 Web3 DApp 类型的系统里，很多状态不是用户手动刷新页面才能看到的，而是应该由服务端主动推送。

例如，一个用户发起提现后，后端可能需要等待链上交易确认。

这个过程里，状态可能从：

PENDING
CONFIRMING
SUCCESS
FAILED

不断变化。

如果前端只靠普通 HTTP 请求，那么通常有两种办法。

第一种是用户手动刷新，这显然体验不好。

第二种是前端轮询：

setInterval(async () => {
  const account = await getAccount();
  updateAccount(account);
}, 3000);

轮询虽然能解决问题，但它有明显缺点。

首先，轮询会产生大量无效请求。即使账户数据没有任何变化，前端还是会每隔几秒请求一次。

其次，轮询的实时性取决于间隔。如果间隔是 5 秒，那么用户可能最多要等 5 秒才能看到最新状态。如果间隔改成 1 秒，请求压力又会上升。

再次，不同业务数据的轮询频率很难统一。订单簿可能需要很高频率，账户余额可能低频就够了，历史记录甚至只需要在关键事件后重新获取。

SSE 的价值就在这里：服务端有数据变化时主动推送给前端，前端收到后更新状态。

它的通信模型可以理解成：

Browser
  -> 建立 SSE 连接
Server
  -> 持续推送事件
Browser
  -> 收到事件后更新 UI / 状态 / 缓存

在这种模式下，前端不需要一直问服务端“有没有变化”，而是等服务端告诉前端“已经变化了”。

这更适合实时状态类业务。

---

2. 为什么不用原生 EventSource

浏览器原生支持 EventSource。最简单的用法是：

const eventSource = new EventSource('/event/market');

eventSource.onmessage = event => {
  console.log(event.data);
};

eventSource.onerror = error => {
  console.error(error);
};

这个 API 很简单，上手成本低。如果只是做一个公开的、无需认证的、简单的消息流，原生 EventSource 完全够用。

但在真实工程里，原生 EventSource 很快会遇到几个限制。

第一个限制是：不方便自定义请求头。

很多私有 SSE 接口需要携带 token。例如：

headers: {
  'ACCESS-TOKEN': accessToken,
}

但是原生 EventSource 不能像 fetch 那样直接传 headers。这对用户私有数据流非常不方便。

比如：

/event/user
/event/account
/event/orders
/event/positions

这些接口通常都需要认证。如果不能带 token，就很难用原生 EventSource 做干净的鉴权设计。

第二个限制是：错误处理能力比较弱。

真实项目里，你往往需要区分：

401：登录过期
403：无权限
404：stream 不存在
429：请求太频繁
500：服务端错误
网络断开
服务端临时不可用

原生 EventSource 的错误处理不够细。它会自动重连，但你很难优雅地控制哪些错误应该重试，哪些错误应该直接停止。

第三个限制是：不方便结合 AbortController 管理生命周期。

在 React 里，一个 SSE 连接通常要随着组件挂载、卸载、参数变化、登录状态变化而创建或关闭。比如：

用户切换 instrument -> 关闭旧连接，打开新连接
用户退出登录 -> 关闭私有 SSE
组件卸载 -> abort 当前连接
stream 参数为空 -> 不连接

原生 EventSource 当然可以 .close()，但它不像 fetch 一样天然适配 AbortController。

第四个限制是：重试策略不够灵活。

真实工程里，你可能希望这样：

最多重试 3 次
401 / 403 不重试
429 可以稍后重试
500 可以重试
手动 abort 后不再重试

原生 EventSource 的自动重连机制比较黑盒，不适合做复杂控制。

所以我更倾向使用：

import {fetchEventSource} from '@microsoft/fetch-event-source';

它的核心好处是：用接近 fetch 的方式处理 SSE。

你可以传：

headers
signal
onopen
onmessage
onerror
onclose
openWhenHidden

也就是说，它既保留了 SSE 的事件流能力，又让连接管理更像普通 HTTP 请求一样可控。

---

3. SSE 基本数据模型

SSE 的本质是服务端持续向浏览器推送文本事件。

在前端工程里，我们通常不会直接处理原始字符串，而是约定一个统一的数据格式。

一个比较常见的 SSE 消息模型是：

type SSEMessage<T> = {
  type: string;
  data: T;
};

也就是说，每条消息至少包含两个字段：

type：当前事件类型
data：当前事件携带的数据

比如近期成交更新：

{
  "type": "recent_trade",
  "data": [
    {
      "price": "3000",
      "amount": "1.2",
      "side": "buy"
    }
  ]
}

订单簿更新：

{
  "type": "order_book",
  "data": {
    "bids": [],
    "asks": []
  }
}

账户面板更新：

{
  "type": "dashboard",
  "data": {
    "balance": "1000",
    "available": "900",
    "frozen": "100"
  }
}

订单更新：

{
  "type": "open_order",
  "data": {
    "order_id": "123",
    "status": "FILLED"
  }
}

这样设计以后，前端收到消息时只需要做一件事：根据 type 分发到不同的状态更新逻辑。

例如：

onmessage(event) {
  const {type, data} = JSON.parse(event.data);

  switch (type) {
    case 'recent_trade':
      setRecentTrades(data);
      break;

    case 'order_book':
      setOrderbook(data);
      break;

    case 'dashboard':
      setUserAccount(data);
      break;

    case 'open_order':
      setOrders(data);
      break;

    default:
      console.log('Unhandled SSE event type:', type);
  }
}

所以，SSE 前端封装的核心不是“怎么打开连接”，而是：

怎么建立连接
怎么识别事件类型
怎么分发事件数据
怎么更新本地状态
怎么和 React Query 缓存同步
怎么控制连接生命周期

这也是为什么在真实项目里，不应该把 SSE 代码随便写在组件里。

4. fetchEventSource 基本模板

先看一个最基础的模板。

import {
  EventStreamContentType,
  fetchEventSource,
} from '@microsoft/fetch-event-source';

const controller = new AbortController();

await fetchEventSource('/event/user', {
  signal: controller.signal,

  headers: {
    'ACCESS-TOKEN': accessToken,
  },

  async onopen(response) {
    if (
      response.ok &&
      response.headers.get('content-type') === EventStreamContentType
    ) {
      console.log('SSE connected');
      return;
    }

    throw new Error('SSE connection failed');
  },

  onmessage(event) {
    const message = JSON.parse(event.data);
    console.log(message);
  },

  onerror(error) {
    console.error('SSE error:', error);
  },

  onclose() {
    console.log('SSE closed');
  },

  openWhenHidden: true,
});

这里有几个关键点。

signal 用于关闭连接。

const controller = new AbortController();

controller.abort();

当组件卸载、用户退出登录、stream 参数变化时，都可以通过 abort() 主动关闭连接。

headers 用于私有 SSE 鉴权。

headers: {
  'ACCESS-TOKEN': accessToken,
}

这也是 fetchEventSource 相比原生 EventSource 的核心优势之一。

onopen 用来校验服务端响应。

async onopen(response) {
  if (
    response.ok &&
    response.headers.get('content-type') === EventStreamContentType
  ) {
    return;
  }

  throw new Error('SSE connection failed');
}

这里不只是判断 HTTP 状态码，还要判断 content-type 是否为 text/event-stream。

如果后端返回的是普通 JSON、HTML 错误页或者其他内容，那么这条连接就不是合法 SSE 连接。

onmessage 用来处理消息。

onmessage(event) {
  const message = JSON.parse(event.data);
}

onerror 用来处理错误和重试。

onerror(error) {
  console.error(error);
}

注意：fetchEventSource 在没有 throw 的情况下可以继续重试。如果你在 onerror 里 throw，就会停止重试。

onclose 用来处理连接关闭。

onclose() {
  console.log('SSE closed');
}

openWhenHidden 表示页面隐藏时是否保持连接。

openWhenHidden: true

对于交易系统、行情系统、订单系统来说，即使用户切到别的浏览器标签页，也可能希望继续接收数据，所以可以设为 true。

---

5. 缓存 controller / stream / retry 状态

在 React 里封装 SSE 时，一个很重要的问题是：连接状态应该放在哪里？

不能所有状态都用 useState。

比如这些状态：

AbortController
当前连接的 stream
当前 retry 次数
内部连接状态

它们更多是“连接生命周期控制状态”，不一定需要触发 UI 重渲染。

所以更合适的做法是用 useRef 缓存。

可以定义一个连接管理对象：

type ConnectionManager = {
  isConnected: boolean;
  controller: AbortController | null;
  retryCount: number;
  currentStream: string | null;
};

然后：

const connectionManager = useRef<ConnectionManager>({
  isConnected: false,
  controller: null,
  retryCount: 0,
  currentStream: null,
});

这样做的好处是，内部连接状态变化不会导致组件频繁 render。

如果 UI 需要显示连接状态，再单独用 useState：

const [isConnected, setIsConnected] = useState(false);

最终分层是：

useRef
  -> 保存 controller、retryCount、currentStream 等内部连接状态

useState
  -> 保存 UI 需要展示的连接状态，例如 isConnected

例如清理连接时：

const cleanup = useCallback(() => {
  const manager = connectionManager.current;

  if (manager.controller) {
    manager.controller.abort();
  }

  manager.controller = null;
  manager.isConnected = false;
  manager.retryCount = 0;
  manager.currentStream = null;

  setIsConnected(false);
}, []);

这段代码的含义很清晰：

如果当前存在连接，就 abort
清空 controller
重置连接状态
重置重试次数
清空当前 stream
同步 UI 连接状态

真实工程里，SSE 的 bug 很多都来自连接生命周期混乱，比如：

重复连接同一个 stream
旧连接没有关闭
组件卸载后还在推送
用户退出登录后私有 SSE 还在连接
切换 instrument 后新旧数据混在一起

所以缓存并管理 controller / stream / retry 是 SSE 封装里非常关键的一步。

---

6. 可复用 useSSE 模板

下面是一个可以作为基础层使用的 useSSE 模板。

这个 Hook 不关心具体业务，只负责连接、断开、重试、消息回调。

import {useCallback, useEffect, useRef, useState} from 'react';
import {
  EventStreamContentType,
  fetchEventSource,
} from '@microsoft/fetch-event-source';

type ConnectionManager = {
  isConnected: boolean;
  controller: AbortController | null;
  retryCount: number;
  currentStream: string | null;
};

type UseSSEOptions<T> = {
  url: string | null;
  headers?: Record<string, string>;
  maxRetries?: number;
  openWhenHidden?: boolean;
  onData: (message: T) => void;
  onConnected?: () => void;
  onDisconnected?: () => void;
  onError?: (error: unknown) => void;
};

export function useSSE<T>({
  url,
  headers,
  maxRetries = 3,
  openWhenHidden = true,
  onData,
  onConnected,
  onDisconnected,
  onError,
}: UseSSEOptions<T>) {
  const managerRef = useRef<ConnectionManager>({
    isConnected: false,
    controller: null,
    retryCount: 0,
    currentStream: null,
  });

  const [isConnected, setIsConnected] = useState(false);

  const cleanup = useCallback(() => {
    const manager = managerRef.current;

    if (manager.controller) {
      manager.controller.abort();
    }

    manager.controller = null;
    manager.isConnected = false;
    manager.retryCount = 0;
    manager.currentStream = null;

    setIsConnected(false);
    onDisconnected?.();
  }, [onDisconnected]);

  const connect = useCallback(async () => {
    if (!url) {
      cleanup();
      return;
    }

    const manager = managerRef.current;

    if (manager.isConnected && manager.currentStream === url) {
      return;
    }

    cleanup();

    const controller = new AbortController();

    manager.controller = controller;
    manager.currentStream = url;
    manager.retryCount = 0;
    manager.isConnected = false;

    try {
      await fetchEventSource(url, {
        headers,
        signal: controller.signal,
        openWhenHidden,

        async onopen(response) {
          if (
            response.ok &&
            response.headers.get('content-type') === EventStreamContentType
          ) {
            manager.isConnected = true;
            manager.retryCount = 0;

            setIsConnected(true);
            onConnected?.();

            return;
          }

          const isClientError =
            response.status >= 400 &&
            response.status < 500 &&
            response.status !== 429;

          throw new Error(
            isClientError ? 'SSE client error' : 'SSE server error',
          );
        },

        onmessage(event) {
          const message = JSON.parse(event.data) as T;
          onData(message);
        },

        onerror(error) {
          manager.retryCount += 1;
          manager.isConnected = false;

          setIsConnected(false);
          onError?.(error);

          if (manager.retryCount >= maxRetries) {
            cleanup();
            throw error;
          }
        },

        onclose() {
          manager.isConnected = false;
          setIsConnected(false);
          onDisconnected?.();
        },
      });
    } catch (error) {
      manager.isConnected = false;
      setIsConnected(false);
      onError?.(error);
    }
  }, [
    url,
    headers,
    maxRetries,
    openWhenHidden,
    onData,
    onConnected,
    onDisconnected,
    onError,
    cleanup,
  ]);

  useEffect(() => {
    connect();

    return () => {
      cleanup();
    };
  }, [connect, cleanup]);

  return {
    isConnected,
    reconnect: connect,
    disconnect: cleanup,
  };
}

这个 Hook 的职责非常单一：管理一条 SSE 连接。

它不应该知道什么是订单簿，不应该知道什么是账户，也不应该直接写 React Query 的 invalidateQueries。

它只是基础连接层。

业务逻辑应该放在更上层的 Hook 里，比如：

useTradeSSE
useUserSSE
useGlobalSSE

这样后续维护会清晰很多。

---

7. 模板优点

这个 useSSE 模板的优势主要有几个。

第一，连接生命周期可控。

组件挂载时连接，组件卸载时关闭。URL 变化时，会关闭旧连接并创建新连接。

第二，支持私有 SSE。

因为 fetchEventSource 支持 headers，所以可以很自然地传 token。

headers: {
  'ACCESS-TOKEN': accessToken,
}

第三，支持主动 abort。

controller.abort();

当用户退出登录、切换 stream、离开页面时，都可以主动关闭连接。

第四，支持最大重试次数。

if (manager.retryCount >= maxRetries) {
  cleanup();
  throw error;
}

这可以避免异常情况下无限重连。

第五，避免重复连接。

if (manager.isConnected && manager.currentStream === url) {
  return;
}

这能防止同一个 stream 被重复订阅。

第六，业务扩展更干净。

基础 Hook 只负责连接。业务 Hook 负责解释消息类型和更新状态。

分层之后，代码会变成：

useSSE
  -> 只负责连接

useTradeSSE
  -> 解释 recent_trade / order_book

useUserSSE
  -> 解释 dashboard / open_order / position

useGlobalSSE
  -> 统一启动市场 SSE 和用户 SSE

这才是更接近真实项目的写法。

---

8. 真实工程落地方式

在真实工程里，不建议每个组件都自己创建 SSE 连接。

错误写法通常是：

Orderbook 组件里创建 /event/trade
RecentTrades 组件里也创建 /event/trade
AccountPanel 组件里创建 /event/user
PositionTable 组件里也创建 /event/user

这样很容易导致重复连接、状态分散、清理混乱。

更合理的做法是把 SSE 分成几层：

基础连接层
  -> useSSE

业务订阅层
  -> useTradeSSE
  -> useUserSSE
  -> useGlobalSSE

状态更新层
  -> Jotai / Zustand / Redux
  -> React Query invalidateQueries

UI 展示层
  -> AccountPanel
  -> Orderbook
  -> RecentTrades
  -> PositionTable

也就是说，组件不直接关心 SSE。

组件只关心状态：

订单簿数据从哪里读
账户数据从哪里读
持仓数据从哪里读
loading 怎么展示
error 怎么展示

SSE 则负责在后台持续把服务端数据同步到前端状态层。

这就形成了一个清晰的架构：

Server SSE
  -> useSSE
  -> useTradeSSE / useUserSSE / useGlobalSSE
  -> Jotai / Zustand / Redux / React Query
  -> UI Components

下面分别看三个典型业务 Hook。

---

9. useTradeSSE 负责当前交易标的的实时数据。

比如当前选中的 instrument 是：

BTC-240705-60000-C

那么它连接的地址可能是：

/event/trade?stream=BTC-240705-60000-C

它主要处理两类事件：

recent_trade
order_book

数据流可以理解成：

selectedOptionAtom
  -> currentInstrument
  -> /event/trade?stream=${instrument}
  -> recent_trade
  -> setRecentTrades(data)

selectedOptionAtom
  -> currentInstrument
  -> /event/trade?stream=${instrument}
  -> order_book
  -> setOrderbook(data)

伪代码如下：

type TradeSSEMessage =
  | {
      type: 'recent_trade';
      data: RecentTrade[];
    }
  | {
      type: 'order_book';
      data: Orderbook;
    };

export function useTradeSSE() {
  const selectedOption = useAtomValue(selectedOptionAtom);

  const {setRecentTrades} = useRecentTrades();
  const {setOrderbook} = useOrderbook();

  const currentInstrument = selectedOption?.instrument ?? null;

  const url = currentInstrument
    ? `${SSE_URL}/event/trade?stream=${currentInstrument}`
    : null;

  return useSSE<TradeSSEMessage>({
    url,

    onData(message) {
      switch (message.type) {
        case 'recent_trade':
          setRecentTrades(message.data);
          break;

        case 'order_book':
          setOrderbook(message.data);
          break;

        default:
          console.log('Unhandled trade SSE message:', message);
      }
    },
  });
}

这个 Hook 的关键点是：当前 instrument 变化时，要关闭旧连接并打开新连接。

比如用户从 BTC 期权切换到 ETH 期权：

旧 stream: /event/trade?stream=BTC-240705-60000-C
新 stream: /event/trade?stream=ETH-240705-3000-C

如果不关闭旧连接，就可能出现 BTC 和 ETH 的订单簿数据混在一起的问题。

所以 useTradeSSE 的核心职责是：

监听当前 instrument
根据 instrument 建立 SSE
切换 instrument 时重连
收到 recent_trade 后更新近期成交
收到 order_book 后更新订单簿
卸载时关闭连接

---

10.useUserSSE 负责登录用户的私有实时数据。

它通常连接：

/event/user

并且需要携带 token：

headers: {
  'ACCESS-TOKEN': accessToken,
}

它处理的数据一般包括：

dashboard：账户面板更新
open_order：挂单更新
position：持仓更新
user_instrument：用户相关合约数据更新

数据流可以理解成：

isClientLoginAtom + accessToken
  -> /event/user
  -> dashboard
  -> update account

isClientLoginAtom + accessToken
  -> /event/user
  -> open_order
  -> update orders
  -> invalidate open order / order history

isClientLoginAtom + accessToken
  -> /event/user
  -> position
  -> update positions
  -> invalidate trade history

伪代码如下：

type UserSSEMessage =
  | {
      type: 'dashboard';
      data: AccountDashboard;
    }
  | {
      type: 'open_order';
      data: OpenOrder[];
    }
  | {
      type: 'position';
      data: Position[];
    }
  | {
      type: 'user_instrument';
      data: UserInstrument[];
    };

export function useUserSSE() {
  const queryClient = useQueryClient();

  const isLogin = useAtomValue(isClientLoginAtom);
  const accessToken = useAccessToken();

  const setUserAccount = useUpdateUserAccount();
  const setOrders = useUpdateOrders();
  const setPositions = useUpdatePositions();
  const setUserInstrument = useUpdateUserInstrument();

  const url = isLogin && accessToken ? `${SSE_URL}/event/user` : null;

  return useSSE<UserSSEMessage>({
    url,

    headers: accessToken
      ? {
          'ACCESS-TOKEN': accessToken,
        }
      : undefined,

    onData(message) {
      switch (message.type) {
        case 'dashboard':
          setUserAccount(message.data);
          break;

        case 'open_order':
          setOrders(message.data);

          queryClient.invalidateQueries({
            queryKey: ['trade', 'order', 'open'],
          });

          queryClient.invalidateQueries({
            queryKey: ['trade', 'order', 'history'],
          });

          break;

        case 'position':
          setPositions(message.data);

          queryClient.invalidateQueries({
            queryKey: ['trade', 'history'],
          });

          break;

        case 'user_instrument':
          setUserInstrument(message.data);
          break;

        default:
          console.log('Unhandled user SSE message:', message);
      }
    },
  });
}

这里有一个重要原则：用户未登录时不要连接私有 SSE。

const url = isLogin && accessToken ? `${SSE_URL}/event/user` : null;

当用户退出登录后，url 变成 null，基础 useSSE 会自动 cleanup。

这可以避免一个严重问题：用户已经退出登录，但旧的私有 SSE 连接还在后台接收数据。

useUserSSE 的核心职责是：

登录后建立私有 SSE
携带 access token
退出登录后关闭连接
收到账户事件后更新账户状态
收到订单事件后更新订单状态并失效订单缓存
收到持仓事件后更新持仓状态并失效成交缓存

---

11. useGlobalSSE 是应用级 SSE 入口，适合放在 App Provider 或全局 Layout 中。

它通常负责两件事：

启动用户私有 SSE
启动公共市场 SSE

例如：

export function useGlobalSSE() {
  useUserSSE();
  useMarketSSE();
}

公共市场 SSE 通常按 underlying index 订阅。

比如当前市场是：

BTC-240705

那么连接可能是：

/event/market?stream=BTC-240705

它处理的事件可能包括：

instrument
Underlying
instrument_greek
instrument_quote

数据流可以理解成：

currentUnderlyingIndexAtom
  -> /event/market?stream=${underlyingIndex}
  -> instrument
  -> update option map

currentUnderlyingIndexAtom
  -> /event/market?stream=${underlyingIndex}
  -> instrument_greek
  -> update greek map

currentUnderlyingIndexAtom
  -> /event/market?stream=${underlyingIndex}
  -> instrument_quote
  -> update quote map

currentUnderlyingIndexAtom
  -> /event/market?stream=${underlyingIndex}
  -> Underlying
  -> invalidate underlying query

公共市场数据和用户私有数据不同。市场数据可能频率更高，尤其是报价、希腊值、订单簿这类数据。

如果每条 SSE 消息都直接 setState，可能会导致大量 render。

所以市场 SSE 常见优化是：

useRef 缓存最新数据
requestAnimationFrame 批量更新
throttle 限制 React Query invalidate 频率

例如：

const latestMarketDataRef = useRef({
  instrument: null,
  underlying: null,
  instrumentGreek: null,
  instrumentQuote: null,
});

const pendingMarketUpdateRef = useRef(false);

收到消息时，不立刻更新 UI，而是先放进 ref：

latestMarketDataRef.current = {
  ...latestMarketDataRef.current,
  [updateType]: data,
};

然后安排下一帧统一更新：

if (!pendingMarketUpdateRef.current) {
  pendingMarketUpdateRef.current = true;

  requestAnimationFrame(() => {
    batchMarketUpdate();
  });
}

批量更新函数：

function batchMarketUpdate() {
  const currentData = latestMarketDataRef.current;

  if (currentData.instrument) {
    setOptionsMap(currentData.instrument);
  }

  if (currentData.instrumentGreek) {
    setOptionsMapByGreek(currentData.instrumentGreek);
  }

  if (currentData.instrumentQuote) {
    setOptionsMapByQuote(currentData.instrumentQuote);
  }

  if (currentData.underlying) {
    throttleInvalidateUnderlying();
  }

  latestMarketDataRef.current = {
    instrument: null,
    underlying: null,
    instrumentGreek: null,
    instrumentQuote: null,
  };

  pendingMarketUpdateRef.current = false;
}

这样可以把同一帧内的多条 SSE 消息合并成一次状态更新。

useGlobalSSE 的核心职责是：

作为全局实时数据入口
启动用户私有 SSE
启动市场公共 SSE
监听当前 underlying index
切换市场时重连
对高频行情做批量更新
必要时触发 React Query 缓存失效

---

12. SSE + React Query 配合

SSE 和 React Query 不是互相替代的关系。

React Query 负责：

请求接口
缓存接口数据
控制 loading / error / stale
分页查询
重新请求
缓存失效

SSE 负责：

接收服务端实时事件
通知前端某些数据变化了
把部分实时数据直接写入状态
触发某些 query 重新获取

在真实工程里，二者应该配合使用。

一般来说，有两种处理方式。

第一种：SSE 推送的是完整数据，前端可以直接更新本地状态。

例如：

order_book
recent_trade
dashboard
position map

这些数据通常可以直接写进 Jotai / Zustand / Redux。

case 'order_book':
  setOrderbook(data);
  break;

case 'recent_trade':
  setRecentTrades(data);
  break;

case 'dashboard':
  setUserAccount(data);
  break;

第二种：SSE 只是告诉前端“某类数据变化了”，然后前端让 React Query 重新请求。

例如：

历史订单
历史成交
账户流水
分页列表
统计数据

这类数据不适合通过 SSE 全量推送，因为它们可能有分页、筛选、排序、时间范围等条件。

更合适的方式是：

queryClient.invalidateQueries({
  queryKey: ['trade', 'order', 'history'],
});

或者：

queryClient.invalidateQueries({
  queryKey: ['account', 'history', address],
});

需要注意的是，高频数据不要频繁 invalidate。

比如 underlying 行情可能频繁变化，如果每条消息都：

queryClient.invalidateQueries({
  queryKey: ['market', 'underlying'],
});

可能会导致大量重复请求。

这时应该加 throttle：

const throttleInvalidate = throttle(
  () => {
    queryClient.invalidateQueries({
      queryKey: ['market', 'underlying'],
    });
  },
  3000,
  {
    leading: true,
    trailing: false,
  },
);

一个比较好的原则是：

实时展示型数据
  -> SSE 直接更新本地状态

服务端权威型数据
  -> SSE 触发 React Query invalidate

高频数据
  -> 批量更新 / 节流 invalidate

低频关键状态
  -> 可以直接 invalidate

所以，SSE 不是为了替代接口请求，而是为了让前端知道什么时候应该更新。

---

13. 推荐目录结构

一个比较清晰的目录结构可以这样设计：

src/
  hooks/
    sse/
      use-sse.ts
      use-global-sse.ts
      use-user-sse.ts
      use-trade-sse.ts
      use-market-sse.ts

  lib/
    services/
      query-keys.ts
      config.ts

  state/
    account.ts
    orders.ts
    positions.ts
    options.ts
    trades.ts

  components/
    account-panel.tsx
    orderbook-panel.tsx
    recent-trades.tsx
    position-table.tsx

每一层职责要清楚。

use-sse.ts：

基础连接能力
不关心业务
只处理连接、断开、重试、headers、signal

use-trade-sse.ts：

当前 instrument 的交易数据订阅
处理 recent_trade / order_book

use-user-sse.ts：

用户私有数据订阅
处理 dashboard / open_order / position

use-market-sse.ts：

公共市场数据订阅
处理 instrument / quote / greek / underlying

use-global-sse.ts：

全局启动入口
一般挂在 App Provider 或 Layout

state/*：

保存实时状态
例如账户、订单、持仓、期权、成交、订单簿

query-keys.ts：

统一管理 React Query keys
方便 SSE 里精准 invalidate

例如：

export const queryKeys = {
  account: (address: string) => ['account', address] as const,

  accountHistory: (address: string) =>
    ['account', 'history', address] as const,

  openOrders: () => ['trade', 'order', 'open'] as const,

  orderHistory: () => ['trade', 'order', 'history'] as const,

  tradeHistory: () => ['trade', 'history'] as const,

  underlying: () => ['market', 'underlying'] as const,
};

这样在 SSE 里就不要到处手写字符串：

queryClient.invalidateQueries({
  queryKey: queryKeys.orderHistory(),
});

这对后期维护很重要。

---

14. 后续扩展方向

当 SSE 封装稳定以后，还可以继续扩展几个方向。

第一个方向是连接状态面板。

比如在开发环境显示：

User SSE: connected
Market SSE: connected
Trade SSE: disconnected
Current stream: BTC-240705
Retry count: 1

这对调试非常有用。

第二个方向是统一日志系统。

不要到处写：

console.log('[SSE USER]: ...');
console.log('[SSE MARKET]: ...');

可以封装成：

sseLogger.info('USER', 'connected');
sseLogger.error('MARKET', error);

第三个方向是指数退避重连。

简单的最大重试次数是：

失败一次 -> 重试
失败两次 -> 重试
失败三次 -> 停止

更成熟的做法是：

1s 后重试
2s 后重试
4s 后重试
8s 后重试

这可以避免服务端故障时大量客户端同时重连。

第四个方向是 token 过期处理。

如果私有 SSE 返回 401，前端不应该一直重试，而应该：

停止 SSE
清除登录态
跳转登录
提示用户重新登录

第五个方向是消息 sequence 校验。

更成熟的服务端推送可以加：

type SSEMessage<T> = {
  type: string;
  sequence: number;
  timestamp: number;
  data: T;
};

前端记录上一次 sequence：

lastSequence = 100
newSequence = 101 -> 正常
newSequence = 103 -> 中间丢了一条

如果发现 sequence 不连续，就触发 React Query refetch，重新校准数据。

第六个方向是多 stream 管理。

比如一个页面同时订阅：

BTC market stream
ETH market stream
user account stream
trade stream
notification stream

这时可以抽象一个 SSE manager，统一维护多个连接。

第七个方向是页面可见性控制。

有些系统需要页面隐藏时继续连接，有些系统可以在页面隐藏时暂停连接，节省资源。

可以根据业务决定：

openWhenHidden: true

或者监听 document.visibilityState 动态暂停。

---

15. 总结

SSE 很适合服务端向前端单向推送实时数据的场景。

相比轮询，SSE 减少了无效请求，实时性更好。

相比 WebSocket，SSE 更简单，更适合行情、账户、订单、持仓、任务状态这类服务端主动推送的数据流。

在真实前端工程里，不建议把原生 EventSource 直接写进组件。原生 EventSource 虽然简单，但在私有鉴权、错误处理、重试控制、生命周期管理方面不够灵活。

更推荐使用：

import {fetchEventSource} from '@microsoft/fetch-event-source';

然后按分层方式封装：

fetchEventSource
  -> 基础连接层 useSSE
  -> 业务订阅层 useTradeSSE / useUserSSE / useGlobalSSE
  -> 状态层 Jotai / Zustand / Redux / React Query
  -> UI 展示层

其中：

useSSE
  -> 只负责连接、断开、重试、headers、signal

useTradeSSE
  -> 负责当前 instrument 的订单簿和近期成交

useUserSSE
  -> 负责用户账户、订单、持仓等私有数据

useGlobalSSE
  -> 负责全局市场订阅和统一启动入口

React Query
  -> 负责缓存、重新请求、服务端数据校准

一个稳定的 SSE 架构，不只是能“收到消息”，还要解决这些问题：

连接什么时候建立
连接什么时候关闭
重复连接怎么避免
stream 切换怎么处理
用户退出登录怎么清理
错误怎么重试
高频数据怎么批量更新
哪些数据直接更新本地状态
哪些数据触发 React Query invalidate

把这些问题处理好以后，SSE 就不再是一段零散的事件监听代码，而会变成一个可以长期维护、可以扩展、可以复用的前端实时数据层。