1. 从一个“打字机效果”说起：SSE 到底解决了什么问题

你可能已经用过 ChatGPT、文心一言或者各种 AI 对话产品，也一定注意到过那个细节：AI 的回答不是“啪”一下整段出现，而是一个字一个字往外蹦，像有人在屏幕后面实时打字。这个体验背后，前端和后端之间用的就是SSE（Server-Sent Events，服务器推送事件）。

SSE 是一种基于 HTTP 的单向、长连接、服务器主动推送技术。浏览器发起一个普通 HTTP 请求，服务器保持这个连接不关闭，持续往客户端写数据，每写一段，前端就渲染一段。它和 WebSocket 最大的区别在于：SSE 是单向的（服务器到浏览器），协议简单，走标准 HTTP，不需要额外握手升级，浏览器原生EventSource就能用。

那为什么 AI 场景偏爱 SSE，而不是 WebSocket？原因很实际：

• AI 推理是单向流式输出，用户提问后，服务器持续吐 token，不需要客户端在同一个连接里频繁回传数据；
• SSE 基于 HTTP/1.1 和 HTTP/2，天然兼容现有网关、负载均衡、鉴权体系，运维成本低；
• 浏览器断线自动重连是EventSource内置行为，省掉大量重连逻辑；
• 调试极其方便，curl就能看到流，排查问题比 WebSocket 友好太多。

这篇文章我会围绕SSE 在前端的完整落地来讲，覆盖EventSource原生用法、fetch + ReadableStream手动解析、AbortController中断控制、Vue 项目中的封装、FastAPI 后端的流式响应写法，以及那些只有踩过坑才知道的细节——比如stream disconnected before completion: idle timeout waiting for sse这种报错到底怎么来的。适合已经会 Vue、想搞懂流式交互的前端，也适合想给 AI 应用加“打字机效果”的全栈开发者。

2. SSE 协议本身：别被“服务器推送”四个字吓到

2.1 数据格式其实就四行规则

SSE 的协议简单到离谱，服务器返回的Content-Type是text/event-stream，body 是一段一段的纯文本，每段用两个换行符\n\n分隔。每一段内部可以包含这几个字段：

event: message data: {"content": "你"} id: 1 retry: 3000

• data：真正的数据内容，可以多行，多行会被拼成一个字符串；
• event：自定义事件名，前端可以addEventListener('message', ...)或监听自定义事件；
• id：事件 ID，断线重连时浏览器会带上Last-Event-ID请求头，服务端可以据此续传；
• retry：重连间隔毫秒数。

一个最朴素的流式响应长这样：

data: {"token": "你"} data: {"token": "好"} data: {"token": "，"} data: [DONE]

注意最后那个[DONE]，这是 OpenAI 风格接口的约定，前端看到它就主动关闭连接。这不是 SSE 协议规定的，而是业务层约定，但已经成了事实标准。

2.2 和 WebSocket、轮询的取舍

很多人一上来就问“为什么不用 WebSocket”。我做过对比，直接上表：

结论很清晰：单向流式场景，SSE 是性价比最高的选择。WebSocket 更适合需要客户端高频回传的场景，比如协同编辑、在线游戏。用 WebSocket 做 AI 流式输出，属于杀鸡用牛刀，还要额外处理心跳、重连、鉴权透传。

2.3 一个容易被忽略的限制

SSE 在 HTTP/1.1 下，同域名并发连接数受浏览器限制（Chrome 是 6 个）。如果你一个页面同时开多个 SSE 连接，很容易把连接池占满，导致其他请求排队。解决办法有两个：一是升级到 HTTP/2，多路复用不受这个限制；二是把多个流合并到一个连接里，用event字段区分类型。这个坑我在一个多模型对比的页面里踩过，同时开 4 个模型的流，加上页面其他请求，直接卡死。

3. 前端两种接流方式：EventSource 与 fetch 流

3.1 EventSource：三行代码就能跑

原生EventSource是最省事的方式：

const es = new EventSource('/api/chat/stream?q=你好') es.onmessage = (e) => { console.log('收到:', e.data) } es.onerror = (err) => { console.error('出错了', err) // EventSource 会自动重连，除非你手动 close } // 主动关闭 // es.close()

它的优点：自动重连、自动解析data:字段、支持自定义事件。缺点也很致命：

• 只能发 GET 请求，参数只能塞 URL，长文本提问很容易超长；
• 不能自定义请求头，想带Authorization只能靠 Cookie 或 query 参数；
• 无法读取响应状态码和响应头，服务端返回 401、500 时，前端只能从onerror里猜；
• 不能中断后精细控制，close()之后重连逻辑要自己接管。

所以现在做 AI 应用，主流方案是fetch+ReadableStream，把 SSE 当成普通流来读，自己解析。

3.2 fetch 流式读取：可控性拉满

async function streamChat(prompt, onToken, signal) { const resp = await fetch('/api/chat/stream', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ prompt }), signal, // AbortController 的 signal }) if (!resp.ok) { throw new Error(`HTTP ${resp.status}`) } const reader = resp.body.getReader() const decoder = new TextDecoder('utf-8') let buffer = '' while (true) { const { done, value } = await reader.read() if (done) break buffer += decoder.decode(value, { stream: true }) // 按 \n\n 切分事件块 const parts = buffer.split('\n\n') buffer = parts.pop() // 最后一段可能不完整，留到下次 for (const part of parts) { const line = part.trim() if (!line.startsWith('data:')) continue const data = line.slice(5).trim() if (data === '[DONE]') return try { const json = JSON.parse(data) onToken(json.token) } catch (e) { // 非 JSON 数据，按纯文本处理 onToken(data) } } } }

这段代码有几个关键点，我逐个拆：

第一，decoder.decode(value, { stream: true })的stream: true必须加。因为一个 UTF-8 中文字符占 3 个字节，网络分片很可能把一个汉字切成两半。不加这个参数，TextDecoder会把不完整的字节序列当成错误，输出乱码。加了之后，它会缓存不完整字节，等下一片到了再拼。

第二，buffer的残留处理。网络分片不会按\n\n边界来，所以每次read()拿到的可能是半个事件。必须把最后一段pop()出来留到下一轮，否则会解析出残缺 JSON。

第三，[DONE]判断要放在 JSON.parse 之前。因为[DONE]不是合法 JSON，直接 parse 会抛异常。

3.3 两种方式怎么选

我的经验是：

• 简单通知、日志推送、参数短小 →EventSource，省事；
• AI 对话、需要 POST、需要鉴权头、需要精细中断 →fetch流；
• 需要断线续传（Last-Event-ID）→EventSource或自己实现 ID 追踪。

现在绝大多数 AI 产品都走fetch流，因为提问内容长、要带 token、要能随时停止生成。

4. AbortController：让“停止生成”按钮真正生效

4.1 为什么必须用 AbortController

AI 生成到一半，用户点了“停止”，如果你只是把 UI 上的 loading 关掉，后端的推理还在跑，连接还开着，token 还在烧。这在按量计费的场景下就是真金白银的浪费。AbortController能真正中断 fetch 请求，浏览器会关闭底层连接，服务端也能感知到客户端断开。

const controller = new AbortController() // 发起请求时传入 signal streamChat(prompt, onToken, controller.signal) // 用户点停止 controller.abort()

abort()之后，reader.read()会抛出一个AbortError，你需要捕获它，避免控制台报红：

try { await streamChat(prompt, onToken, controller.signal) } catch (err) { if (err.name === 'AbortError') { console.log('用户主动中断') } else { console.error('真实错误', err) } }

4.2 一个 controller 只能中断一次

这是新手最容易踩的坑：AbortController是一次性的。一旦abort()，这个 controller 就废了，下次请求必须new一个新的。所以在 Vue 组件里，我通常这样组织：

import { ref } from 'vue' const loading = ref(false) let controller = null async function send(prompt) { // 每次请求前重置 controller = new AbortController() loading.value = true try { await streamChat(prompt, (t) => { /* 追加到消息 */ }, controller.signal) } catch (e) { if (e.name !== 'AbortError') throw e } finally { loading.value = false controller = null } } function stop() { controller?.abort() }

4.3 组件卸载时别忘了中断

如果用户在生成过程中切换了路由，组件卸载了，但流还在跑，回调里还在操作已经销毁的响应式数据，轻则警告，重则内存泄漏。在onUnmounted里补一刀：

import { onUnmounted } from 'vue' onUnmounted(() => { controller?.abort() })

这个细节看起来小，但在单页应用里非常关键。我见过一个项目，用户来回切页面，后台堆了十几个没关的流，最后浏览器直接卡死。

5. Vue 项目中的 SSE 封装实践

5.1 抽一个 composable 出来

在 Vue 3 里，把流式逻辑封装成 composable 是最舒服的写法。下面这个useSSEStream是我在多个项目里迭代出来的版本：

// composables/useSSEStream.js import { ref, onUnmounted } from 'vue' export function useSSEStream() { const loading = ref(false) const error = ref(null) let controller = null async function start({ url, body, onToken, onDone, onError }) { stop() // 防止并发 controller = new AbortController() loading.value = true error.value = null try { const resp = await fetch(url, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify(body), signal: controller.signal, }) if (!resp.ok) throw new Error(`HTTP ${resp.status}`) const reader = resp.body.getReader() const decoder = new TextDecoder('utf-8') let buffer = '' while (true) { const { done, value } = await reader.read() if (done) break buffer += decoder.decode(value, { stream: true }) const parts = buffer.split('\n\n') buffer = parts.pop() for (const part of parts) { const line = part.trim() if (!line.startsWith('data:')) continue const data = line.slice(5).trim() if (data === '[DONE]') { onDone?.() return } try { onToken(JSON.parse(data)) } catch { onToken(data) } } } onDone?.() } catch (e) { if (e.name === 'AbortError') return error.value = e onError?.(e) } finally { loading.value = false controller = null } } function stop() { controller?.abort() controller = null loading.value = false } onUnmounted(stop) return { loading, error, start, stop } }

5.2 在组件里怎么用

<script setup> import { ref } from 'vue' import { useSSEStream } from '@/composables/useSSEStream' const input = ref('') const answer = ref('') const { loading, start, stop } = useSSEStream() function send() { answer.value = '' start({ url: '/api/chat/stream', body: { prompt: input.value }, onToken: (data) => { answer.value += typeof data === 'string' ? data : data.token }, }) } </script> <template> <div> <textarea v-model="input" /> <button :disabled="loading" @click="send">发送</button> <button v-if="loading" @click="stop">停止</button> <pre>{{ answer }}</pre> </div> </template>

5.3 渲染性能：别每来一个 token 就触发一次大重排

answer.value += token这种写法，在 token 密集时（比如每秒几十个）会触发大量响应式更新。如果消息列表很长，页面会明显卡顿。两个优化方向：

• 批量更新：用一个缓冲区，每 50ms 或每积累 10 个 token 才 flush 一次到响应式变量；
• 虚拟滚动：消息列表用虚拟列表，只渲染可视区域。

批量更新的实现：

let pending = '' let timer = null function pushToken(t) { pending += t if (timer) return timer = setTimeout(() => { answer.value += pending pending = '' timer = null }, 50) }

这个 50ms 是我实测下来比较平衡的值，肉眼几乎感觉不到延迟，但渲染压力能降一个数量级。

6. FastAPI 后端：把流“吐”出来

6.1 最小可用的流式接口

FastAPI 用StreamingResponse就能实现 SSE：

from fastapi import FastAPI from fastapi.responses import StreamingResponse import asyncio, json app = FastAPI() async def event_generator(prompt: str): # 模拟逐字生成 for ch in f"你问的是：{prompt}": yield f"data: {json.dumps({'token': ch}, ensure_ascii=False)}\n\n" await asyncio.sleep(0.05) yield "data: [DONE]\n\n" @app.post("/api/chat/stream") async def chat_stream(payload: dict): return StreamingResponse( event_generator(payload.get("prompt", "")), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "X-Accel-Buffering": "no", # 关键：禁用 Nginx 缓冲 }, )

6.2 三个必须设置的响应头

第三个头是最容易被忽略的。Nginx 默认会缓冲上游响应，导致你后端明明在逐字 yield，前端却要等全部生成完才收到。加上X-Accel-Buffering: no或者在 Nginx 配置里写proxy_buffering off;才能解决。

6.3 对接 Ollama 这类本地模型

如果后端要转发 Ollama 的流，用httpx的流式请求：

import httpx from fastapi.responses import StreamingResponse @app.post("/api/chat/stream") async def chat_stream(payload: dict): async def gen(): async with httpx.AsyncClient(timeout=None) as client: async with client.stream( "POST", "http://localhost:11434/api/generate", json={"model": "qwen2", "prompt": payload["prompt"], "stream": True}, ) as resp: async for line in resp.aiter_lines(): if not line: continue data = json.loads(line) token = data.get("response", "") if token: yield f"data: {json.dumps({'token': token}, ensure_ascii=False)}\n\n" if data.get("done"): yield "data: [DONE]\n\n" break return StreamingResponse(gen(), media_type="text/event-stream")

注意timeout=None，因为模型推理可能很久，默认超时会中途掐断连接，前端就会看到stream disconnected before completion。

7. 那些让人抓狂的报错与排查

7.1 idle timeout waiting for sse 到底怎么回事

stream disconnected before completion: idle timeout waiting for sse这个报错，本质是连接空闲超时。链路里任何一环（浏览器、Nginx、网关、后端框架）都有一个“多久没数据就断开”的阈值。AI 首 token 延迟高的时候，比如模型要思考 30 秒才吐第一个字，中间这段时间连接上没有任何数据流动，就会被判定为空闲，直接掐断。

解决办法有几种，我一般组合使用：

• 后端发心跳：每隔 10~15 秒发一个注释行: ping\n\n，注释行会被 SSE 解析器忽略，但能保持连接活跃；
• 调大网关超时：Nginx 的proxy_read_timeout、proxy_send_timeout默认 60s，长任务要调到 300s 以上；
• 前端加超时提示：首 token 超过一定时间给用户一个“模型正在思考”的反馈，避免用户以为卡死。

心跳实现：

async def gen(): while not done: try: token = await asyncio.wait_for(queue.get(), timeout=15) yield f"data: {json.dumps({'token': token})}\n\n" except asyncio.TimeoutError: yield ": ping\n\n" # 心跳

7.2 常见问题速查表

7.3 几个只有踩过才知道的细节

第一，EventSource的自动重连是双刃剑。服务端返回 204 会停止重连，返回其他错误码会按retry间隔重连。如果你不希望它重连，必须手动close()。我有次调试，服务端一直返回 500，浏览器疯狂重连，日志刷屏。

第二，代理层可能改写Content-Type。有些网关会把text/event-stream改成text/plain，导致EventSource直接报错。用fetch流就没这个问题，这也是我倾向 fetch 的原因之一。

第三，reader.read()在连接被服务端关闭时，done会是true，但不代表正常结束。要结合业务判断，比如有没有收到[DONE]。没收到就说明是异常断开，可以提示用户重试。

第四，移动端浏览器切后台会挂起连接。用户切到微信再切回来，流可能已经断了。要做好重连或续传，或者干脆在切后台时主动中断，回来重新发起。

8. 生产环境的加固建议

8.1 鉴权怎么传

EventSource不能自定义头，所以用fetch流时，Authorization正常带就行。如果用EventSource，只能靠 Cookie 或 query 参数。query 参数传 token 有泄露风险（会进日志、进 Referer），不推荐。所以需要鉴权的场景，直接用 fetch 流。

8.2 限流与并发控制

SSE 连接是长连接，一个用户开多个标签页就是多条连接。后端要限制单用户并发流数量，否则容易被拖垮。简单做法是用 Redis 记录用户活跃连接数，超过阈值就拒绝新连接。

8.3 日志与可观测性

流式接口的日志和普通接口不一样，请求持续时间长，要记录：首 token 延迟、总时长、token 数、是否被中断。这些指标对优化体验很关键。首 token 延迟高，用户就会觉得“卡”；总时长和 token 数能帮你估算成本。

8.4 降级方案

不是所有环境都支持流式。有些企业内网代理会缓冲所有响应，SSE 直接失效。这时候要有降级：检测到流式不可用，自动切换成普通请求，等全部生成完再一次性返回。虽然体验差，但至少能用。

9. 我个人的一些实战体会

做流式交互这两年，最大的感受是：SSE 本身不难，难的是链路上每一环都可能给你使绊子。浏览器、Nginx、网关、后端框架、模型服务，任何一环的默认配置都可能把流“攒”起来或者“掐”断。所以排查流式问题时，我的习惯是从后往前逐段验证：先用curl直接打后端，确认后端在逐字吐；再加一层 Nginx，看是否还流式；最后才看前端。这样能快速定位是哪一环的问题。

另一个体会是，中断控制一定要做扎实。用户点“停止”是高频操作，如果只是 UI 层面停掉，后端还在烧 token，成本会悄悄涨上去。AbortController配合后端感知客户端断开，是必须的。

最后分享一个小技巧：调试 SSE 时，在浏览器 Network 面板里，找到那个流式请求，点开 Response 标签，能看到数据实时追加。如果这里是一次性出现的，那基本可以确定是缓冲问题，往 Nginx 或网关方向查；如果是逐段出现的，那问题就在前端解析逻辑。这个判断方法帮我省了大量时间。