1. 为什么突然都在聊SSE从AI聊天打字机效果说起如果你这两年做过AI应用的前端或者刷过前端面试题大概率会碰上一个词SSEServer-Sent Events服务器推送事件。市面上铺天盖地的大模型流式输出打字机效果实时渲染回答底层基本都是它。最直观的场景就是你打开Kimi、文心一言、通义千问这类AI对话产品按下回车之后答案不是一整块蹦出来的而是一个字一个字往外蹦。这个体验背后的核心技术就是SSE。服务端通过一个长连接把大模型生成的内容分段推给浏览器前端每收到一小段就立刻追加到页面上用户的感知就是它正在打字。SSE不是新技术早年常用于股票行情推送、服务器日志实时滚动这类场景但真正让它重回聚光灯下的是AI大模型应用的爆发。原因也很简单大模型生成Token本身是流式的服务端边生成边推送客户端边接收边渲染配合起来天衣无缝。这篇文章就从零开始把SSE的来龙去脉讲清楚重点落在封装和实战上。内容分三块先说SSE和WebSocket的区别帮你搞清楚什么场景该选谁再深入原生EventSource API的局限性说明为什么单一原生API扛不住真实业务然后给出一套完整的封装方案包含取消请求、自动重连、事件分发、心跳保活这些真实项目里绕不开的点最后配一个AI对话渲染案例和一份高频问题排查清单。无论你是刚接触前端的新手还是在后台管理系统里被各种数据实时更新需求折磨的资深开发这篇文章的目标都是让你看完能直接动手写出自己的SSE封装而不是停留在用过EventSource的层面。2. SSE和WebSocket一对被经常搞混的兄弟很多同学第一次听到SSE的完整名字Server-Sent Events时第一反应就是这不就是WebSocket吗一开始学的时候我也这么想过。两个都是浏览器和服务器之间的长连接都能实现服务端主动推送但实际用起来差异非常大。2.1 连接方向与通信模式WebSocket是全双工通信客户端和服务端可以同时互相收发消息连接建立后通道是双向的。SSE是单工通信数据只能从服务端往客户端推客户端要发消息得另开一个普通的HTTP请求。全双工和单工的区别在AI对话场景里有很实际的影响。AI对话产品通常是你发送一条Prompt然后服务端持续Push内容数据流方向是单向的SSE天然匹配。但如果是一个在线协作白板用户A画一笔要同步给用户BB的操作又要回传给A这种双向实时交互就必须上WebSocket。用生活化的方式理解SSE像一个单向广播电台服务器是电台主播客户端是收音机主播说话你只能听着你想回话得打电话过去另发HTTP请求WebSocket像电话通话两边都能随时开口。选型的时候先问自己一句我这个功能需要双向实时通信吗不需要就别背WebSocket的重量。2.2 协议与实现成本WebSocket需要专门的通讯协议后端要支持协议升级HTTP Upgrade部署时还得考虑网关、负载均衡器的WebSocket代理配置一不小心就是各种握手失败。SSE就没有这么多弯弯绕它就是标准的HTTP协议在响应头里声明Content-Type: text/event-stream即可。这个差异直接影响了接入成本。我见过不少团队为了一个后台有数据前端推送的功能硬上WebSocket结果后端同事配Nginx配了一整天还得处理连接断开后的状态同步。同样的需求如果用SSE后端只需要在HTTP响应里不断写数据即可不需要处理额外的协议连接生命周期维护成本低一截。2.3 自动重连与断线恢复SSE的原生APIEventSource自带断线重连机制连接意外断开后浏览器会自动重新发起连接服务端还可以通过retry字段指定重连间隔。WebSocket没有这个原生能力断线后要自己实现重连逻辑搭配心跳检测踢掉假死连接。很多人忽略了这一点但这恰恰是SSE在稳定性上最讨喜的地方。做前端的人应该都有过这种经历WebSocket突然静默断线页面上一排灰色消息没人管用户刷新页面才发现全都断了。SSE有浏览器兜底配合Last-Event-ID还能断线续传少写很多代码。2.4 什么时候选SSE什么时候硬上WebSocket给一个实际上能用的选型总结纯推送场景AI流式输出、通知推送、日志流、行情刷新优先SSE双向交互场景IM聊天、协同编辑、白板协作、实时游戏选WebSocket需要兼容老浏览器必须上WebSocketSSE在IE里基本上是不可用的虽然现在还在用IE的已经不多了有一点经常被忽略WebSocket对服务器连接数的压力比SSE大得多SSE走普通HTTP连接配合现有负载均衡体系更平滑记住一个原则能用SSE解决的问题不要没事就上WebSocket这不是技术炫技的问题纯粹是成本问题。3. 原生EventSource能跑AI示例但扛不住真实业务很多AI应用前端的入门教程里SSE的写法就是一个new EventSource(url)加一个onmessage监听看起来非常简单。但你照着搬到真实项目里就会发现远远不够原因在于EventSource API有硬性限制。3.1 原生API的基本形态先看一个标准用法const source new EventSource(/api/chat/stream); // 监听默认消息类型 source.onmessage (event) { console.log(event.data); }; // 监听服务端自定义的事件类型 source.addEventListener(custom-event, (event) { console.log(event.data); }); // 错误处理 source.onerror (err) { console.error(连接异常, err); };服务端只需要这样做Content-Type: text/event-stream Cache-Control: no-cache Connection: keep-alive data: {content: 你好} data: {content: 我} data: {content: 是AI}每两段数据之间用空行分隔data:后面的内容会在前端event.data里拿到。协议本身简洁到有点原始但这种简洁恰恰带来了后面一堆问题。3.2 限制一只能GET请求无法自定义HeadersEventSource创建连接用的是内建的HTTP GET请求不能设置自定义Header。这在真实业务里几乎是致命的AI对话接口要带Token鉴权通常Token放在Authorization头的Bearer字段里或者放在自定义Header里。EventSource没法做这件事。你可以安慰自己用query参数传Token也能活但Token出现在访问日志里这件事本身就够写一篇安全复盘。而且很多后端框架对URL有长度限制Token加上大段Prompt塞在URL里不是所有网关都扛得住。如果只是内部管理系统把鉴权方式改成Cookie还能圆过去。但对外部客户提供的API或者需要走统一网关鉴权的场景EventSource基本就是用不了。这时候需要用fetch自己实现SSE读取。3.3 限制二无法携带请求体EventSource只有GET一条路意味着请求体里的参数只能通过URL query拼接。SSE接口往往不只是连接一个地址这么简单服务端可能需要客户端传一个任务ID、一组筛选条件、甚至一段Prompt文本。这些参数塞在query里一方面不美观另一方面有长度上限更重要的是遇到特殊字符要做一套转义处理。相比之下用fetch实现SSE时可以配置method: POST把参数放进请求体Content-Type用application/json服务端解析方式和普通POST接口一致后端同事们不用学新东西前端传参也不需要小心翼翼拼URL。3.4 限制三事件类型处理弱原生EventSource支持自定义事件类型addEventListener监听不同的事件名。听起来够用但封装成通用库的时候会暴露问题这种监听模式是点对点的如果页面上多个模块共享同一条SSE连接每个模块各自监听不同类型的事件代码会在连接管理和事件分发之间纠缠不清。更合理的方式是借鉴EventEmitter的发布订阅模式把SSE连接收到的数据统一收集起来做一个事件分发中心。任何模块可以订阅自己关心的事件不需要知道数据具体从哪条连接来。原生EventSource没有这种能力。3.5 限制四自动重连行为不可控前面说过EventSource自带自动重连这是优点。但它重连的行为是浏览器黑盒控制的不能控制重连次数上限不能设置重连前的延迟策略也不能在连续失败N次后停止重连。在真实生产环境里一个接口持续报错的时候浏览器还在背后无休止地发请求排查问题时会看到后台一排刷屏日志很有画面感。用fetch实现SSE之后重连策略就可以完全自己掌控失败多少次后停止重连、指数退避的时间计算、断线前收到的Last-Event-ID要不要带上统统能写成明确的代码逻辑。3.6 小结为什么要自己封装总结一下原生EventSource适合跑通Demo、做原型验证但真实业务场景中至少会遇到鉴权和请求方式两个硬门槛。这就是市面上各种SSE封装库存在的原因也是这篇文章把重点放在封装上的理由。封装的核心目标就三个支持请求方法可配置、Headers可配置支持取消请求和自动重连支持事件订阅分发而非单一onmessage满足这三点一个SSE封装才算真正能落地到项目中。4. 用fetch读取SSE流从响应字节到业务数据的桥梁封装SSE的第一步是用fetch替换EventSource建立连接解决请求方式受限的问题。fetch返回的响应体是一个ReadableStream可读流需要逐块读取数据。这里有两个必须处理的技术点一是读取二进制数据块时要用TextDecoder做解码二是解析SSE协议格式时要处理多行data字段。4.1 建立可取消的SSE流fetch配合AbortController使用就可以在任意时刻主动中断SSE连接这是实现停止生成功能的基础。class SSEConnection { constructor(options) { this.url options.url; this.options options; this.controller new AbortController(); this.reader null; } async connect() { const fetchOptions { method: this.options.method || GET, headers: this.options.headers || {}, signal: this.controller.signal, }; if (this.options.method POST) { fetchOptions.body JSON.stringify(this.options.body || {}); fetchOptions.headers[Content-Type] application/json; } const response await fetch(this.url, fetchOptions); if (!response.ok) { throw new Error(HTTP ${response.status} ${response.statusText}); } this.reader response.body.getReader(); return this; } async readStream(onParse) { const decoder new TextDecoder(utf-8); let buffer ; while (true) { const { done, value } await this.reader.read(); if (done) break; buffer decoder.decode(value, { stream: true }); // 按SSE协议的空行分隔符切分消息 const messages buffer.split(\n\n); // 最后一段可能是不完整的留在buffer里下次拼接 buffer messages.pop(); for (const message of messages) { onParse(message); } } } abort() { this.controller.abort(); if (this.reader) { this.reader.cancel().catch(() {}); } } }注意decoder.decode(value, { stream: true })里这个stream: true参数它的作用是告诉解码器后面还有数据要来。因为一个字符可能被拆成两个字节片段分别在两次read()中返回如果不用流式解码第二次解码时碰到半个字符就会输出乱码中文场景尤其常见。4.2 SSE协议的数据格式解析SSE协议格式并不复杂核心规则是每行格式为字段名: 值空行代表一条消息结束data:字段保存数据内容多个data行会拼接成一条消息用换行符分隔event:字段指定事件类型缺省值为messageid:字段记录消息ID重连时通过HTTP头Last-Event-ID上报retry:字段指定重连间隔毫秒数写一个解析器把原始字符串解析成结构化的SSE事件对象function parseSSEMessage(rawMessage) { const lines rawMessage.split(\n); const result { event: message, data: , id: , retry: null, }; const dataLines []; for (const line of lines) { const colonIndex line.indexOf(:); if (colonIndex -1) continue; const field line.slice(0, colonIndex).trim(); let value line.slice(colonIndex 1); // SSE协议注释行以冒号开头直接忽略 if (field ) continue; // 规范规定“data: ”开头的值如果有前导空格需要去掉一个 if (value.startsWith( )) { value value.slice(1); } switch (field) { case event: result.event value.trim(); break; case data: dataLines.push(value); break; case id: result.id value.trim(); break; case retry: result.retry parseInt(value.trim(), 10); break; default: break; } } result.data dataLines.join(\n); return result; }把这段解析逻辑放在connect之后的readStream回调里就能拿到干净的SSE事件对象了。注意注释行这条规则容易被忽略服务端偶尔会因为调试往流里写注释行如果解析器不识别就会当垃圾数据处理所以解析时要先判断field 的场景。4.3 常见的数据格式约定根据服务端的不同实现data字段里可能是纯文本、JSON字符串、或者是按[DONE]标记结束的特殊格式。AI对话场景里最常见的两种约定OpenAI官方风格每段data是JSON字符串结束标记是data: [DONE]业务自定义风格每段data是JSON字符串带有type字段区分内容增量工具调用结束等状态前端封装应该在解析层就做容错封装把字符串转成JSON后统一抛出给业务层。不要指望服务端的格式固定不变解析层多一层try/catch业务层就少一堆崩溃function parseData(rawMessage) { const sse parseSSEMessage(rawMessage); if (sse.data) { try { sse.parsedData JSON.parse(sse.data); } catch (e) { sse.parsedData sse.data; } } return sse; }这样一来不管服务端返回的是JSON还是纯文本上层接口可以通过parsedData的类型做统一判断封装的通用性就出来了。5. 封装一个通用SSE客户端事件订阅、重连与状态管理封装SSE客户端的核心是设计一个兼具事件中心、连接管理、错误恢复能力的类。我会把代码拆成模块逐一说明最后给出一个可以直接复制使用的完整版本。5.1 事件中心设计发布订阅模式EventSource自带的事件监听方式在复杂业务中不够用原因是连接层和业务层耦合太深。封装时应借鉴EventEmitter的发布订阅模式让连接层只负责把数据从网络上拿下来并解析业务层通过订阅关心的事件业务来自行处理。class EventCenter { constructor() { this.listeners new Map(); } on(eventName, callback) { if (!this.listeners.has(eventName)) { this.listeners.set(eventName, []); } this.listeners.get(eventName).push(callback); } off(eventName, callback) { const callbacks this.listeners.get(eventName); if (!callbacks) return; const index callbacks.indexOf(callback); if (index ! -1) { callbacks.splice(index, 1); } } emit(eventName, data) { const callbacks this.listeners.get(eventName); if (!callbacks) return; for (const callback of callbacks) { callback(data); } } }使用者在业务里只需要client.on(message, handler)不用关心底层是用什么方式连接的、解析逻辑长什么样多个模块也可以共用同一个连接实例各自订阅各自的事件互不干扰。5.2 状态管理连接生命周期SSE连接的状态通常有四态IDLE空闲未启动、CONNECTING连接中、OPEN已连接、CLOSED已关闭。用枚举把状态收敛起来暴露status访问器就能让UI层根据状态展示对应的UI——比如连接中显示加载已关闭显示重连按钮。const SSEStatus { IDLE: idle, CONNECTING: connecting, OPEN: open, CLOSED: closed, };状态切换的同时派发statuschange事件让想监听状态的模块统一走事件中心避免到处设置回调。5.3 重连策略指数退避与主动停止重连的目的是给服务端恢复的时间但如果服务端一直不可用无休止重连就是给服务器雪上加霜。工程上的通用做法是指数退避每次重连失败后等待时间翻倍同时设置一个最大等待时间上限。class ReconnectManager { constructor({ maxRetries 10, baseDelay 1000, maxDelay 30000 } {}) { this.maxRetries maxRetries; this.baseDelay baseDelay; this.maxDelay maxDelay; this.retries 0; } nextDelay() { const delay Math.min(this.baseDelay * Math.pow(2, this.retries), this.maxDelay); // 增加随机抖动避免多个客户端同时重连导致服务端瞬间被打满 const jitter Math.round(Math.random() * 300); this.retries 1; return delay jitter; } reset() { this.retries 0; } canRetry() { return this.retries this.maxRetries; } }抖动jitter这个细节容易被忽视。想象用户量稍大一点大家都在同一时刻断线重连请求同时到达服务端正常情况下服务端会被瞬间击穿。有了随机抖动请求分散到一个时间区间里服务端的压力曲线就平滑很多。重连时要不要带上Last-Event-ID如果服务端支持增量推送应该把上次成功收到的id放在HTTP头里。这样服务端可以从中断的位置继续推而不是重新推一遍全部数据。这个大而全的方案在实际对接中后端通常会自行考虑但前端封装时把Last-Event-ID逻辑预留好对接成本会低很多。5.4 心跳保活HTTP连接在半挂状态时不容易被发现TCP层面的死连接要等系统超时才能感知这期间前端可能一直在等永远不会来的数据。常规做法是设置心跳检测在收到服务端数据时刷新一个时间戳用定时器定期检查——如果超过N秒没收到任何值就主动断开重连。setupHeartbeat(timeout 15000) { this.lastMessageTime Date.now(); this.heartbeatTimer setInterval(() { if (Date.now() - this.lastMessageTime timeout) { this.reconnect(); } }, timeout / 2); } touchHeartbeat() { this.lastMessageTime Date.now(); }心跳的时间参数需要结合具体的服务端推送频率来调整。如果服务端本来就每5秒推一条数据心跳超时设成15秒就不合适可能数据还在正常推前端因为误判已经踢掉重连了。合理的做法是心跳超时时间 预期的最大数据间隔 * 2.5倍左右。5.5 完整的SSEClient类把事件中心、状态管理、重连策略、心跳检测组装到一起就是一个完整的可上生产的SSE封装class SSEClient { constructor({ url, method GET, headers {}, body null, maxRetries 10, baseDelay 1000, maxDelay 30000, heartbeatTimeout 15000 } {}) { this.url url; this.method method; this.headers headers; this.body body; this.heartbeatTimeout heartbeatTimeout; this.eventCenter new EventCenter(); this.reconnectManager new ReconnectManager({ maxRetries, baseDelay, maxDelay }); this.controller null; this.reader null; this.status SSEStatus.IDLE; this.lastEventId ; this.manualClosed false; } on(eventName, callback) { this.eventCenter.on(eventName, callback); return this; } off(eventName, callback) { this.eventCenter.off(eventName, callback); return this; } async start() { if (this.status SSEStatus.OPEN || this.status SSEStatus.CONNECTING) return; this.manualClosed false; await this.connect(); } async connect() { this.setStatus(SSEStatus.CONNECTING); this.controller new AbortController(); try { const fetchOptions { method: this.method, headers: { ...this.headers }, signal: this.controller.signal, }; if (this.lastEventId) { fetchOptions.headers[Last-Event-ID] this.lastEventId; } if (this.method POST this.body) { fetchOptions.headers[Content-Type] application/json; fetchOptions.body JSON.stringify(this.body); } const response await fetch(this.url, fetchOptions); if (!response.ok) { throw new Error(SSE connection failed: ${response.status}); } this.reconnectManager.reset(); this.setStatus(SSEStatus.OPEN); this.eventCenter.emit(open, response); this.reader response.body.getReader(); await this.readStream(); } catch (err) { if (err.name AbortError) { this.setStatus(SSEStatus.CLOSED); this.eventCenter.emit(abort, null); return; } this.eventCenter.emit(error, err); if (!this.manualClosed) { this.handleReconnect(); } else { this.setStatus(SSEStatus.CLOSED); } } } async readStream() { const decoder new TextDecoder(utf-8); let buffer ; this.lastMessageTime Date.now(); this.setupHeartbeat(); while (true) { const { done, value } await this.reader.read(); if (done) break; const text decoder.decode(value, { stream: true }); // 只要收到任何字节就刷新心跳时间 this.touchHeartbeat(); buffer text; const messages buffer.split(\n\n); buffer messages.pop(); for (const message of messages) { const sseMessage parseSSEMessage(message); if (sseMessage.id) { this.lastEventId sseMessage.id; } this.eventCenter.emit(sseMessage.event, sseMessage.parsedData); this.eventCenter.emit(message, sseMessage.parsedData); } } // 流正常结束时做了清理 this.clearHeartbeat(); if (!this.manualClosed) { this.handleReconnect(); } else { this.setStatus(SSEStatus.CLOSED); } } handleReconnect() { if (!this.reconnectManager.canRetry()) { this.setStatus(SSEStatus.CLOSED); this.eventCenter.emit(fatal, this.reconnectManager.retries); return; } this.setStatus(SSEStatus.CONNECTING); const delay this.reconnectManager.nextDelay(); this.eventCenter.emit(retry, delay); this.reconnectTimer setTimeout(() this.connect(), delay); } close() { this.manualClosed true; this.clearHeartbeat(); if (this.controller) { this.controller.abort(); } this.setStatus(SSEStatus.CLOSED); } setStatus(status) { this.status status; this.eventCenter.emit(statuschange, status); } setupHeartbeat() { this.lastMessageTime Date.now(); this.heartbeatTimer setInterval(() { if (Date.now() - this.lastMessageTime this.heartbeatTimeout) { this.eventCenter.emit(heartbeat-timeout, null); this.reconnect(); } }, this.heartbeatTimeout / 2); } touchHeartbeat() { this.lastMessageTime Date.now(); } clearHeartbeat() { if (this.heartbeatTimer) { clearInterval(this.heartbeatTimer); this.heartbeatTimer null; } } get currentStatus() { return this.status; } } export { SSEClient, SSEStatus };这样一套封装下来业务侧的使用方式非常简洁const client new SSEClient({ url: /api/ai/chat, method: POST, headers: { Authorization: Bearer ${token}, }, body: { messages: [{ role: user, content: 你好 }], }, }); client.on(delta, (data) { // 追加渲染 contentRef.current.textContent data.content; }); client.on(done, (data) { // 结束处理 client.close(); }); client.on(error, (err) { console.error(SSE error, err); }); client.on(statuschange, (status) { // 更新UI }); client.start();6. 实战场景AI对话流式渲染与停止生成封装完之后用一个完整的AI对话场景来串联所有能力。这个场景覆盖了SSE封装里几乎全部的核心能力鉴权、POST请求、流式渲染、取消、状态管理。把这套逻辑理解透换成天气推送、日志流、任务进度等场景也只是换汤不换药。6.1 整体交互流程一个AI对话页面的核心流程是这样的用户输入Prompt点击发送页面把用户消息渲染到消息列表创建SSE连接携带鉴权信息和用户消息服务端返回SSE流每个delta事件携带一小段文本增量前端把增量追加到一个累积字符串中实时更新到Message组件收到done事件后关闭连接并做后续清理用户点击停止生成按钮时调用client.close()立即中断流关键在于第5步重复渲染会导致每收到一次delta就新建一个React组件整个消息区域奇卡无比。正确的方案是让SSE的数据独立于React状态流之外用ref持有累积文本每次收到增量更新DOM节点内容或者利用React 18的useSyncExternalStore做受控状态更新避免组件树重挂载。下面是一段采用React ref方案的核心逻辑。6.2 实现一个ChatStream组件function ChatStream() { const [messages, setMessages] useState([]); const [status, setStatus] useState(idle); const streamingRef useRef(null); const abortRef useRef(null); const appendDelta (messageId, deltaText) { setMessages((prev) prev.map((msg) { if (msg.id ! messageId) return msg; return { ...msg, content: msg.content deltaText }; })); }; const sendMessage async (userText) { const messageId msg_${Date.now()}; // 先渲染用户消息和空白的AI回复占位 setMessages((prev) [ ...prev, { id: user_${Date.now()}, role: user, content: userText }, { id: messageId, role: assistant, content: }, ]); abortRef.current new SSEClient({ url: /api/ai/chat, method: POST, headers: { Authorization: Bearer ${localStorage.getItem(token)}, }, body: { message: userText, history: messages, }, }); const client abortRef.current; client.on(delta, (data) { appendDelta(messageId, data.content); }); client.on(tool_call, (data) { // 服务端发来工具调用时展示一个调用卡片 appendDelta(messageId, \n[调用工具: ${data.toolName}]\n); }); client.on(done, () { client.close(); setStatus(done); }); client.on(open, () { setStatus(streaming); }); client.on(error, (err) { console.error(SSE error, err); setStatus(error); }); await client.start(); }; const stopGeneration () { if (abortRef.current) { abortRef.current.abort(); setStatus(stopped); } }; return ( div classNamechatWrapper div classNamemessageList {messages.map((msg) ( div key{msg.id} className{msg.role} {msg.content} /div ))} /div div classNameinputArea input placeholder请输入 onKeyDown{(e) { if (e.key Enter) { const value e.target.value; e.target.value ; sendMessage(value); } }} / {status streaming ? ( button onClick{stopGeneration}停止生成/button ) : null} /div /div ); }这个例子里有个细节值得专门说明为什么要在sendMessage里保存abortRef而不是在组件内用一个普通的局部变量因为停止生成按钮可能是另一个组件触发的或者放在页面的另一块区域这时候需要通过ref把客户端实例提升到组件生命周期级别。6.3 服务端格式约定示例服务端返回的SSE数据格式建议如下这个格式在对接大模型时最通用event: delta data: {content: 你} event: delta data: {content: 好} event: tool_call data: {toolName: search_engine, query: 前端SSE教程} event: delta data: {content: 关于前端SSE封装} event: done data: {suggestions: [...]}服务端每生成一小段Token就触发一个delta事件前端每收到一个delta就追加一段。工具调用单独用tool_call事件好处是前端可以做不同的渲染策略。最终用done事件通知前端流已经完整结束了。6.4 停止生成与中断清理用户点击停止后调用client.close()内部执行的是controller.abort()。此时fetch的Promise会抛出一个AbortError在connect的catch分支里被捕获。注意一个关键细节AbortError应该被视为正常关闭而不是错误需要在catch里单独判断并静默处理否则会走到重连逻辑里去用户点停止结果下一秒又给你重新接上了体验非常糟糕。另一个容易被忽略的点是client.close()之后要清理定时器。如果心跳检测的setInterval还活着即使连接已经关闭定时器还会定期执行检查不断触发超时重连逻辑。封装里的clearHeartbeat就是干这个的。6.5 大模型场景下的渲染性能优化当SSE数据量很大时每来一个Token就触发一次setState这些setState又导致组件重渲染很快会发现页面打字机效果失灵变成一顿一顿输出。用户感知是是不是卡了其实瓶颈在React渲染而不是网络。可以这样优化把消息内容存进ref用requestAnimationFrame控制DOM更新频率。每帧只更新一次DOM节点即使一帧内来了比如50个Token也只合并渲染一次。这种场景下React的setState机制反而成了一个负担因为React由调度器决定何时re-render无法精确控制到每个Token。// 以requestAnimationFrame批量更新为例 const contentRef useRef(null); const pendingText useRef(); client.on(delta, (data) { pendingText.current data.content; if (!rafHandle.current) { rafHandle.current requestAnimationFrame(() { contentRef.current.textContent pendingText.current; pendingText.current ; rafHandle.current null; }); } });如果文本中还包含Markdown格式比如代码块、标题、加粗那还需要考虑流式Markdown渲染的问题。常见方案是引入marked或markdown-it实时解析但在流式输出过程中解析到不完整的Markdown标记会产生闪烁这时候最简单的方案是先累积文本等done事件之后做一次完整渲染过程中的打字机效果用纯文本展示成本低而且体验不差。7. 常用框架与场景速通uniapp、大文件上传、进度展示聊完核心封装和AI对话场景再补充几个常见的延伸场景这些场景实际开发中经常被问到但网上完整答案并不多。7.1 uniapp中使用SSE的注意事项H5端的uniapp可以直接使用浏览器原生EventSource也可以使用fetch封装。但小程序端微信、支付宝就比较麻烦因为小程序环境的网络API和浏览器不完全一致。Taro或者uni-app的小程序端要使用SSE思路是使用小程序的request接口它支持enableChunked能开启流式响应。具体实现思路以uni.request为例// 注意uni.request的enableChunked只在部分小程序平台支持 uni.request({ url: https://api.example.com/stream, method: POST, data: { prompt: 你好 }, enableChunked: true, responseType: arraybuffer, header: { Content-Type: application/json, Authorization: Bearer token, }, success: (res) { // 流式返回时会在onChunkReceived中分块返回 }, });实测下来微信小程序的enableChunked在部分低版本基础库上支持不完整报各种诡异的错。如果要在小程序里稳定做AI流式输出最省心的方式其实是把WebSocket作为传输层服务端把SSE消息转成WebSocket消息推过来虽然重一点但兼容性好得多。uniapp里关于封装H5如何指向两个域名的配置问题本质上是环境变量的域名字段配置在manifest.json里配置h5的devServer代理规则也能处理跨域场景。7.2 大文件上传进度推送SSE也常用于大文件上传场景。前端使用worker上传大文件时worker把文件切片上传到服务器服务端在合并切片时通过SSE推送进度。相比轮询接口查进度SSE推送进度的体验确实好很多进度条是丝滑的不是一秒一蹦的。实现方式是在创建上传任务时服务端返回一个taskId前端拿着taskId建立SSE连接const client new SSEClient({ url: /api/upload/progress?taskId${taskId}, method: GET, headers: { Authorization: Bearer ${token} }, }); client.on(progress, (data) { // data: { percent: 45, speed: 2.3MB/s } progressBar.style.width ${data.percent}%; speedText.textContent data.speed; }); client.on(completed, (data) { fileStatusRef.current done; client.close(); });这个场景还有个选择普通HTTP请求轮询和SSE的真实对比。轮询的实现简单但延迟高而且会产生大量无效请求SSE的实现稍微多花一点后端功夫换来的是毫秒级延迟和更少的请求量。目前比例合理的中大型项目中新做的上传模块不少已经选了SSE方案。7.3 页面大屏数据动态刷新大屏展示页经常有实时数据需求以往的做法是setInterval每分钟轮询一次接口数据的实时性差而且大屏重启时瞬时请求量容易把后端打懵。用SSE之后数据是推送的延迟降到秒级以内。大屏场景和前两个场景有个显著区别数据源通常不只是一个比如一个大屏上可能有CPU使用率、业务订单量、在线人数、告警事件四个模块每个数据源都有自己的推送频率。与其开四条SSE连接不如让服务端把数据聚合到一条连接上根据不同的event类型区分模块const client new SSEClient({ url: /api/dashboard/realtime, method: GET, }); client.on(cpu, (data) updateCpuChart(data)); client.on(orders, (data) updateOrderNum(data)); client.on(online, (data) updateOnlineUsers(data)); client.on(alert, (data) pushAlertToast(data));监听不同的event类型视图层的代码就非常干净新增一个指标时只要在服务端多映射一种event前端加一行client.on监听就够了不影响其他模块。8. 高频问题与排查从idle timeout到连接断开最后把实际开发中高频踩到的问题汇总成一张速查表配合排查思路省去到处搜索的时间。8.1 常见问题速查表问题现象根本原因解决方案连接建立后一直不收到数据直到报错stream disconnected before completion: idle timeout waiting for sse服务端或网关在空闲一段时间后主动断开连接配置心跳保活服务端增加注释行以冒号开头的行定期发送网关层调整idle timeout比如Nginx的proxy_read_timeout无法携带Authorization头原生EventSource只支持GET不能自定义Headers使用本文的fetch封装方案或改用Cookie鉴权中文乱码TextDecoder没有启用流式解码decoder.decode(value, { stream: true })点击停止后仍然重连AbortError被当成错误处理走了重连逻辑在catch分支单独判断err.name AbortError收不到最后一帧数据SSE协议要求每帧不漏发空行结束符部分服务端实现漏写和后端核对响应格式最后一行必须补一个空行页面卸载后连接未关闭报错组件销毁时没有调用closeuseEffect清理函数中调用client.close()同时abort fetch重连风暴打垮服务端大量客户端同时重连重连延迟加随机抖动控制重连上限次数后端返回被压缩乱码网关开启了gzip压缩流式响应被缓存压缩设置响应头 Content-Encoding: identity或在Nginx中关闭gzip对event-stream的压缩8.2 最折磨人的idle timeout问题这个问题在AI对话场景中很典型模型思考时间较长比如用户问了一个复杂问题大模型需要思考十几秒才输出第一个Token这期间SSE连接上没有任何数据。如果服务端或网关的idle timeout设置的比较短比如常见Nginx默认的proxy_read_timeout 60s连接就会在思考期间被断开。前端的表现非常迷惑连接一开始是正常的但过了几十秒后突然报stream disconnected before completion而页面上什么都没有显示。排查思路分三步先确认断开的到底是谁。在Chrome DevTools的Network面板里找到SSE请求查看Network的时间列看看断开的时间点和idle timeout的时间点是否吻合和后端确认响应链路上是否有网关或代理层超时。Nginx的proxy_read_timeout、ALB的idle timeout、Serverless平台的socket timeout任何一个都可能掐断连接双管齐下补救前端开启心跳保活定时发送注释行SSE协议支持:开头的注释行服务端可以定期发空行网关就不会因为空闲而断开连接后端加padding设置比如OpenAI在模型思考时会推送一些事件行保持活跃8.3 如何用控制台快速定位SSE问题Chrome的Network面板对SSE连接的处理已经比较友好但很多人不知道的是点击SSE请求之后可以切到EventStream标签页它能直接看到每条消息的事件名、id和数据内容。这是排查SSE问题最直接的观察窗口。同时有两个数据值得关注Response Headers里的Content-Type必须是text/event-stream否则浏览器可能不按流式处理响应头里不应有Content-LengthSSE是分块的流式响应如果看到了Content-Length说明响应被缓冲了那流式效果就没了如果发现自己发的请求在Network里显示成普通响应而不是EventStream问题通常出在这两个地方直接抓后端确认即可。8.4 生产环境的总结性提醒再强调几个生产环境的坑第一HTTP/2的多路复用特性理论上允许同一域名下开多个SSE连接但浏览器对HTTP/1.1下同域名连接数有限制一般是6个。如果你在一个页面上开了好几条SSE连接注意别撞上连接数上限。实践上尽量合并连接按事件类型区分。第二移动端和弱网环境下SSE的长连接更容易被运营商或系统拦截。除了心跳保活还要做好自动重连的心理预期UI上要有断线提示和重连状态。第三如果是内部系统且允许使用WebSocket短连接场景还是可以用EventSource的但外部客户服务、AI开放平台这类场景用带鉴权的fetch封装是唯一稳的路线。9. 写在最后SSE封装值得用心做一次做前端这些年我越来越觉得会用API和能造工具是两码事。EventSource的API只有那几行但真正要让它服务好业务需要处理鉴权、重连、心跳、事件分发、状态管理这些脏活累活。把这些活儿封装成一个通用SSE客户端就算一劳永逸了——后续AI对话、推送通知、数据大屏、上传进度做起来复用同一套东西不用重新踩坑。我在实际项目中最大的体会是SSE封装的设计比实现更重要尤其是事件中心这层抽象它决定了封装的灵活度。如果你只是把onmessage包一层方法那不叫封装叫函数提取。真正好用的封装业务方完全不需要知道底层用的是SSE还是WebSocket只需要订阅事件、接收数据、完成任务剩下的都应该是透明的东西。最后留一个可用于扩展的方向如果你正在做多租户系统同一个SSE连接可能需要按用户区分数据通道这时可以在封装里增加一个namespace维度实现连接共享和订阅隔离。这块如果感兴趣可以自己试试原理就是在服务器端把不同用户的数据通过同一个连接的不同事件名推送前端的封装层再做一层路由即可。希望这篇文章能帮你把SSE这块从会调接口升级到能应付真实业务下次面试官问起SSE和WebSocket的区别你也可以不只是说一句一个双向一个单向而是把重连、鉴权、心跳、选型这几个维度都梳理得清清楚楚。 SSE流式技术详解与前端封装实战：从EventSource到通用客户端
1. 为什么突然都在聊SSE从AI聊天打字机效果说起如果你这两年做过AI应用的前端或者刷过前端面试题大概率会碰上一个词SSEServer-Sent Events服务器推送事件。市面上铺天盖地的大模型流式输出打字机效果实时渲染回答底层基本都是它。最直观的场景就是你打开Kimi、文心一言、通义千问这类AI对话产品按下回车之后答案不是一整块蹦出来的而是一个字一个字往外蹦。这个体验背后的核心技术就是SSE。服务端通过一个长连接把大模型生成的内容分段推给浏览器前端每收到一小段就立刻追加到页面上用户的感知就是它正在打字。SSE不是新技术早年常用于股票行情推送、服务器日志实时滚动这类场景但真正让它重回聚光灯下的是AI大模型应用的爆发。原因也很简单大模型生成Token本身是流式的服务端边生成边推送客户端边接收边渲染配合起来天衣无缝。这篇文章就从零开始把SSE的来龙去脉讲清楚重点落在封装和实战上。内容分三块先说SSE和WebSocket的区别帮你搞清楚什么场景该选谁再深入原生EventSource API的局限性说明为什么单一原生API扛不住真实业务然后给出一套完整的封装方案包含取消请求、自动重连、事件分发、心跳保活这些真实项目里绕不开的点最后配一个AI对话渲染案例和一份高频问题排查清单。无论你是刚接触前端的新手还是在后台管理系统里被各种数据实时更新需求折磨的资深开发这篇文章的目标都是让你看完能直接动手写出自己的SSE封装而不是停留在用过EventSource的层面。2. SSE和WebSocket一对被经常搞混的兄弟很多同学第一次听到SSE的完整名字Server-Sent Events时第一反应就是这不就是WebSocket吗一开始学的时候我也这么想过。两个都是浏览器和服务器之间的长连接都能实现服务端主动推送但实际用起来差异非常大。2.1 连接方向与通信模式WebSocket是全双工通信客户端和服务端可以同时互相收发消息连接建立后通道是双向的。SSE是单工通信数据只能从服务端往客户端推客户端要发消息得另开一个普通的HTTP请求。全双工和单工的区别在AI对话场景里有很实际的影响。AI对话产品通常是你发送一条Prompt然后服务端持续Push内容数据流方向是单向的SSE天然匹配。但如果是一个在线协作白板用户A画一笔要同步给用户BB的操作又要回传给A这种双向实时交互就必须上WebSocket。用生活化的方式理解SSE像一个单向广播电台服务器是电台主播客户端是收音机主播说话你只能听着你想回话得打电话过去另发HTTP请求WebSocket像电话通话两边都能随时开口。选型的时候先问自己一句我这个功能需要双向实时通信吗不需要就别背WebSocket的重量。2.2 协议与实现成本WebSocket需要专门的通讯协议后端要支持协议升级HTTP Upgrade部署时还得考虑网关、负载均衡器的WebSocket代理配置一不小心就是各种握手失败。SSE就没有这么多弯弯绕它就是标准的HTTP协议在响应头里声明Content-Type: text/event-stream即可。这个差异直接影响了接入成本。我见过不少团队为了一个后台有数据前端推送的功能硬上WebSocket结果后端同事配Nginx配了一整天还得处理连接断开后的状态同步。同样的需求如果用SSE后端只需要在HTTP响应里不断写数据即可不需要处理额外的协议连接生命周期维护成本低一截。2.3 自动重连与断线恢复SSE的原生APIEventSource自带断线重连机制连接意外断开后浏览器会自动重新发起连接服务端还可以通过retry字段指定重连间隔。WebSocket没有这个原生能力断线后要自己实现重连逻辑搭配心跳检测踢掉假死连接。很多人忽略了这一点但这恰恰是SSE在稳定性上最讨喜的地方。做前端的人应该都有过这种经历WebSocket突然静默断线页面上一排灰色消息没人管用户刷新页面才发现全都断了。SSE有浏览器兜底配合Last-Event-ID还能断线续传少写很多代码。2.4 什么时候选SSE什么时候硬上WebSocket给一个实际上能用的选型总结纯推送场景AI流式输出、通知推送、日志流、行情刷新优先SSE双向交互场景IM聊天、协同编辑、白板协作、实时游戏选WebSocket需要兼容老浏览器必须上WebSocketSSE在IE里基本上是不可用的虽然现在还在用IE的已经不多了有一点经常被忽略WebSocket对服务器连接数的压力比SSE大得多SSE走普通HTTP连接配合现有负载均衡体系更平滑记住一个原则能用SSE解决的问题不要没事就上WebSocket这不是技术炫技的问题纯粹是成本问题。3. 原生EventSource能跑AI示例但扛不住真实业务很多AI应用前端的入门教程里SSE的写法就是一个new EventSource(url)加一个onmessage监听看起来非常简单。但你照着搬到真实项目里就会发现远远不够原因在于EventSource API有硬性限制。3.1 原生API的基本形态先看一个标准用法const source new EventSource(/api/chat/stream); // 监听默认消息类型 source.onmessage (event) { console.log(event.data); }; // 监听服务端自定义的事件类型 source.addEventListener(custom-event, (event) { console.log(event.data); }); // 错误处理 source.onerror (err) { console.error(连接异常, err); };服务端只需要这样做Content-Type: text/event-stream Cache-Control: no-cache Connection: keep-alive data: {content: 你好} data: {content: 我} data: {content: 是AI}每两段数据之间用空行分隔data:后面的内容会在前端event.data里拿到。协议本身简洁到有点原始但这种简洁恰恰带来了后面一堆问题。3.2 限制一只能GET请求无法自定义HeadersEventSource创建连接用的是内建的HTTP GET请求不能设置自定义Header。这在真实业务里几乎是致命的AI对话接口要带Token鉴权通常Token放在Authorization头的Bearer字段里或者放在自定义Header里。EventSource没法做这件事。你可以安慰自己用query参数传Token也能活但Token出现在访问日志里这件事本身就够写一篇安全复盘。而且很多后端框架对URL有长度限制Token加上大段Prompt塞在URL里不是所有网关都扛得住。如果只是内部管理系统把鉴权方式改成Cookie还能圆过去。但对外部客户提供的API或者需要走统一网关鉴权的场景EventSource基本就是用不了。这时候需要用fetch自己实现SSE读取。3.3 限制二无法携带请求体EventSource只有GET一条路意味着请求体里的参数只能通过URL query拼接。SSE接口往往不只是连接一个地址这么简单服务端可能需要客户端传一个任务ID、一组筛选条件、甚至一段Prompt文本。这些参数塞在query里一方面不美观另一方面有长度上限更重要的是遇到特殊字符要做一套转义处理。相比之下用fetch实现SSE时可以配置method: POST把参数放进请求体Content-Type用application/json服务端解析方式和普通POST接口一致后端同事们不用学新东西前端传参也不需要小心翼翼拼URL。3.4 限制三事件类型处理弱原生EventSource支持自定义事件类型addEventListener监听不同的事件名。听起来够用但封装成通用库的时候会暴露问题这种监听模式是点对点的如果页面上多个模块共享同一条SSE连接每个模块各自监听不同类型的事件代码会在连接管理和事件分发之间纠缠不清。更合理的方式是借鉴EventEmitter的发布订阅模式把SSE连接收到的数据统一收集起来做一个事件分发中心。任何模块可以订阅自己关心的事件不需要知道数据具体从哪条连接来。原生EventSource没有这种能力。3.5 限制四自动重连行为不可控前面说过EventSource自带自动重连这是优点。但它重连的行为是浏览器黑盒控制的不能控制重连次数上限不能设置重连前的延迟策略也不能在连续失败N次后停止重连。在真实生产环境里一个接口持续报错的时候浏览器还在背后无休止地发请求排查问题时会看到后台一排刷屏日志很有画面感。用fetch实现SSE之后重连策略就可以完全自己掌控失败多少次后停止重连、指数退避的时间计算、断线前收到的Last-Event-ID要不要带上统统能写成明确的代码逻辑。3.6 小结为什么要自己封装总结一下原生EventSource适合跑通Demo、做原型验证但真实业务场景中至少会遇到鉴权和请求方式两个硬门槛。这就是市面上各种SSE封装库存在的原因也是这篇文章把重点放在封装上的理由。封装的核心目标就三个支持请求方法可配置、Headers可配置支持取消请求和自动重连支持事件订阅分发而非单一onmessage满足这三点一个SSE封装才算真正能落地到项目中。4. 用fetch读取SSE流从响应字节到业务数据的桥梁封装SSE的第一步是用fetch替换EventSource建立连接解决请求方式受限的问题。fetch返回的响应体是一个ReadableStream可读流需要逐块读取数据。这里有两个必须处理的技术点一是读取二进制数据块时要用TextDecoder做解码二是解析SSE协议格式时要处理多行data字段。4.1 建立可取消的SSE流fetch配合AbortController使用就可以在任意时刻主动中断SSE连接这是实现停止生成功能的基础。class SSEConnection { constructor(options) { this.url options.url; this.options options; this.controller new AbortController(); this.reader null; } async connect() { const fetchOptions { method: this.options.method || GET, headers: this.options.headers || {}, signal: this.controller.signal, }; if (this.options.method POST) { fetchOptions.body JSON.stringify(this.options.body || {}); fetchOptions.headers[Content-Type] application/json; } const response await fetch(this.url, fetchOptions); if (!response.ok) { throw new Error(HTTP ${response.status} ${response.statusText}); } this.reader response.body.getReader(); return this; } async readStream(onParse) { const decoder new TextDecoder(utf-8); let buffer ; while (true) { const { done, value } await this.reader.read(); if (done) break; buffer decoder.decode(value, { stream: true }); // 按SSE协议的空行分隔符切分消息 const messages buffer.split(\n\n); // 最后一段可能是不完整的留在buffer里下次拼接 buffer messages.pop(); for (const message of messages) { onParse(message); } } } abort() { this.controller.abort(); if (this.reader) { this.reader.cancel().catch(() {}); } } }注意decoder.decode(value, { stream: true })里这个stream: true参数它的作用是告诉解码器后面还有数据要来。因为一个字符可能被拆成两个字节片段分别在两次read()中返回如果不用流式解码第二次解码时碰到半个字符就会输出乱码中文场景尤其常见。4.2 SSE协议的数据格式解析SSE协议格式并不复杂核心规则是每行格式为字段名: 值空行代表一条消息结束data:字段保存数据内容多个data行会拼接成一条消息用换行符分隔event:字段指定事件类型缺省值为messageid:字段记录消息ID重连时通过HTTP头Last-Event-ID上报retry:字段指定重连间隔毫秒数写一个解析器把原始字符串解析成结构化的SSE事件对象function parseSSEMessage(rawMessage) { const lines rawMessage.split(\n); const result { event: message, data: , id: , retry: null, }; const dataLines []; for (const line of lines) { const colonIndex line.indexOf(:); if (colonIndex -1) continue; const field line.slice(0, colonIndex).trim(); let value line.slice(colonIndex 1); // SSE协议注释行以冒号开头直接忽略 if (field ) continue; // 规范规定“data: ”开头的值如果有前导空格需要去掉一个 if (value.startsWith( )) { value value.slice(1); } switch (field) { case event: result.event value.trim(); break; case data: dataLines.push(value); break; case id: result.id value.trim(); break; case retry: result.retry parseInt(value.trim(), 10); break; default: break; } } result.data dataLines.join(\n); return result; }把这段解析逻辑放在connect之后的readStream回调里就能拿到干净的SSE事件对象了。注意注释行这条规则容易被忽略服务端偶尔会因为调试往流里写注释行如果解析器不识别就会当垃圾数据处理所以解析时要先判断field 的场景。4.3 常见的数据格式约定根据服务端的不同实现data字段里可能是纯文本、JSON字符串、或者是按[DONE]标记结束的特殊格式。AI对话场景里最常见的两种约定OpenAI官方风格每段data是JSON字符串结束标记是data: [DONE]业务自定义风格每段data是JSON字符串带有type字段区分内容增量工具调用结束等状态前端封装应该在解析层就做容错封装把字符串转成JSON后统一抛出给业务层。不要指望服务端的格式固定不变解析层多一层try/catch业务层就少一堆崩溃function parseData(rawMessage) { const sse parseSSEMessage(rawMessage); if (sse.data) { try { sse.parsedData JSON.parse(sse.data); } catch (e) { sse.parsedData sse.data; } } return sse; }这样一来不管服务端返回的是JSON还是纯文本上层接口可以通过parsedData的类型做统一判断封装的通用性就出来了。5. 封装一个通用SSE客户端事件订阅、重连与状态管理封装SSE客户端的核心是设计一个兼具事件中心、连接管理、错误恢复能力的类。我会把代码拆成模块逐一说明最后给出一个可以直接复制使用的完整版本。5.1 事件中心设计发布订阅模式EventSource自带的事件监听方式在复杂业务中不够用原因是连接层和业务层耦合太深。封装时应借鉴EventEmitter的发布订阅模式让连接层只负责把数据从网络上拿下来并解析业务层通过订阅关心的事件业务来自行处理。class EventCenter { constructor() { this.listeners new Map(); } on(eventName, callback) { if (!this.listeners.has(eventName)) { this.listeners.set(eventName, []); } this.listeners.get(eventName).push(callback); } off(eventName, callback) { const callbacks this.listeners.get(eventName); if (!callbacks) return; const index callbacks.indexOf(callback); if (index ! -1) { callbacks.splice(index, 1); } } emit(eventName, data) { const callbacks this.listeners.get(eventName); if (!callbacks) return; for (const callback of callbacks) { callback(data); } } }使用者在业务里只需要client.on(message, handler)不用关心底层是用什么方式连接的、解析逻辑长什么样多个模块也可以共用同一个连接实例各自订阅各自的事件互不干扰。5.2 状态管理连接生命周期SSE连接的状态通常有四态IDLE空闲未启动、CONNECTING连接中、OPEN已连接、CLOSED已关闭。用枚举把状态收敛起来暴露status访问器就能让UI层根据状态展示对应的UI——比如连接中显示加载已关闭显示重连按钮。const SSEStatus { IDLE: idle, CONNECTING: connecting, OPEN: open, CLOSED: closed, };状态切换的同时派发statuschange事件让想监听状态的模块统一走事件中心避免到处设置回调。5.3 重连策略指数退避与主动停止重连的目的是给服务端恢复的时间但如果服务端一直不可用无休止重连就是给服务器雪上加霜。工程上的通用做法是指数退避每次重连失败后等待时间翻倍同时设置一个最大等待时间上限。class ReconnectManager { constructor({ maxRetries 10, baseDelay 1000, maxDelay 30000 } {}) { this.maxRetries maxRetries; this.baseDelay baseDelay; this.maxDelay maxDelay; this.retries 0; } nextDelay() { const delay Math.min(this.baseDelay * Math.pow(2, this.retries), this.maxDelay); // 增加随机抖动避免多个客户端同时重连导致服务端瞬间被打满 const jitter Math.round(Math.random() * 300); this.retries 1; return delay jitter; } reset() { this.retries 0; } canRetry() { return this.retries this.maxRetries; } }抖动jitter这个细节容易被忽视。想象用户量稍大一点大家都在同一时刻断线重连请求同时到达服务端正常情况下服务端会被瞬间击穿。有了随机抖动请求分散到一个时间区间里服务端的压力曲线就平滑很多。重连时要不要带上Last-Event-ID如果服务端支持增量推送应该把上次成功收到的id放在HTTP头里。这样服务端可以从中断的位置继续推而不是重新推一遍全部数据。这个大而全的方案在实际对接中后端通常会自行考虑但前端封装时把Last-Event-ID逻辑预留好对接成本会低很多。5.4 心跳保活HTTP连接在半挂状态时不容易被发现TCP层面的死连接要等系统超时才能感知这期间前端可能一直在等永远不会来的数据。常规做法是设置心跳检测在收到服务端数据时刷新一个时间戳用定时器定期检查——如果超过N秒没收到任何值就主动断开重连。setupHeartbeat(timeout 15000) { this.lastMessageTime Date.now(); this.heartbeatTimer setInterval(() { if (Date.now() - this.lastMessageTime timeout) { this.reconnect(); } }, timeout / 2); } touchHeartbeat() { this.lastMessageTime Date.now(); }心跳的时间参数需要结合具体的服务端推送频率来调整。如果服务端本来就每5秒推一条数据心跳超时设成15秒就不合适可能数据还在正常推前端因为误判已经踢掉重连了。合理的做法是心跳超时时间 预期的最大数据间隔 * 2.5倍左右。5.5 完整的SSEClient类把事件中心、状态管理、重连策略、心跳检测组装到一起就是一个完整的可上生产的SSE封装class SSEClient { constructor({ url, method GET, headers {}, body null, maxRetries 10, baseDelay 1000, maxDelay 30000, heartbeatTimeout 15000 } {}) { this.url url; this.method method; this.headers headers; this.body body; this.heartbeatTimeout heartbeatTimeout; this.eventCenter new EventCenter(); this.reconnectManager new ReconnectManager({ maxRetries, baseDelay, maxDelay }); this.controller null; this.reader null; this.status SSEStatus.IDLE; this.lastEventId ; this.manualClosed false; } on(eventName, callback) { this.eventCenter.on(eventName, callback); return this; } off(eventName, callback) { this.eventCenter.off(eventName, callback); return this; } async start() { if (this.status SSEStatus.OPEN || this.status SSEStatus.CONNECTING) return; this.manualClosed false; await this.connect(); } async connect() { this.setStatus(SSEStatus.CONNECTING); this.controller new AbortController(); try { const fetchOptions { method: this.method, headers: { ...this.headers }, signal: this.controller.signal, }; if (this.lastEventId) { fetchOptions.headers[Last-Event-ID] this.lastEventId; } if (this.method POST this.body) { fetchOptions.headers[Content-Type] application/json; fetchOptions.body JSON.stringify(this.body); } const response await fetch(this.url, fetchOptions); if (!response.ok) { throw new Error(SSE connection failed: ${response.status}); } this.reconnectManager.reset(); this.setStatus(SSEStatus.OPEN); this.eventCenter.emit(open, response); this.reader response.body.getReader(); await this.readStream(); } catch (err) { if (err.name AbortError) { this.setStatus(SSEStatus.CLOSED); this.eventCenter.emit(abort, null); return; } this.eventCenter.emit(error, err); if (!this.manualClosed) { this.handleReconnect(); } else { this.setStatus(SSEStatus.CLOSED); } } } async readStream() { const decoder new TextDecoder(utf-8); let buffer ; this.lastMessageTime Date.now(); this.setupHeartbeat(); while (true) { const { done, value } await this.reader.read(); if (done) break; const text decoder.decode(value, { stream: true }); // 只要收到任何字节就刷新心跳时间 this.touchHeartbeat(); buffer text; const messages buffer.split(\n\n); buffer messages.pop(); for (const message of messages) { const sseMessage parseSSEMessage(message); if (sseMessage.id) { this.lastEventId sseMessage.id; } this.eventCenter.emit(sseMessage.event, sseMessage.parsedData); this.eventCenter.emit(message, sseMessage.parsedData); } } // 流正常结束时做了清理 this.clearHeartbeat(); if (!this.manualClosed) { this.handleReconnect(); } else { this.setStatus(SSEStatus.CLOSED); } } handleReconnect() { if (!this.reconnectManager.canRetry()) { this.setStatus(SSEStatus.CLOSED); this.eventCenter.emit(fatal, this.reconnectManager.retries); return; } this.setStatus(SSEStatus.CONNECTING); const delay this.reconnectManager.nextDelay(); this.eventCenter.emit(retry, delay); this.reconnectTimer setTimeout(() this.connect(), delay); } close() { this.manualClosed true; this.clearHeartbeat(); if (this.controller) { this.controller.abort(); } this.setStatus(SSEStatus.CLOSED); } setStatus(status) { this.status status; this.eventCenter.emit(statuschange, status); } setupHeartbeat() { this.lastMessageTime Date.now(); this.heartbeatTimer setInterval(() { if (Date.now() - this.lastMessageTime this.heartbeatTimeout) { this.eventCenter.emit(heartbeat-timeout, null); this.reconnect(); } }, this.heartbeatTimeout / 2); } touchHeartbeat() { this.lastMessageTime Date.now(); } clearHeartbeat() { if (this.heartbeatTimer) { clearInterval(this.heartbeatTimer); this.heartbeatTimer null; } } get currentStatus() { return this.status; } } export { SSEClient, SSEStatus };这样一套封装下来业务侧的使用方式非常简洁const client new SSEClient({ url: /api/ai/chat, method: POST, headers: { Authorization: Bearer ${token}, }, body: { messages: [{ role: user, content: 你好 }], }, }); client.on(delta, (data) { // 追加渲染 contentRef.current.textContent data.content; }); client.on(done, (data) { // 结束处理 client.close(); }); client.on(error, (err) { console.error(SSE error, err); }); client.on(statuschange, (status) { // 更新UI }); client.start();6. 实战场景AI对话流式渲染与停止生成封装完之后用一个完整的AI对话场景来串联所有能力。这个场景覆盖了SSE封装里几乎全部的核心能力鉴权、POST请求、流式渲染、取消、状态管理。把这套逻辑理解透换成天气推送、日志流、任务进度等场景也只是换汤不换药。6.1 整体交互流程一个AI对话页面的核心流程是这样的用户输入Prompt点击发送页面把用户消息渲染到消息列表创建SSE连接携带鉴权信息和用户消息服务端返回SSE流每个delta事件携带一小段文本增量前端把增量追加到一个累积字符串中实时更新到Message组件收到done事件后关闭连接并做后续清理用户点击停止生成按钮时调用client.close()立即中断流关键在于第5步重复渲染会导致每收到一次delta就新建一个React组件整个消息区域奇卡无比。正确的方案是让SSE的数据独立于React状态流之外用ref持有累积文本每次收到增量更新DOM节点内容或者利用React 18的useSyncExternalStore做受控状态更新避免组件树重挂载。下面是一段采用React ref方案的核心逻辑。6.2 实现一个ChatStream组件function ChatStream() { const [messages, setMessages] useState([]); const [status, setStatus] useState(idle); const streamingRef useRef(null); const abortRef useRef(null); const appendDelta (messageId, deltaText) { setMessages((prev) prev.map((msg) { if (msg.id ! messageId) return msg; return { ...msg, content: msg.content deltaText }; })); }; const sendMessage async (userText) { const messageId msg_${Date.now()}; // 先渲染用户消息和空白的AI回复占位 setMessages((prev) [ ...prev, { id: user_${Date.now()}, role: user, content: userText }, { id: messageId, role: assistant, content: }, ]); abortRef.current new SSEClient({ url: /api/ai/chat, method: POST, headers: { Authorization: Bearer ${localStorage.getItem(token)}, }, body: { message: userText, history: messages, }, }); const client abortRef.current; client.on(delta, (data) { appendDelta(messageId, data.content); }); client.on(tool_call, (data) { // 服务端发来工具调用时展示一个调用卡片 appendDelta(messageId, \n[调用工具: ${data.toolName}]\n); }); client.on(done, () { client.close(); setStatus(done); }); client.on(open, () { setStatus(streaming); }); client.on(error, (err) { console.error(SSE error, err); setStatus(error); }); await client.start(); }; const stopGeneration () { if (abortRef.current) { abortRef.current.abort(); setStatus(stopped); } }; return ( div classNamechatWrapper div classNamemessageList {messages.map((msg) ( div key{msg.id} className{msg.role} {msg.content} /div ))} /div div classNameinputArea input placeholder请输入 onKeyDown{(e) { if (e.key Enter) { const value e.target.value; e.target.value ; sendMessage(value); } }} / {status streaming ? ( button onClick{stopGeneration}停止生成/button ) : null} /div /div ); }这个例子里有个细节值得专门说明为什么要在sendMessage里保存abortRef而不是在组件内用一个普通的局部变量因为停止生成按钮可能是另一个组件触发的或者放在页面的另一块区域这时候需要通过ref把客户端实例提升到组件生命周期级别。6.3 服务端格式约定示例服务端返回的SSE数据格式建议如下这个格式在对接大模型时最通用event: delta data: {content: 你} event: delta data: {content: 好} event: tool_call data: {toolName: search_engine, query: 前端SSE教程} event: delta data: {content: 关于前端SSE封装} event: done data: {suggestions: [...]}服务端每生成一小段Token就触发一个delta事件前端每收到一个delta就追加一段。工具调用单独用tool_call事件好处是前端可以做不同的渲染策略。最终用done事件通知前端流已经完整结束了。6.4 停止生成与中断清理用户点击停止后调用client.close()内部执行的是controller.abort()。此时fetch的Promise会抛出一个AbortError在connect的catch分支里被捕获。注意一个关键细节AbortError应该被视为正常关闭而不是错误需要在catch里单独判断并静默处理否则会走到重连逻辑里去用户点停止结果下一秒又给你重新接上了体验非常糟糕。另一个容易被忽略的点是client.close()之后要清理定时器。如果心跳检测的setInterval还活着即使连接已经关闭定时器还会定期执行检查不断触发超时重连逻辑。封装里的clearHeartbeat就是干这个的。6.5 大模型场景下的渲染性能优化当SSE数据量很大时每来一个Token就触发一次setState这些setState又导致组件重渲染很快会发现页面打字机效果失灵变成一顿一顿输出。用户感知是是不是卡了其实瓶颈在React渲染而不是网络。可以这样优化把消息内容存进ref用requestAnimationFrame控制DOM更新频率。每帧只更新一次DOM节点即使一帧内来了比如50个Token也只合并渲染一次。这种场景下React的setState机制反而成了一个负担因为React由调度器决定何时re-render无法精确控制到每个Token。// 以requestAnimationFrame批量更新为例 const contentRef useRef(null); const pendingText useRef(); client.on(delta, (data) { pendingText.current data.content; if (!rafHandle.current) { rafHandle.current requestAnimationFrame(() { contentRef.current.textContent pendingText.current; pendingText.current ; rafHandle.current null; }); } });如果文本中还包含Markdown格式比如代码块、标题、加粗那还需要考虑流式Markdown渲染的问题。常见方案是引入marked或markdown-it实时解析但在流式输出过程中解析到不完整的Markdown标记会产生闪烁这时候最简单的方案是先累积文本等done事件之后做一次完整渲染过程中的打字机效果用纯文本展示成本低而且体验不差。7. 常用框架与场景速通uniapp、大文件上传、进度展示聊完核心封装和AI对话场景再补充几个常见的延伸场景这些场景实际开发中经常被问到但网上完整答案并不多。7.1 uniapp中使用SSE的注意事项H5端的uniapp可以直接使用浏览器原生EventSource也可以使用fetch封装。但小程序端微信、支付宝就比较麻烦因为小程序环境的网络API和浏览器不完全一致。Taro或者uni-app的小程序端要使用SSE思路是使用小程序的request接口它支持enableChunked能开启流式响应。具体实现思路以uni.request为例// 注意uni.request的enableChunked只在部分小程序平台支持 uni.request({ url: https://api.example.com/stream, method: POST, data: { prompt: 你好 }, enableChunked: true, responseType: arraybuffer, header: { Content-Type: application/json, Authorization: Bearer token, }, success: (res) { // 流式返回时会在onChunkReceived中分块返回 }, });实测下来微信小程序的enableChunked在部分低版本基础库上支持不完整报各种诡异的错。如果要在小程序里稳定做AI流式输出最省心的方式其实是把WebSocket作为传输层服务端把SSE消息转成WebSocket消息推过来虽然重一点但兼容性好得多。uniapp里关于封装H5如何指向两个域名的配置问题本质上是环境变量的域名字段配置在manifest.json里配置h5的devServer代理规则也能处理跨域场景。7.2 大文件上传进度推送SSE也常用于大文件上传场景。前端使用worker上传大文件时worker把文件切片上传到服务器服务端在合并切片时通过SSE推送进度。相比轮询接口查进度SSE推送进度的体验确实好很多进度条是丝滑的不是一秒一蹦的。实现方式是在创建上传任务时服务端返回一个taskId前端拿着taskId建立SSE连接const client new SSEClient({ url: /api/upload/progress?taskId${taskId}, method: GET, headers: { Authorization: Bearer ${token} }, }); client.on(progress, (data) { // data: { percent: 45, speed: 2.3MB/s } progressBar.style.width ${data.percent}%; speedText.textContent data.speed; }); client.on(completed, (data) { fileStatusRef.current done; client.close(); });这个场景还有个选择普通HTTP请求轮询和SSE的真实对比。轮询的实现简单但延迟高而且会产生大量无效请求SSE的实现稍微多花一点后端功夫换来的是毫秒级延迟和更少的请求量。目前比例合理的中大型项目中新做的上传模块不少已经选了SSE方案。7.3 页面大屏数据动态刷新大屏展示页经常有实时数据需求以往的做法是setInterval每分钟轮询一次接口数据的实时性差而且大屏重启时瞬时请求量容易把后端打懵。用SSE之后数据是推送的延迟降到秒级以内。大屏场景和前两个场景有个显著区别数据源通常不只是一个比如一个大屏上可能有CPU使用率、业务订单量、在线人数、告警事件四个模块每个数据源都有自己的推送频率。与其开四条SSE连接不如让服务端把数据聚合到一条连接上根据不同的event类型区分模块const client new SSEClient({ url: /api/dashboard/realtime, method: GET, }); client.on(cpu, (data) updateCpuChart(data)); client.on(orders, (data) updateOrderNum(data)); client.on(online, (data) updateOnlineUsers(data)); client.on(alert, (data) pushAlertToast(data));监听不同的event类型视图层的代码就非常干净新增一个指标时只要在服务端多映射一种event前端加一行client.on监听就够了不影响其他模块。8. 高频问题与排查从idle timeout到连接断开最后把实际开发中高频踩到的问题汇总成一张速查表配合排查思路省去到处搜索的时间。8.1 常见问题速查表问题现象根本原因解决方案连接建立后一直不收到数据直到报错stream disconnected before completion: idle timeout waiting for sse服务端或网关在空闲一段时间后主动断开连接配置心跳保活服务端增加注释行以冒号开头的行定期发送网关层调整idle timeout比如Nginx的proxy_read_timeout无法携带Authorization头原生EventSource只支持GET不能自定义Headers使用本文的fetch封装方案或改用Cookie鉴权中文乱码TextDecoder没有启用流式解码decoder.decode(value, { stream: true })点击停止后仍然重连AbortError被当成错误处理走了重连逻辑在catch分支单独判断err.name AbortError收不到最后一帧数据SSE协议要求每帧不漏发空行结束符部分服务端实现漏写和后端核对响应格式最后一行必须补一个空行页面卸载后连接未关闭报错组件销毁时没有调用closeuseEffect清理函数中调用client.close()同时abort fetch重连风暴打垮服务端大量客户端同时重连重连延迟加随机抖动控制重连上限次数后端返回被压缩乱码网关开启了gzip压缩流式响应被缓存压缩设置响应头 Content-Encoding: identity或在Nginx中关闭gzip对event-stream的压缩8.2 最折磨人的idle timeout问题这个问题在AI对话场景中很典型模型思考时间较长比如用户问了一个复杂问题大模型需要思考十几秒才输出第一个Token这期间SSE连接上没有任何数据。如果服务端或网关的idle timeout设置的比较短比如常见Nginx默认的proxy_read_timeout 60s连接就会在思考期间被断开。前端的表现非常迷惑连接一开始是正常的但过了几十秒后突然报stream disconnected before completion而页面上什么都没有显示。排查思路分三步先确认断开的到底是谁。在Chrome DevTools的Network面板里找到SSE请求查看Network的时间列看看断开的时间点和idle timeout的时间点是否吻合和后端确认响应链路上是否有网关或代理层超时。Nginx的proxy_read_timeout、ALB的idle timeout、Serverless平台的socket timeout任何一个都可能掐断连接双管齐下补救前端开启心跳保活定时发送注释行SSE协议支持:开头的注释行服务端可以定期发空行网关就不会因为空闲而断开连接后端加padding设置比如OpenAI在模型思考时会推送一些事件行保持活跃8.3 如何用控制台快速定位SSE问题Chrome的Network面板对SSE连接的处理已经比较友好但很多人不知道的是点击SSE请求之后可以切到EventStream标签页它能直接看到每条消息的事件名、id和数据内容。这是排查SSE问题最直接的观察窗口。同时有两个数据值得关注Response Headers里的Content-Type必须是text/event-stream否则浏览器可能不按流式处理响应头里不应有Content-LengthSSE是分块的流式响应如果看到了Content-Length说明响应被缓冲了那流式效果就没了如果发现自己发的请求在Network里显示成普通响应而不是EventStream问题通常出在这两个地方直接抓后端确认即可。8.4 生产环境的总结性提醒再强调几个生产环境的坑第一HTTP/2的多路复用特性理论上允许同一域名下开多个SSE连接但浏览器对HTTP/1.1下同域名连接数有限制一般是6个。如果你在一个页面上开了好几条SSE连接注意别撞上连接数上限。实践上尽量合并连接按事件类型区分。第二移动端和弱网环境下SSE的长连接更容易被运营商或系统拦截。除了心跳保活还要做好自动重连的心理预期UI上要有断线提示和重连状态。第三如果是内部系统且允许使用WebSocket短连接场景还是可以用EventSource的但外部客户服务、AI开放平台这类场景用带鉴权的fetch封装是唯一稳的路线。9. 写在最后SSE封装值得用心做一次做前端这些年我越来越觉得会用API和能造工具是两码事。EventSource的API只有那几行但真正要让它服务好业务需要处理鉴权、重连、心跳、事件分发、状态管理这些脏活累活。把这些活儿封装成一个通用SSE客户端就算一劳永逸了——后续AI对话、推送通知、数据大屏、上传进度做起来复用同一套东西不用重新踩坑。我在实际项目中最大的体会是SSE封装的设计比实现更重要尤其是事件中心这层抽象它决定了封装的灵活度。如果你只是把onmessage包一层方法那不叫封装叫函数提取。真正好用的封装业务方完全不需要知道底层用的是SSE还是WebSocket只需要订阅事件、接收数据、完成任务剩下的都应该是透明的东西。最后留一个可用于扩展的方向如果你正在做多租户系统同一个SSE连接可能需要按用户区分数据通道这时可以在封装里增加一个namespace维度实现连接共享和订阅隔离。这块如果感兴趣可以自己试试原理就是在服务器端把不同用户的数据通过同一个连接的不同事件名推送前端的封装层再做一层路由即可。希望这篇文章能帮你把SSE这块从会调接口升级到能应付真实业务下次面试官问起SSE和WebSocket的区别你也可以不只是说一句一个双向一个单向而是把重连、鉴权、心跳、选型这几个维度都梳理得清清楚楚。