前段时间把一个 AI 问答模块从轮询改成 SSE 流式输出顺手把断线丢内容、打字机渲染卡顿、半截 Markdown 标签渲染出错这几个问题一起解决了。这套方案在真实业务里跑了快一个月稳定性和体验都比之前好太多。今天我把整个落地过程、核心代码和踩过的坑完整梳理出来包含 SSE 协议细节、断点续传思路、打字机渲染实现以及“idle timeout waiting for SSE”这类典型报错的排查方法。不管你是刚接触流式输出的前端新人还是准备把已有聊天框升级成流式体验的人这篇都能直接拿来用。1. 整体设计与思路拆解1.1 为什么是 SSE而不是 WebSocket 或轮询很多同学一看到“服务器推送”就条件反射想到 WebSocket但 AI 对话这种场景SSE 才是更合适的方案。用一个直观的对比就能看懂方案连接方向数据格式断线重连实现成本轮询请求-响应任意不需要但要自己处理请求频率最低WebSocket全双工二进制/文本需要自己实现较高SSE单向服务端推送纯文本EventSource 内置较低SSE 是 HTTP 协议上的单向通道服务端可以持续往客户端推数据客户端不需要反复发请求。AI 对话场景本质上就是“用户发一句、服务端持续推送回复”完全符合 SSE 的单向特性。WebSocket 不是不能用但它需要升级握手协议、处理心跳、处理二进制帧还要专门维护连接状态。如果业务里没有“服务端主动发消息给用户”的实时交互需求用 WebSocket 属于大材小用还平白增加复杂度。轮询的问题更明显。AI 大模型生成一段完整回复可能耗时十几秒甚至几十秒轮询要么固定间隔导致响应慢要么频繁请求导致服务端压力大。早期版本我就是用轮询结果用户看到的是“转圈很久然后一次性冒出一大段文字”交互体验很生硬。选用 SSE 之后用户能看着文字像真人打字一样逐字出现这个体验差异是决定性的。而且 SSE 走的就是普通 HTTP 协议不需要额外端口能被各种网关和代理正常识别。1.2 断点续传解决的真实问题这里的“断点续传”不是下载文件那种断点续传而是指AI 回复流式输出到一半网络连接突然断了怎么办最朴素的做法是让用户重新点发送但这样会导致大模型重新生成一遍内容成本和等待时间翻倍之前已经显示出来的文字和新生成的内容无法衔接用户的对话上下文被打断体验极差。所以断点续传要做到的是连接断开后自动重连并只获取中间缺失的那部分内容而不是重新生成整段回复。原理上SSE 规范本身就提供了Last-Event-ID机制。服务端每推送一条数据时带上递增的id客户端断线后重连时把Last-Event-ID带回服务端服务端从这个位置继续推送。EventSource API 内置了这个能力但如果你用 fetch 自己封装 SSE 客户端就需要手动实现。实际业务里服务端还得缓存当前这次回复已经生成的内容。因为大模型本身是无状态的断线后如果服务端没存下之前生成的内容重连后也拿不回来。最简单的方案是内存缓存数据量大一点就上 Redis。这块我会在第 3 章给出完整实现。1.3 打字机渲染的前端本质打字机效果看起来炫但原理极其简单前端拿到流式数据后不要一次性渲染到页面而是把文字放进一个缓冲区再用定时器或动画帧分批拿出来追加到 DOM。真正麻烦的是两个问题第一渲染频率和数据到达频率不一致。网络数据可能一瞬间到达几百个数据块如果每个数据块到达都立刻更新 DOM页面可能卡顿但如果用固定setInterval从缓冲区头重新读又容易重复渲染。所以要采用“消费指针”的思路记录已经渲染到哪个位置每次只消费增量。第二流式 Markdown 渲染的问题。如果用户的消息带代码块、列表、表格你在流式中间阶段用 Markdown 解析器去渲染会碰到代码块还没闭合、表格只渲染了一半这类问题。常见的做法是等待完整输出后再统一渲染但这会和打字机效果冲突。目前业内比较成熟的折中方案是逐字显示纯文本流结束后再渲染 Markdown或者只渲染已经完整闭合的块未闭合的部分先用纯文本展示。2. 核心细节解析与实操要点2.1 SSE 协议必须吃透的字段SSE 的数据格式并不复杂但有几个关键点容易踩坑。先看标准协议格式event: message id: 1 data: 第一段内容 event: message id: 2 data: 第二段内容每条事件之间用空行分隔事件由若干个字段组成。常见字段如下字段作用说明data:数据内容意思就是纯文本但支持多行多个 data 行会被解析时用换行符拼接event:事件类型默认是message客户端可以自定义监听事件id:事件序号用于断线续传客户端会自动记录最后一次的 idretry:重连时间毫秒告诉客户端断线后隔多久重连:注释行保活以冒号开头的行是注释客户端会忽略但能维持连接不超时实际传输时data字段往往是 JSON 字符串。需要注意的是如果 JSON 里有换行符必须转义成\n因为 SSE 协议规定data行本身不能包含真实换行。用JSON.stringify序列化时它会自动帮你转义所以不能直接把对象拼到data:后面。还有一个隐藏的坑响应头的 Content-Type 必须是text/event-stream且建议显式加上Cache-Control: no-cache否则部分浏览器或代理层会把流式响应缓存下来导致数据一次性到达或者乱序。连接关闭时服务端要主动断开。2.2 EventSource 和 fetch 怎么选前端接收 SSE 有两种方式EventSource是浏览器原生 API优点是真省事自动重连、自动记录Last-Event-ID代码只有几行fetch需要自己解析响应流但灵活性高能自定义请求头也支持 POST 请求。有一个实际问题很多 AI 平台的接口鉴权要求带Authorization头同时请求内容较长需要 POST 提交。EventSource只支持 GET 请求也无法自定义请求头所以实际项目里八成以上都要用fetch自己封装。能力EventSourcefetch 流式封装GET 请求支持支持POST 请求不支持支持自定义请求头不支持支持自动重连内置默认约 3 秒需自己实现Last-Event-ID 自动回传内置需自己实现自定义事件类型支持需自己解析我用 fetch 封装时的思路是用response.body.getReader()读取底层字节流再用TextDecoder解码成字符串然后按 SSE 帧格式切分数据。解码时特别要注意中文字符可能被拆到两个 chunk 里所以解码器要设置{ stream: true }并保留上一次剩余的字节避免中文乱码。2.3 打字机渲染的进阶处理基础版打字机是“每到一个数据块就把整块追加上去”这只能叫“流式加载”不叫打字机。真正的打字机效果要控制字符出现速度。推荐的做法是维护两个变量contentBuffer已经接收到的完整文本缓冲displayIndex已经渲染到缓冲区中的位置。每次收到新数据只在contentBuffer末尾追加然后通过一个定时器每次从displayIndex开始取 1 到 3 个字符插入 DOM渲染后更新displayIndex。这样做的好处是无论网络多快、数据块多碎DOM 更新节奏始终是平滑的而且断线重连后只要contentBuffer还在displayIndex就不会混乱新到达的内容接着旧的继续显示。但如果消息非常长定时器每 20 毫秒操作一次 DOM 依然有压力。建议把“修改 DOM”和“安排下一次渲染”拆开用requestAnimationFrame代替setInterval。每帧最多追加几个字符帧率不够时自动降频用户的视觉感受反而是最舒服的。如果页面里同时有多个打字机效果需要各自维护独立的displayIndex。3. 实操过程与核心环节实现3.1 后端 Node 实现支持续传的 SSE 接口我用 Node.js 的 Express 写一个最小可用版本只保留关键逻辑设置响应头、按条推送数据、发心跳保持连接、缓存已生成内容以支持断线续传。// sse-server.js const express require(express); const app express(); // 模拟生成内容实际业务中替换为大模型流式返回 function generateContent(sentence) { const chars sentence.split(); let index 0; return (cb) { if (index chars.length) { cb(null, chars.slice(index, index 3).join()); index 3; } else { cb(new Error(END)); } }; } const sessionCache new Map(); // 按连接 id 缓存已经生成的全文 app.get(/api/chat/stream, (req, res) { const { sessionId default, lastEventId } req.query; // 设置 SSE 必要响应头 res.setHeader(Content-Type, text/event-stream; charsetutf-8); res.setHeader(Cache-Control, no-cache); res.setHeader(Connection, keep-alive); res.setHeader(X-Accel-Buffering, no); // 防止 Nginx 缓冲 // 心跳保活每 15 秒发一行注释 const heartbeat setInterval(() { res.write(: heartbeat\n\n); }, 15000); const session sessionCache.get(sessionId) || { count: 0, content: }; sessionCache.set(sessionId, session); // 如果带了 Last-Event-ID说明是断线重连 const startId lastEventId ? Number(lastEventId) : 0; const gen generateContent(这是一段用于模拟 AI 流式输出的内容足够长可以观察到逐字效果。); let restCount Math.floor(startId / 3); // 模拟已经生成过的块数 const push () { // 先跳过已经生成过的部分 for (let i 0; i restCount; i) { try { gen((_, text) { session.content text; }); } catch (e) {} } restCount 0; gen((err, text) { if (err err.message END) { clearInterval(heartbeat); res.end(); return; } session.count 1; session.content text; // data 里用 JSON 传输避免换行符问题 const payload { content: text, sessionId, done: false }; // 每个事件都带自增 id用于断点续传 res.write(id: ${session.count}\n); res.write(data: ${JSON.stringify(payload)}\n\n); }); }; // 模拟流式输出每 200ms 推 5 条左右 const timer setInterval(() { for (let i 0; i 5; i) { push(); } }, 200); req.on(close, () { clearInterval(timer); clearInterval(heartbeat); // 连接断开时保留 sessionCache下次用 lastEventId 续传 }); }); app.listen(3000, () { console.log(SSE server run at http://localhost:3000); });上面的代码有几个关键点需要解释id必须递增且连续客户端才能知道缺失了哪些块sessionCache存的是已生成内容实际业务里可以换成 Redis持久化到 generator 之外req.on(close)要清理定时器否则连接断开后服务端还在计算白白浪费资源。3.2 前端 SSE 客户端完整封装接下来是我认为最核心的部分用 fetch 封装一个 SSE 客户端。它要处理字节解码、按空行切帧、解析 data 行、维护 lastEventId还要支持心跳注释行忽略。我直接给一个可以直接拷贝的类// sse-client.js class SSEConnection { constructor({ url, headers {}, body null, onMessage, onError, onDone }) { this.url url; this.headers headers; this.body body; this.onMessage onMessage; this.onError onError; this.onDone onDone; this.abortController null; this.lastEventId 0; this.buffer ; this.decoder new TextDecoder(utf-8); this.retryTimer null; this.maxRetries 5; this.retryCount 0; } async connect() { this.abortController new AbortController(); try { const res await fetch(this.url, { method: this.body ? POST : GET, headers: { Content-Type: application/json, ...this.headers, }, body: this.body ? JSON.stringify(this.body) : undefined, signal: this.abortController.signal, }); if (!res.ok) { throw new Error(HTTP ${res.status}); } this.retryCount 0; const reader res.body.getReader(); while (true) { const { value, done } await reader.read(); if (done) break; this.buffer this.decoder.decode(value, { stream: true }); this.parseBuffer(); } this.onDone this.onDone(); } catch (err) { if (err.name AbortError) { return; } this.handleError(err); } } parseBuffer() { // SSE 用空行分隔事件 let eventEndIndex; while ((eventEndIndex this.buffer.indexOf(\n\n)) ! -1) { const rawEvent this.buffer.slice(0, eventEndIndex); this.buffer this.buffer.slice(eventEndIndex 2); const event this.parseEvent(rawEvent); if (!event) continue; if (event.id) { this.lastEventId event.id; } if (this.lastEventId this.lastEventId 0) { // 把 lastEventId 记录到 URL断线重连时传给后端 this.url this.updateUrlWithLastEventId(this.url, this.lastEventId); } this.onMessage this.onMessage(event); } } parseEvent(raw) { const lines raw.split(\n); const event { data: , eventName: message, id: undefined, retry: undefined }; for (const line of lines) { if (line.startsWith(:)) { // 注释行忽略 continue; } const colonIndex line.indexOf(:); const field colonIndex -1 ? line : line.slice(0, colonIndex); const value colonIndex -1 ? : line.slice(colonIndex 1).trimStart(); switch (field) { case data: // 多行 data 用换行拼接但 JSON 里一般不会有跨行 event.data (event.data ? \n : ) value; break; case event: event.eventName value; break; case id: event.id value; break; case retry: event.retry Number(value); break; default: break; } } return event; } updateUrlWithLastEventId(url, lastEventId) { // 简单处理把 lastEventId 作为 query 参数拼接 const sep url.includes(?) ? : ?; const queryKey lastEventId; const urlObj new URL(url, window.location.origin); urlObj.searchParams.set(queryKey, lastEventId); return urlObj.toString(); } handleError(err) { console.error([SSE] error:, err); if (this.retryCount this.maxRetries) { this.retryCount 1; const delay Math.min(1000 * Math.pow(2, this.retryCount), 15000); console.log([SSE] ${delay}ms 后重连当前次数 ${this.retryCount}); this.retryTimer setTimeout(() this.connect(), delay); } else { this.onError this.onError(err); } } disconnect() { if (this.retryTimer) { clearTimeout(this.retryTimer); } if (this.abortController) { this.abortController.abort(); } } } export default SSEConnection;这段代码的实际使用场景是用户发送一条消息后调用connect()建立连接onMessage里拿到 SSE 帧并渲染文字。如果连接意外断开类的内部会自动指数退避重连并带上lastEventId服务端从上次位置继续推。有一点我很早就踩过坑TextDecoder的{ stream: true }必须加上否则中文字符跨 chunk 的时候直接乱码。举个例子如果第一个 chunk 结尾是“你”第二个 chunk 开头是“好”不加 stream 模式可能第一次解码抛错或者输出乱码加了之后底层会缓存未完成字节等第二个 chunk 来了再一起解码。3.3 打字机渲染与断线重连集成光有 SSE 客户端还不够还需要把数据和渲染层接起来。我用一个简单的前端页面示例体现 buffer、displayIndex 和连接状态三者的配合。!DOCTYPE html html body div idstatus等待连接/div div idoutput/div button idsendBtn发送/button script typemodule import SSEConnection from ./sse-client.js; const outputEl document.getElementById(output); const statusEl document.getElementById(status); const sendBtn document.getElementById(sendBtn); let contentBuffer ; let displayIndex 0; let typeTimer null; let sseClient null; // 打字机渲染每帧最多渲染 3 个字符 function startTyping() { if (typeTimer) return; let lastTime 0; const typeLoop (time) { if (displayIndex contentBuffer.length) { typeTimer null; return; } const diff time - lastTime; if (diff 60) { const charsToShow 3; outputEl.textContent contentBuffer.slice(displayIndex, displayIndex charsToShow); displayIndex Math.min(displayIndex charsToShow, contentBuffer.length); outputEl.scrollTop outputEl.scrollHeight; lastTime time; } typeTimer requestAnimationFrame(typeLoop); }; typeTimer requestAnimationFrame(typeLoop); } function appendIncomingText(text) { contentBuffer text; // 如果此时没有打字机在运行立即启动 if (!typeTimer) { startTyping(); } // 如果打字机已经追不上最新内容可以逐步加速或一次多取字符 // 这里简化处理如果缓冲差距超过 50 字符下一帧直接多渲染 10 个 if (contentBuffer.length - displayIndex 50) { // 动态调整逻辑可以在 typeLoop 里实现 } } function renderEvent(event) { try { const data JSON.parse(event.data); if (data.content) { appendIncomingText(data.content); } if (data.done) { sseClient.disconnect(); statusEl.textContent 已完成; } } catch (e) { // 非 JSON 数据或半截内容直接作为文本追加 appendIncomingText(event.data); } } function connect(sessionId, lastEventId 0) { sseClient new SSEConnection({ url: /api/chat/stream?sessionId sessionId lastEventId lastEventId, onMessage: renderEvent, onError: () { statusEl.textContent 连接失败或重试耗尽; }, onDone: () { if (typeTimer) { // 全部渲染完成后再处理一次 Markdown } }, }); sseClient.connect(); } sendBtn.addEventListener(click, () { statusEl.textContent 连接中; contentBuffer ; displayIndex 0; outputEl.textContent ; connect(session-123); }); /script /body /html这套代码的核心逻辑是contentBuffer只管累积收到的文本displayIndex记录已经显示的位置requestAnimationFrame驱动渲染。连接断开时SSEConnection内部自动重连并带上lastEventId后端继续推新数据前端接着累加到contentBuffer尾部打字机继续跑用户基本感觉不到中间断过。4. 常见问题与排查技巧实录4.1 高频问题速查表我在这个项目里前后踩了几组坑整理成速查表遇到类似报错可以先对照现象 / 报错根因解决方案stream disconnected before completion: idle timeout waiting for sse网关或代理层的 idle timeout 到了比如 Nginx 默认 read timeout 60 秒SSE 长时间没有数据就被掐断服务端每 15 秒发一行注释心跳调大代理超时时间客户端断线自动重连并传 lastEventId打字机显示一段时间后突然停止但接口还在返回前端的定时器因为浏览器标签页切到后台被节流或暂停切换回页面时检查contentBuffer和displayIndex差距主动调一次渲染中文乱码字符被拆开在多个 chunk未正确处理 UTF-8TextDecoder加{ stream: true }消息重复显示断线重连后服务端从旧位置重新推了一遍完整内容前端没有做去重依赖id编号在onMessage里判断lastEventId是否小于当前 id重连后只追加增量页面疯狂重连请求风暴断线重连没有退避策略指数退避最大重试次数限制发送按钮连点两次发出两条消息前端没有禁用按钮也没有中断上一个请求发送后立即disabled或者用上一次的 AbortController 取消旧请求Markdown 渲染出现半截代码块或半截标签流式输出过程中就用 markdown 解析器渲染了不完整内容流结束后统一渲染或只在完整闭合的块出现时渲染未闭合用纯文本显示4.2 一次真实的 idle timeout 排查这个报错我觉得值得单独展开说。第一次遇到“idle timeout waiting for SSE”时我第一反应是后端代码有问题但接口明明在持续推数据。后来去看 Nginx 日志和错误信息才发现根因是用户侧和 AI 服务之间隔了一层负载均衡和一层 Nginx默认配置proxy_read_timeout 60s意思是如果 60 秒内没有从后端读到新数据连接就被判定为超时我的大模型在生成某些复杂回答时中间停顿可能超过 60 秒后端虽然连接还挂着但没有任何数据填充于是在停顿期间连接被网关掐断客户端收到这个报错。我在服务端加了 15 秒一次的注释行心跳后连接不会再出现 60 秒无数据的情况问题直接消失。同时我在前端加了断线重连逻辑即使中间被掐断也能自动续上。这个组合方案比单纯调大proxy_read_timeout更可靠因为调大只是延长了“无数据存活时间”并不能根治停顿而心跳是真正保持连接活跃的手段。4.3 Markdown 流式渲染的半截标签处理另一个高频问题是流式输出中直接解析 Markdown 导致半截标签出现在页面上。比如马上要输出一个代码块流式中间状态可能是这是一个示例代码下面是一个代如果这时的完整文本是“这是一个示例代码下面是一个代js”你用 Markdown 解析器渲染就会出现一个未闭合的代码块页面布局直接崩掉而且另一半内容还没到。我采用的方案分成三层第一层如果对交互要求不高最简单的是流式期间用innerText展示纯文本流结束后再渲染 Markdown。这样最稳但牺牲了“边生成边看到 Markdown 格式”的体验。第二层如果你必须边流式边渲染可以只允许渲染已经能安全闭合的块。具体做法是每次数据到达后把完整文本交给 Markdown 解析器但解析结果里如果包含未闭合标签就用上一份安全快照再把新增部分以纯文本追加到末尾。第三层对代码块这种特殊情况可以单独检测当前文本里的开闭数量是否为偶数如果是偶数且已经触发过代码块渲染才把代码块渲染成带高亮的样式否则只显示纯文本。实际操作中我推荐第一层简单稳定等流结束后再调用marked这类库做完整渲染用户往往不会太在意那几秒钟的格式差异但崩溃的页面布局会直接导致“不可用”的观感。5. 实战心得与后续扩展5.1 连接状态与用户输入的一致性除了流式本身还需要注意发送状态和连接状态的一致性。我见过不少项目在处理“连点两次发送两条消息”时只简单禁用按钮但没考虑取消上一轮未完成的请求。在高频点击后即使按钮禁用了之前已在途的请求依然会继续接收数据导致页面同时出现两条 AI 回复。正确的做法是在发送新消息之前调用sseClient.disconnect()或abortController.abort()再重新创建连接。如果你在状态管理里维护了一个isStreaming标志这个标志要在onDone、onError、disconnect三个地方都重置否则会出现发送按钮永远灰着的假死状态。还有一点SSE 断线重连时用户可能已经离开了当前页面。我的经验是切到后台时不要立即断掉连接浏览器对活跃 fetch 请求的管理和 EventSource 不同即使标签页不可见只要系统没有强杀页面连接通常还在。但如果确实是网络切换导致的断开重连逻辑会兜底这个不用太担心。5.2 还能继续扩展的几个方向这套 SSE 客户端封装已经能直接用于生产项目但它还可以继续增强一是把解析部分放到 Web Worker 里执行。当每秒数据块特别多时主线程同时做解码、JSON 解析、DOM 更新长文本场景下会有点压力。搬到 Worker 后主线程只负责把解析好的纯文本追加进去渲染更流畅。二是服务端缓存可以换成 Redis特别是多实例部署时内存缓存不能跨实例同步会导致用户重连后请求打到一个没有缓存的实例上续传失败。用 Redis 记录 topic 到已生成内容的映射每个实例都能从同一份缓存里续读。三是在前端增加“已完整显示内容”的持久化。比如用户刷新页面后从本地缓存里恢复已经生成的文字再通过 lastEventId 续传剩余部分。这个体验提升不小但要注意 token 数和文本长度的配合。最后提醒一句SSE 的retry字段在 EventSource 里默认为 3 秒我用 fetch 实现时通常会设置成 1 到 5 秒之间。太短会放大瞬时抖动导致请求风暴太长又会让用户明显感觉到“停住了”。我这边最终用的是指数退避从 1 秒起步最大 15 秒重试 5 次后给用户一个友好的错误提示不要把所有重试都放在后台静默执行。