From 4403380d49a1e8c994bae8ae5f2d064ced634f02 Mon Sep 17 00:00:00 2001 From: "Ray.Hao" <1490493387@qq.com> Date: Thu, 30 Jul 2026 18:43:45 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20SSE=20=E8=BF=9E=E6=8E=A5=E5=A2=9E?= =?UTF-8?q?=E5=8A=A0=20401/403=20=E4=B8=8D=E9=87=8D=E8=BF=9E=E3=80=81?= =?UTF-8?q?=E6=97=A0=20token=20=E5=AE=9A=E6=97=B6=E9=87=8D=E8=AF=95?= =?UTF-8?q?=E3=80=81=E6=B5=81=E5=BC=82=E5=B8=B8=E9=87=8D=E8=BF=9E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- package.json | 2 +- src/components/NoticeDropdown/useNotice.ts | 55 +++++++++- src/composables/index.ts | 4 +- src/composables/sse/index.ts | 21 ++-- src/composables/sse/sseTopics.ts | 17 +++ src/composables/sse/useDictSync.ts | 61 +++++------ src/composables/sse/useOnlineUsers.ts | 50 +++++++++ src/composables/sse/useSse.ts | 114 ++++++++++++++------- 8 files changed, 239 insertions(+), 85 deletions(-) create mode 100644 src/composables/sse/sseTopics.ts create mode 100644 src/composables/sse/useOnlineUsers.ts diff --git a/package.json b/package.json index f41aca07..f2ec3b03 100644 --- a/package.json +++ b/package.json @@ -1,7 +1,7 @@ { "name": "vue3-element-admin", "description": "Vue3 + Vite + TypeScript + Element-Plus 的后台管理模板，vue-element-admin 的 Vue3 版本", - "version": "4.8.1", + "version": "4.8.3", "private": true, "type": "module", "scripts": { diff --git a/src/components/NoticeDropdown/useNotice.ts b/src/components/NoticeDropdown/useNotice.ts index 8d7f4aff..6ffd374c 100644 --- a/src/components/NoticeDropdown/useNotice.ts +++ b/src/components/NoticeDropdown/useNotice.ts @@ -1,15 +1,16 @@ import { computed, onBeforeUnmount, onMounted, ref } from "vue"; import type { NoticeDetail, NoticeItem, NoticeQueryParams } from "@/api/system/notice"; import NoticeAPI from "@/api/system/notice"; -import { useSse } from "@/composables"; +import { useSse, SseTopics } from "@/composables"; import router from "@/router"; +/** 下拉面板每页展示条数 */ const PAGE_SIZE = 5; -const NOTICE_EVENT = "notice"; -const NOTICE_REVOKE_EVENT = "notice-revoke"; +/** 通知读取状态：0=未读，1=已读 */ type NoticeStatus = 0 | 1; +/** SSE 推送的新通知消息体 */ interface NoticeMessage { id: string; title: string; @@ -17,22 +18,38 @@ interface NoticeMessage { publishTime?: Date; } +/** SSE 推送的通知撤回消息体 */ interface NoticeRevokeMessage { id: string; } +/** + * 通知下拉面板的响应式数据与业务逻辑 + * 在组件挂载时拉取列表、建立 SSE 订阅，卸载时自动清理 + */ export function useNotice() { const { on } = useSse(); + /** 当前 Tab 下的通知列表（最多 PAGE_SIZE 条） */ const list = ref([]); + /** 未读通知总数（红点/角标数字） */ const unreadTotal = ref(0); + /** 当前激活的 Tab：0=未读，1=已读 */ const activeStatus = ref(0); + /** 查看详情时加载的完整通知数据 */ const detail = ref(null); + /** 详情弹窗可见性 */ const dialogVisible = ref(false); + /** 列表为空时的占位文案，根据当前 Tab 切换 */ const emptyText = computed(() => (activeStatus.value === 0 ? "暂无未读消息" : "暂无已读消息")); + /** SSE 订阅的取消函数集合，用于组件卸载时解绑 */ let stopSubscriptions: (() => void) | null = null; + /** + * 拉取通知分页列表 + * 查询未读 Tab 时同步更新 unreadTotal + */ async function fetchList(params?: Partial) { const query: NoticeQueryParams = { pageNum: 1, @@ -48,6 +65,7 @@ export function useNotice() { } } + /** 仅查询未读通知总数（不更新列表），用于切换到已读 Tab 后刷新角标 */ async function fetchUnreadTotal() { const page = await NoticeAPI.getMyNoticePage({ pageNum: 1, @@ -57,6 +75,10 @@ export function useNotice() { unreadTotal.value = page.total ?? 0; } + /** + * 切换未读/已读 Tab + * 同一 Tab 重复点击不重复请求 + */ async function switchStatus(status: NoticeStatus) { if (activeStatus.value === status) return; @@ -64,6 +86,10 @@ export function useNotice() { await fetchList(); } + /** + * 刷新数据 + * 未读 Tab：刷新列表即可；已读 Tab：额外刷新未读总数以更新角标 + */ async function refresh() { await Promise.all([ fetchList(), @@ -71,6 +97,14 @@ export function useNotice() { ]); } + /** + * 点击单条通知查看详情 + * 1. 标记原列表项是否为未读 + * 2. 拉取详情并打开弹窗 + * 3. 从当前列表中移除该项（下拉面板内不再显示） + * 4. 若为未读，本地角标 -1 + * 5. 刷新数据与角标 + */ async function read(id: string) { const item = list.value.find((notice: NoticeItem) => notice.id === id); const wasUnread = item?.isRead !== 1; @@ -85,6 +119,7 @@ export function useNotice() { await refresh(); } + /** 全部标为已读：调用接口 + 清空本地未读数 + 刷新列表 */ async function readAll() { if (unreadTotal.value <= 0) return; @@ -98,19 +133,28 @@ export function useNotice() { ElMessage.success("已全部标记为已读"); } + /** 跳转到通知列表页 */ function goMore() { router.push({ name: "MyNotice" }); } + /** + * 建立 SSE 实时推送订阅 + * - NOTICE 事件：新通知到达时插入列表头部、更新角标、弹出浏览器通知 + * - NOTICE_REVOKE 事件：通知被撤回时从列表中移除并更新角标 + * 重复调用会跳过，避免多次挂载时重复订阅 + */ function setupSubscription() { if (stopSubscriptions) return; - const stopNotice = on(NOTICE_EVENT, (data) => { + const stopNotice = on(SseTopics.NOTICE, (data) => { try { if (!data.id) return; unreadTotal.value += 1; + // 当前在已读 Tab 时不操作列表 if (activeStatus.value !== 0) return; + // 已存在则跳过（防重） if (list.value.some((item: NoticeItem) => item.id === data.id)) return; list.value.unshift({ @@ -124,6 +168,7 @@ export function useNotice() { isRead: 0, }); + // 超出 PAGE_SIZE 时截断尾部 if (list.value.length > PAGE_SIZE) { list.value.length = PAGE_SIZE; } @@ -139,7 +184,7 @@ export function useNotice() { } }); - const stopRevoke = on(NOTICE_REVOKE_EVENT, (data) => { + const stopRevoke = on(SseTopics.NOTICE_REVOKE, (data) => { try { if (!data.id) return; diff --git a/src/composables/index.ts b/src/composables/index.ts index 1f425bd7..297c1697 100644 --- 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"; diff --git a/src/composables/sse/index.ts b/src/composables/sse/index.ts index 74e6226d..d2b130cb 100644 --- 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"; diff --git a/src/composables/sse/sseTopics.ts b/src/composables/sse/sseTopics.ts new file mode 100644 index 00000000..aec8ad1e --- /dev/null +++ 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]; diff --git a/src/composables/sse/useDictSync.ts b/src/composables/sse/useDictSync.ts index e0acd61b..d174f9c2 100644 --- 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 | null = null; +let globalInstance: ReturnType | null = null; function createDictSyncComposable() { const dictStore = useDictStoreHook(); const sse = useSse(); - const messageCallbacks = ref([]); - + 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; } diff --git a/src/composables/sse/useOnlineUsers.ts b/src/composables/sse/useOnlineUsers.ts new file mode 100644 index 00000000..9901467a --- /dev/null +++ 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 | 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; +} diff --git a/src/composables/sse/useSse.ts b/src/composables/sse/useSse.ts index 9fa5ae47..7f901a29 100644 --- 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 | 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 | 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) => { 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 = (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();