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

你肯定用过那种 AI 对话产品，问它一个问题，它的回答不是“啪”一下整段甩出来，而是一个字一个字往外蹦，像有人在屏幕后面敲键盘。这个体验背后，前端拿到的其实是一条持续不断的单向数据流，而不是一次性的完整响应。实现这种效果的主流方案之一，就是SSE（Server-Sent Events，服务器推送事件）。

先把结论摆在这：SSE 是一种基于 HTTP 的服务器到客户端的单向流式通信技术。浏览器发起一个普通请求，服务器保持连接不关闭，持续往客户端推送文本数据，客户端通过EventSource接口接收。它天然适合 AI 流式回答、实时日志、进度推送、股票行情这类“服务器主动说、客户端只管听”的场景。

那它和 WebSocket 有什么区别？为什么 AI 场景大多选 SSE 而不是 WebSocket？这是很多人第一个卡壳的地方。我刚开始接触的时候也疑惑，后来踩过几次坑才明白：AI 回答本质上是“一问一答”的请求-响应模型，只是响应被拆成了很多片。用 WebSocket 相当于为了喝杯水专门修了条双向高速公路，而 SSE 就是那根现成的水管，接上 HTTP 就能用，服务端实现简单，客户端浏览器原生支持，还自带断线重连。选型上的这点差异，直接决定了你后面代码的复杂度。

这篇文章我打算把 SSE 从前到后讲透：协议长什么样、前端怎么接、后端（以 FastAPI 为例）怎么推、Vue 里怎么封装、中断怎么处理、线上会遇到哪些坑。适合正在做 AI 对话、实时推送功能的前端和后端同学，也适合想搞懂“流式响应”底层原理的开发者。看完你应该能自己从零搭一套可用的流式对话链路。

2. SSE 协议拆解：为什么它能“一个字一个字蹦出来”

2.1 一次 SSE 请求的完整生命周期

要理解 SSE，先得看它底层到底传了什么。SSE 本质上就是一个长连接的 HTTP 响应，响应头里有个关键字段：

Content-Type: text/event-stream Cache-Control: no-cache Connection: keep-alive

text/event-stream这个 MIME 类型是 SSE 的身份证。浏览器一看到它，就知道“这不是一次性返回的普通响应，而是一条流”，于是不会等整个响应结束，而是边收边解析。这就是“一个字一个字蹦出来”的物理基础——数据不是攒齐了再给你，而是来一段解析一段。

服务器推送的数据有固定格式，每条消息由若干字段组成，字段之间用换行分隔，消息之间用空行分隔：

data: 你 data: 好 data: 呀 data: 这是第二条消息

这里有个容易搞混的点：多个data:行会被拼接成一条消息，中间用换行符连接。也就是说上面第一组三行data:其实是一条消息，内容是你\n好\n呀。而空行才是消息的分隔符。很多人第一次写 SSE 解析逻辑时，就是因为没搞懂这个规则，导致消息粘在一起或者被错误切分。

除了data，还有几个常用字段：

id和retry这两个字段是 SSE 相比“自己用 fetch 读流”的额外优势。浏览器原生EventSource会自动记住最后收到的id，断线重连时把它放进Last-Event-ID请求头，服务端就能知道从哪继续。这个机制在 AI 长回答场景里非常值钱——网络抖一下不至于让用户从头再问一遍。

2.2 为什么 AI 场景偏爱 SSE 而不是 WebSocket

我做过几个 AI 对话项目，选型时反复权衡过。结论是：除非你需要客户端频繁主动往服务器推数据，否则 SSE 几乎总是更优解。原因有这么几条。

第一，协议简单，调试友好。SSE 就是 HTTP，你可以直接用curl看流：

curl -N http://localhost:8000/chat

-N关闭缓冲，你就能在终端里看到数据一行行冒出来。WebSocket 就没这么直观，得专门写客户端或者用工具。调试成本低，意味着出问题时你能更快定位。

第二，服务端实现轻。FastAPI 里返回一个StreamingResponse就完事了，不需要维护连接状态机、不需要处理握手升级。WebSocket 虽然 FastAPI 也支持，但连接管理、心跳、房间这些都得自己写。

第三，天然适配 HTTP 基础设施。SSE 走标准 HTTP 端口，负载均衡、网关、鉴权中间件基本都能复用。当然这里有个坑后面会讲——很多网关默认会缓冲响应，导致流式效果失效，这个必须处理。

第四，自动重连是白送的。EventSource内置重连逻辑，配合id字段还能续传。自己用 fetch 读流的话，这些都得手写。

那 SSE 的短板在哪？它是单向的，客户端不能通过这条连接发数据；另外浏览器对同域名下的 SSE 连接数有限制（HTTP/1.1 下通常是 6 个），不过 HTTP/2 多路复用后这个问题基本消失。所以如果你的场景是“客户端也要高频推消息”，比如协同编辑、游戏，那还是 WebSocket 更合适。

2.3 EventSource 与 fetch 流读取：两条技术路线怎么选

前端接收 SSE，有两条路：用原生EventSource，或者用fetch+ReadableStream手动读。

EventSource的优点是简单，几行代码就能跑：

const es = new EventSource('/api/chat') es.onmessage = (e) => { console.log(e.data) } es.onerror = (err) => { console.error('连接出错', err) }

但它有几个硬伤：不能自定义请求头（比如没法带Authorization）、只能发 GET 请求、不能传复杂请求体。而 AI 对话通常需要 POST 一段 JSON 过去，还要带 token，这时候EventSource就歇菜了。

所以实际项目里，我更常用fetch+ 流读取：

const response = await fetch('/api/chat', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ prompt: '你好' }), signal: controller.signal }) const reader = response.body.getReader() const decoder = new TextDecoder() while (true) { const { done, value } = await reader.read() if (done) break const chunk = decoder.decode(value, { stream: true }) // 解析 chunk 里的 SSE 格式 }

这条路灵活得多，能带 header、能 POST、能配合AbortController中断。代价是重连逻辑要自己写，SSE 格式也要自己解析。我的建议是：简单场景用 EventSource，AI 对话这种需要 POST + 鉴权 + 中断的场景，老老实实用 fetch 流读取。后面 Vue 封装那节我会给一套完整的解析方案。

3. 后端怎么把数据“挤”出来：FastAPI 流式响应实战

3.1 StreamingResponse 的核心用法

FastAPI 里实现 SSE，核心就是StreamingResponse配合一个异步生成器。生成器每yield一次，数据就往客户端推一次。这是整个流式效果的源头。

from fastapi import FastAPI from fastapi.responses import StreamingResponse import asyncio app = FastAPI() async def event_generator(): for char in "你好呀，这是一段流式回答": yield f"data: {char}\n\n" await asyncio.sleep(0.1) @app.get("/chat") async def chat(): return StreamingResponse( event_generator(), media_type="text/event-stream" )

注意yield的内容格式：data: xxx\n\n。结尾必须是两个换行，第一个换行结束data字段，第二个换行表示消息结束。少一个换行，浏览器就不会触发onmessage，数据会一直卡在缓冲区里。这个坑我踩过，当时盯着屏幕纳闷为什么数据不显示，最后发现就是少了个\n。

media_type="text/event-stream"必须写对，否则浏览器不会按流处理。另外建议加上禁用缓冲的响应头：

return StreamingResponse( event_generator(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "X-Accel-Buffering": "no", # 关键：禁用 Nginx 缓冲 } )

X-Accel-Buffering: no这个头是给 Nginx 看的，告诉它别缓冲这个响应。没有它，Nginx 会把你的流攒起来一次性发出去，流式效果直接消失。这是线上部署最常见的坑之一，本地开发好好的，一上服务器就变成“整段蹦出来”，八成就是它。

3.2 对接大模型：把 token 流转发出去

真实场景里，数据源通常是大模型的流式接口。以调用本地模型服务为例，思路是边收边转：

import httpx async def model_stream(prompt: str): async with httpx.AsyncClient(timeout=None) as client: async with client.stream( "POST", "http://localhost:11434/api/generate", json={"model": "qwen", "prompt": prompt, "stream": True} ) as resp: async for line in resp.aiter_lines(): if not line: continue # 假设上游返回 JSON，取出文本片段 import json try: data = json.loads(line) token = data.get("response", "") if token: yield f"data: {json.dumps({'text': token})}\n\n" except json.JSONDecodeError: continue yield "event: done\ndata: [DONE]\n\n"

这里有几个设计考量。第一，用aiter_lines()而不是aiter_bytes()，因为上游通常按行返回 JSON，按行读更省事。第二，把 token 包成 JSON 再放进data，因为 token 里可能有换行符，直接塞进 SSE 的data字段会破坏格式。包一层 JSON 转义后就安全了。第三，结束时发一个event: done，让前端知道流结束了，可以关闭 loading 状态。

timeout=None这个设置也别忽略。大模型生成慢的时候，默认超时可能几十秒就断了，导致回答到一半戛然而止。设成None表示不超时，但生产环境建议设一个合理的大值，比如 300 秒，避免连接泄漏。

3.3 处理客户端断开：别让生成器空转

有个容易被忽视的问题：用户关掉页面或者点了“停止生成”，后端怎么知道？如果不处理，生成器还在傻乎乎地跑，白白消耗算力。

FastAPI 里可以通过request.is_disconnected()检测：

from fastapi import Request @app.post("/chat") async def chat(request: Request): async def gen(): for char in "很长的回答内容": if await request.is_disconnected(): print("客户端已断开，停止生成") break yield f"data: {char}\n\n" await asyncio.sleep(0.1) return StreamingResponse(gen(), media_type="text/event-stream")

每次循环前检查一下连接状态，断开就break。这个检查很轻量，但能省下大量无效计算。我在一个项目里忘了加这个，结果用户频繁刷新页面，后端堆了一堆僵尸生成任务，CPU 直接飙满。加上之后问题立刻消失。

4. Vue 前端封装：从裸流到可复用的对话组件

4.1 手写一个健壮的 SSE 解析器

前面说了，AI 场景用fetch流读取更合适。但fetch拿到的是一段段Uint8Array，它不保证每次read()返回的正好是一条完整消息。可能半条消息被切在两个 chunk 里，也可能一个 chunk 里塞了好几条消息。所以必须自己维护一个缓冲区来拼接和切分。

function createSSEParser(onMessage) { let buffer = '' return function parse(chunk) { buffer += chunk // SSE 消息以空行分隔，兼容 \n\n 和 \r\n\r\n const parts = buffer.split(/\r?\n\r?\n/) // 最后一段可能不完整，留在缓冲区 buffer = parts.pop() for (const part of parts) { if (!part.trim()) continue let eventType = 'message' const dataLines = [] for (const line of part.split(/\r?\n/)) { if (line.startsWith('data:')) { dataLines.push(line.slice(5).trimStart()) } else if (line.startsWith('event:')) { eventType = line.slice(6).trim() } } if (dataLines.length) { onMessage({ event: eventType, data: dataLines.join('\n') }) } } } }

这段代码的关键在buffer = parts.pop()。最后一段永远留在缓冲区，因为它可能是被截断的半条消息，等下一个 chunk 来了再拼。这个“留尾巴”的思路是流式解析的通用套路，处理任何分块数据都用得上。

另外注意dataLines.join('\n')，这对应前面说的“多个data:行拼成一条消息”的规则。还有line.slice(5).trimStart()，data:后面那个空格是可选的，规范里建议去掉一个前导空格，用trimStart更稳妥。

4.2 用 AbortController 实现“停止生成”

AI 回答经常需要中途停止，这时候AbortController就派上用场了。它和fetch的signal配合，能干净地中断请求。

import { ref } from 'vue' export function useChat() { const answer = ref('') const loading = ref(false) let controller = null async function send(prompt) { answer.value = '' loading.value = true controller = new AbortController() const parse = createSSEParser(({ event, data }) => { if (event === 'done') return try { const { text } = JSON.parse(data) answer.value += text } catch (e) { // 忽略解析失败的分片 } }) try { const resp = await fetch('/api/chat', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ prompt }), signal: controller.signal }) const reader = resp.body.getReader() const decoder = new TextDecoder() while (true) { const { done, value } = await reader.read() if (done) break parse(decoder.decode(value, { stream: true })) } } catch (err) { if (err.name === 'AbortError') { console.log('用户主动停止') } else { console.error('请求出错', err) } } finally { loading.value = false controller = null } } function stop() { if (controller) controller.abort() } return { answer, loading, send, stop } }

这里有个细节值得说：decoder.decode(value, { stream: true })里的stream: true很重要。一个多字节字符（比如中文）可能被切在两个 chunk 之间，如果不用流式解码，就会出现乱码。TextDecoder加上这个选项后，会把不完整的字节留在内部缓冲，等下一个 chunk 补齐再输出。这个坑在处理中文流时特别常见，很多人看到乱码还以为是编码问题，其实是分块切断导致的。

AbortError的判断也别漏。用户主动停止时，fetch会抛一个name为AbortError的异常，这是正常流程，不该当成错误弹提示。区分开“用户停止”和“真实错误”，体验会好很多。

4.3 在组件里落地：一个最小可用的对话界面

把上面的逻辑接到 Vue 组件里，就是一个能跑的对话界面：

<template> <div class="chat"> <div class="messages"> <div v-for="(msg, i) in messages" :key="i" class="msg"> {{ msg }} </div> <div v-if="loading" class="answer">{{ answer }}</div> </div> <div class="input-area"> <input v-model="input" @keyup.enter="handleSend" /> <button v-if="!loading" @click="handleSend">发送</button> <button v-else @click="stop">停止</button> </div> </div> </template> <script setup> import { ref } from 'vue' import { useChat } from './useChat' const input = ref('') const messages = ref([]) const { answer, loading, send, stop } = useChat() async function handleSend() { if (!input.value.trim()) return const prompt = input.value messages.value.push(`我：${prompt}`) input.value = '' await send(prompt) messages.value.push(`AI：${answer.value}`) } </script>

这个组件虽然简单，但把核心链路都串起来了：输入、发送、流式渲染、停止。实际项目里你可能会加上 Markdown 渲染、代码高亮、滚动到底部这些，但骨架就是这个。

有个体验优化点：流式渲染时不要每来一个 token 就触发一次 Vue 的响应式更新。如果 token 来得很快，频繁更新会导致页面卡顿。可以用requestAnimationFrame或者定时器做节流，把多个 token 攒一帧再更新。我在一个高频场景里做过对比，节流后 CPU 占用明显下降。

5. 线上部署与排查：那些文档里不会写的坑

5.1 流式失效的三大元凶

本地跑得好好的，一上线就变成“整段蹦出来”，这是 SSE 最经典的线上问题。根据我的排查经验，元凶基本就三个。

第一是 Nginx 缓冲。前面提过，加X-Accel-Buffering: no响应头能解决。如果还不行，检查 Nginx 配置里有没有proxy_buffering on，改成off：

location /api/chat { proxy_pass http://backend; proxy_buffering off; proxy_cache off; proxy_set_header Connection ''; proxy_http_version 1.1; chunked_transfer_encoding off; }

proxy_http_version 1.1和Connection ''这两行是为了让 Nginx 用长连接转发，否则可能被当成短连接处理。

第二是压缩中间件。有些框架默认开启 gzip 压缩，而压缩需要攒够一定数据量才输出，流式效果就没了。SSE 响应应该排除在压缩之外。

第三是应用层缓冲。比如某些 WSGI 服务器会缓冲响应，或者代码里用了print而不是yield。确认你的生成器是真的在yield，而不是先拼成完整字符串再返回。

排查顺序建议从外到内：先用curl -N直连后端，确认后端没问题；再经过 Nginx 测，确认网关没问题；最后到浏览器。这样能快速定位是哪一层在缓冲。

5.2 超时与断连：idle timeout 怎么破

热词里有个stream disconnected before completion: idle timeout waiting for sse，这是非常典型的报错。意思是连接空闲太久，被某一层超时机制掐断了。

SSE 连接如果长时间没有数据推送，中间的任何一层（浏览器、Nginx、负载均衡、后端服务器）都可能因为“空闲超时”把它关掉。解决办法有两个方向。

一是发心跳。定期推一个注释行（以:开头）或者空消息，保持连接活跃：

async def gen(): while True: try: token = await queue.get(timeout=15) yield f"data: {token}\n\n" except asyncio.TimeoutError: yield ": keep-alive\n\n" # 注释行，客户端会忽略

注释行以冒号开头，客户端解析时会忽略，但能起到“我还活着”的作用，防止超时。

二是调大各层超时。Nginx 的proxy_read_timeout、负载均衡的空闲超时、后端的请求超时，都要检查一遍。默认值往往只有 60 秒，对长回答来说不够。

我一般两个都做：心跳间隔设成超时时间的一半左右，比如超时 60 秒就 25 秒发一次心跳，双保险。

5.3 常见问题速查表

把踩过的坑整理成一张表，方便对照排查：

这张表基本覆盖了 90% 的线上问题。遇到新问题，先往这几个方向套，通常能快速定位。

5.4 几个提升稳定性的实操心得

最后分享几个我在实际项目里总结的小技巧。

给流加一个总时长上限。万一模型卡住不返回，连接会一直挂着。设一个最大生成时间，超时就主动结束并返回已生成的内容，比无限等待体验好。

前端做重试要谨慎。SSE 断线后如果无脑重试，可能重复请求导致重复计费或重复回答。建议只在“还没收到任何数据”时才自动重试，已经收到部分内容就提示用户手动重试。

日志要记录流的开始和结束。流式请求的生命周期比普通请求长，出问题时如果没有日志，很难判断是卡在哪一步。在生成器开始和结束时各打一条日志，排查效率会高很多。

测试时用真实网络环境。本地 localhost 几乎不会触发超时和缓冲问题，一定要在接近生产的网络环境里测，才能暴露真实问题。

SSE 这套东西，原理不复杂，但细节特别多，每一个细节都可能让你在线上多熬一个晚上。把协议格式、缓冲、超时、中断这几块吃透，剩下的就是熟练度问题了。我个人在实际操作中的体会是，与其等出了问题再排查，不如一开始就把心跳、断开检测、缓冲禁用这些防御性措施都加上，后面能省下大量救火时间。