refactor: SSE 连接增加 401/403 不重连、无 token 定时重试、流异常重连

2026-07-30 18:43:45 +08:00
parent 1ff46bc33b
commit 4403380d49
8 changed files with 239 additions and 85 deletions
--- a/src/composables/index.ts
+++ b/src/composables/index.ts
@@ -1,7 +1,7 @@
 // SSE 服务
 export { setupSse, cleanupSseServices } from "./sse";
-export { useSse, useDictSync, useOnlineCount, cleanupSse, SseConnectionState } from "./sse";
-export type { DictMessage, DictChangeMessage, DictChangeCallback } from "./sse";
+export { useSse, useDictSync, useOnlineUsers, cleanupSse, SseConnectionState, SseTopics } from "./sse";
+export type { DictChangeMessage, DictChangeCallback, SseTopic } from "./sse";

 // 表格相关
 export { useTableSelection } from "./useTableSelection";
--- a/src/composables/sse/index.ts
+++ b/src/composables/sse/index.ts
@@ -1,16 +1,19 @@
 import { useDictSync } from "./useDictSync";
-import { useOnlineCount } from "./useOnlineCount";
-import { cleanupSse } from "./useSse";
+import { useOnlineUsers } from "./useOnlineUsers";
+import { useSse, cleanupSse } from "./useSse";

 /**
 * 初始化所有 SSE 服务
 */
 export function setupSse() {
+  const sse = useSse();
+  sse.connect();
+
  const dictSync = useDictSync();
  dictSync.initialize();

-  const onlineCount = useOnlineCount();
-  onlineCount.initialize();
+  const onlineUsers = useOnlineUsers();
+  onlineUsers.initialize();
 }

 /**
@@ -20,13 +23,15 @@ export function cleanupSseServices() {
  const dictSync = useDictSync();
  dictSync.cleanup();

-  const onlineCount = useOnlineCount();
-  onlineCount.cleanup();
+  const onlineUsers = useOnlineUsers();
+  onlineUsers.cleanup();

  cleanupSse();
 }

 export { useDictSync } from "./useDictSync";
-export { useOnlineCount } from "./useOnlineCount";
+export { useOnlineUsers } from "./useOnlineUsers";
 export { useSse, cleanupSse, SseConnectionState } from "./useSse";
-export type { DictMessage, DictChangeMessage, DictChangeCallback } from "./useDictSync";
+export { SseTopics } from "./sseTopics";
+export type { DictChangeMessage, DictChangeCallback } from "./useDictSync";
+export type { SseTopic } from "./sseTopics";
--- a/src/composables/sse/sseTopics.ts
+++ b/src/composables/sse/sseTopics.ts
@@ -0,0 +1,17 @@
+/** SSE 事件名常量，与后端 SseTopics.java 一一对应 */
+export const SseTopics = {
+  /** 字典变更事件 */
+  DICT: "dict",
+  /** 在线用户数事件 */
+  ONLINE_USERS: "online-users",
+  /** 系统消息事件 */
+  SYSTEM: "system",
+  /** 心跳事件 */
+  PING: "ping",
+  /** 通知事件 */
+  NOTICE: "notice",
+  /** 通知撤回事件 */
+  NOTICE_REVOKE: "notice-revoke",
+} as const;
+
+export type SseTopic = (typeof SseTopics)[keyof typeof SseTopics];
--- a/src/composables/sse/useDictSync.ts
+++ b/src/composables/sse/useDictSync.ts
@@ -1,65 +1,61 @@
 import { useDictStoreHook } from "@/stores/dict";
 import { useSse } from "./useSse";
+import { SseTopics } from "./sseTopics";

+/** 字典变更消息体 */
 export interface DictChangeMessage {
+  /** 字典编码 */
  dictCode: string;
-  timestamp: number;
 }

-export type DictMessage = DictChangeMessage;
-
+/** 字典变更回调函数类型 */
 export type DictChangeCallback = (message: DictChangeMessage) => void;

-let singletonInstance: ReturnType<typeof createDictSyncComposable> | null = null;
+let globalInstance: ReturnType<typeof createDictSyncComposable> | null = null;

 function createDictSyncComposable() {
  const dictStore = useDictStoreHook();
  const sse = useSse();

-  const messageCallbacks = ref<DictChangeCallback[]>([]);
-
+  const callbacks: DictChangeCallback[] = [];
  let unsubscribe: (() => void) | null = null;

-  const handleDictChangeMessage = (data: DictChangeMessage) => {
+  // 处理字典变更消息：清除指定字典缓存，并通知所有已注册回调
+  const handleDictChange = (data: DictChangeMessage) => {
    const { dictCode } = data;
-
    if (!dictCode) {
      console.warn("[DictSync] 收到无效的字典变更消息：缺少 dictCode");
      return;
    }

    dictStore.removeDictItem(dictCode);
-
-    messageCallbacks.value.forEach((callback) => {
+    callbacks.forEach((cb) => {
      try {
-        callback(data);
-      } catch (error) {
-        console.error("[DictSync] 回调函数执行失败:", error);
+        cb(data);
+      } catch (err) {
+        console.error("[DictSync] 回调执行失败:", err);
      }
    });
  };

+  // 订阅 SSE 字典变更事件
  const initialize = () => {
-    sse.connect();
-    unsubscribe = sse.on("dict", handleDictChangeMessage);
+    unsubscribe = sse.on(SseTopics.DICT, handleDictChange);
  };

+  // 取消 SSE 订阅并清空所有回调
  const cleanup = () => {
-    if (unsubscribe) {
-      unsubscribe();
-      unsubscribe = null;
-    }
-    messageCallbacks.value = [];
+    unsubscribe?.();
+    unsubscribe = null;
+    callbacks.length = 0;
  };

-  const onDictChange = (callback: DictChangeCallback) => {
-    messageCallbacks.value.push(callback);
-
+  // 注册字典变更回调，返回取消注册函数
+  const onDictChange = (cb: DictChangeCallback) => {
+    callbacks.push(cb);
    return () => {
-      const index = messageCallbacks.value.indexOf(callback);
-      if (index !== -1) {
-        messageCallbacks.value.splice(index, 1);
-      }
+      const idx = callbacks.indexOf(cb);
+      if (idx !== -1) callbacks.splice(idx, 1);
    };
  };

@@ -74,10 +70,15 @@ function createDictSyncComposable() {

 /**
 * 字典同步组合式函数（单例模式）
+ *
+ * 监听 SSE 字典变更事件，收到变更时自动清除对应字典缓存，
+ * 并通知所有已注册的回调函数。
+ *
+ * @returns 字典同步实例，包含连接状态、初始化、清理和回调注册方法
 */
 export function useDictSync() {
-  if (!singletonInstance) {
-    singletonInstance = createDictSyncComposable();
+  if (!globalInstance) {
+    globalInstance = createDictSyncComposable();
  }
-  return singletonInstance;
+  return globalInstance;
 }
--- a/src/composables/sse/useOnlineUsers.ts
+++ b/src/composables/sse/useOnlineUsers.ts
@@ -0,0 +1,50 @@
+import { ref, readonly } from "vue";
+import { useSse } from "./useSse";
+import { SseTopics } from "./sseTopics";
+
+let globalInstance: ReturnType<typeof createOnlineUsersComposable> | null = null;
+
+function createOnlineUsersComposable() {
+  const onlineUserCount = ref(0);
+  const lastUpdateTime = ref(0);
+
+  const sse = useSse();
+
+  let unsubscribe: (() => void) | null = null;
+
+  const handleOnlineUsersMessage = (count: number) => {
+    if (!Number.isFinite(count) || count < 0) return;
+    onlineUserCount.value = count;
+    lastUpdateTime.value = Date.now();
+  };
+
+  const initialize = () => {
+    unsubscribe = sse.on(SseTopics.ONLINE_USERS, handleOnlineUsersMessage);
+  };
+
+  const cleanup = () => {
+    if (unsubscribe) {
+      unsubscribe();
+      unsubscribe = null;
+    }
+    onlineUserCount.value = 0;
+    lastUpdateTime.value = 0;
+  };
+
+  return {
+    onlineUserCount: readonly(onlineUserCount),
+    lastUpdateTime: readonly(lastUpdateTime),
+    isConnected: sse.isConnected,
+    connectionState: sse.connectionState,
+    initialize,
+    cleanup,
+  };
+}
+
+/** 在线用户数组合式函数（单例模式） */
+export function useOnlineUsers() {
+  if (!globalInstance) {
+    globalInstance = createOnlineUsersComposable();
+  }
+  return globalInstance;
+}
--- a/src/composables/sse/useSse.ts
+++ b/src/composables/sse/useSse.ts
@@ -1,27 +1,36 @@
 import { AuthStorage } from "@/utils/auth";

+/** SSE 连接配置选项 */
 export interface UseSseOptions {
-  url?: string; // SSE 连接地址，默认走 VITE_APP_BASE_API 代理
-  debug?: boolean; // 是否在控制台打印调试日志
-  connectionTimeout?: number; // 连接超时时间(ms)
-  /** 重连间隔基数，实际间隔 = min(基数 × 2^n, 最大间隔) */
+  /** SSE 连接地址，默认走 VITE_APP_BASE_API 代理 */
+  url?: string;
+  /** 是否在控制台打印调试日志 */
+  debug?: boolean;
+  /** 连接超时时间（ms），默认 10000 */
+  connectionTimeout?: number;
+  /** 重连间隔基数（ms），实际间隔 = min(基数 × 2^n, maxReconnectInterval) */
  reconnectInterval?: number;
-  maxReconnectInterval?: number; // 重连间隔上限(ms)
-  maxReconnectAttempts?: number; // 最大重试次数，超过后停止重连
+  /** 重连间隔上限（ms），默认 120000 */
+  maxReconnectInterval?: number;
+  /** 最大重试次数，超过后停止重连，默认 10 */
+  maxReconnectAttempts?: number;
 }

+/** SSE 事件处理器类型 */
 type EventHandler = (data: unknown) => void;

+/** SSE 流解析中间状态 */
 type SseParseState = {
  currentEvent: string;
  currentData: string;
  buffer: string;
 };

+/** SSE 连接状态 */
 export enum SseConnectionState {
-  DISCONNECTED = "DISCONNECTED", // 未连接
-  CONNECTING = "CONNECTING", // 连接中
-  CONNECTED = "CONNECTED", // 已连接
+  DISCONNECTED = "DISCONNECTED",
+  CONNECTING = "CONNECTING",
+  CONNECTED = "CONNECTED",
 }

 let globalInstance: ReturnType<typeof createSseConnection> | null = null;
@@ -59,27 +68,26 @@ function createSseConnection(options: UseSseOptions = {}) {
  };
  const logError = (...args: unknown[]) => console.error("[SSE]", ...args);

-  // 清理定时器并返回空值
-  const clearTimer = (timer: typeof connectionTimeoutTimer) => {
+  // 清除定时器并返回 null，用于链式赋值
+  const clearTimer = (timer: ReturnType<typeof setTimeout> | null): null => {
    if (timer) {
      clearTimeout(timer);
-      return null;
    }
-    return timer;
+    return null;
  };

-  // 重置重连计数和间隔
+  // 重置重连状态：次数归零、间隔恢复基数
  const resetReconnectState = () => {
    reconnectAttempts = 0;
    currentReconnectInterval = config.reconnectInterval;
  };

-  // 更新下一次重连间隔
+  // 指数退避：当前间隔翻倍，不超过上限
  const advanceReconnectState = () => {
    currentReconnectInterval = Math.min(currentReconnectInterval * 2, config.maxReconnectInterval);
  };

-  // 分发一条完整的 SSE 事件
+  // 分发 SSE 事件：先尝试 JSON.parse，失败则传原始字符串
  const flushSseEvent = (eventName: string, data: string) => {
    if (!data) return;
    const handlers = eventHandlers.get(eventName);
@@ -94,7 +102,7 @@ function createSseConnection(options: UseSseOptions = {}) {
    log(`收到事件[${eventName}]:`, data);
  };

-  // 解析单行 SSE 文本并更新当前事件状态
+  // 解析单行 SSE 数据：区分 event/data/注释/空行（触发分发）
  const handleSseLine = (line: string, state: SseParseState) => {
    if (line.startsWith(":")) return;
    if (line.startsWith("event:")) {
@@ -113,30 +121,42 @@ function createSseConnection(options: UseSseOptions = {}) {
    }
  };

-  // 持续读取 SSE 流并按行解析
+  // 持续读取流数据并按行解析，异常时触发重连
  const consumeSseStream = async (streamReader: ReadableStreamDefaultReader<Uint8Array>) => {
    const decoder = new TextDecoder();
    const state: SseParseState = { currentEvent: "message", currentData: "", buffer: "" };

-    while (true) {
-      const { done, value } = await streamReader.read();
-      if (done) {
-        connectionState.value = SseConnectionState.DISCONNECTED;
-        log("SSE 连接已关闭");
-        return;
+    try {
+      while (true) {
+        const { done, value } = await streamReader.read();
+        if (done) {
+          reader = null;
+          connectionState.value = SseConnectionState.DISCONNECTED;
+          log("SSE 连接已关闭");
+          return;
+        }
+
+        state.buffer += decoder.decode(value, { stream: true });
+        const lines = state.buffer.split("\n");
+        state.buffer = lines.pop() || "";
+
+        for (const line of lines) {
+          handleSseLine(line, state);
+        }
      }
-
-      state.buffer += decoder.decode(value, { stream: true });
-      const lines = state.buffer.split("\n");
-      state.buffer = lines.pop() || "";
-
-      for (const line of lines) {
-        handleSseLine(line, state);
+    } catch (err) {
+      reader = null;
+      connectionState.value = SseConnectionState.DISCONNECTED;
+      if (err instanceof Error && err.name === "AbortError") {
+        log("SSE 流读取已主动断开");
+      } else {
+        logError("SSE 流读取错误:", err);
+        scheduleReconnect();
      }
    }
  };

-  // 指数退避重连
+  // 调度重连：指数退避，达到上限或主动断开时停止
  const scheduleReconnect = () => {
    if (isManualDisconnect) return;
    if (config.maxReconnectAttempts > 0 && reconnectAttempts >= config.maxReconnectAttempts) {
@@ -148,12 +168,12 @@ function createSseConnection(options: UseSseOptions = {}) {
    log(`将在 ${currentReconnectInterval}ms 后重试（${reconnectAttempts}）`);

    reconnectTimer = setTimeout(() => {
-      connect();
      advanceReconnectState();
+      connect();
    }, currentReconnectInterval);
  };

-  // 建立 SSE 连接
+  // 建立连接：校验 token → fetch → 超时检测 → 消费流；401/403 不重连
  const connect = () => {
    isManualDisconnect = false;

@@ -168,14 +188,14 @@ function createSseConnection(options: UseSseOptions = {}) {

    const token = AuthStorage.getAccessToken();
    if (!token) {
-      log("未检测到有效令牌，跳过 SSE 连接");
+      log("未检测到有效令牌，稍后重试");
+      reconnectTimer = setTimeout(() => connect(), config.reconnectInterval);
      return;
    }

    connectionState.value = SseConnectionState.CONNECTING;
    abortController = new AbortController();

-    // 超时自动断开
    connectionTimeoutTimer = setTimeout(() => {
      if (connectionState.value === SseConnectionState.CONNECTING) {
        log("SSE 连接超时");
@@ -195,6 +215,12 @@ function createSseConnection(options: UseSseOptions = {}) {
    })
      .then((response) => {
        if (!response.ok) {
+          if (response.status === 401 || response.status === 403) {
+            isManualDisconnect = true;
+            connectionState.value = SseConnectionState.DISCONNECTED;
+            log(`SSE 连接被拒绝（HTTP ${response.status}），不再重连`);
+            return null;
+          }
          throw new Error(`HTTP ${response.status}`);
        }
        connectionTimeoutTimer = clearTimer(connectionTimeoutTimer);
@@ -219,7 +245,7 @@ function createSseConnection(options: UseSseOptions = {}) {
      });
  };

-  // 订阅事件，返回取消函数
+  // 订阅指定事件，返回取消订阅函数
  const on = <T = unknown>(eventName: string, handler: (data: T) => void): (() => void) => {
    if (!eventHandlers.has(eventName)) {
      eventHandlers.set(eventName, new Set());
@@ -239,7 +265,7 @@ function createSseConnection(options: UseSseOptions = {}) {
    };
  };

-  // 主动断开，不会触发重连
+  // 主动断开：清除定时器、取消流读取、中止请求，不触发重连
  const disconnect = () => {
    isManualDisconnect = true;
    connectionTimeoutTimer = clearTimer(connectionTimeoutTimer);
@@ -252,7 +278,7 @@ function createSseConnection(options: UseSseOptions = {}) {
    log("SSE 连接已断开");
  };

-  // 登出时调用，断开并释放所有资源
+  // 断开连接并清空所有事件订阅
  const cleanup = () => {
    disconnect();
    eventHandlers.clear();
@@ -269,6 +295,15 @@ function createSseConnection(options: UseSseOptions = {}) {
  };
 }

+/**
+ * SSE 连接组合式函数（单例模式）
+ *
+ * 基于 fetch + ReadableStream 实现，支持指数退避重连、
+ * 事件订阅/取消订阅、主动断开与资源清理。
+ *
+ * @param options - 连接配置选项
+ * @returns SSE 连接实例，包含连接状态、connect/disconnect/on/cleanup 方法
+ */
 export function useSse(options: UseSseOptions = {}) {
  if (!globalInstance) {
    globalInstance = createSseConnection(options);
@@ -276,6 +311,7 @@ export function useSse(options: UseSseOptions = {}) {
  return globalInstance;
 }

+/** 清理 SSE 单例：断开连接、清空订阅、释放全局引用 */
 export function cleanupSse() {
  if (globalInstance) {
    globalInstance.cleanup();