流式输出不只是加个 stream:true：SSE 的五个工程细节

前言

第一次做流式输出，代码通常是这样：

const res = await fetch('/api/chat', {
  method: 'POST',
  body: JSON.stringify({ messages, stream: true }),
});
const reader = res.body.getReader();
while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  const text = new TextDecoder().decode(value);
  appendToUI(text);
}

本地跑，字一个个出来，效果很好。上线之后问题一个接一个来。

这篇文章不讲"流式输出有什么好处"，只讲从能跑到能用之间那五个坑，以及怎么判断自己踩的是哪一个。

一、先搞清楚：为什么大模型流式用 SSE 而不是 WebSocket

大多数模型 API（OpenAI、通义、DeepSeek、华为云 ModelArts 的推理接口）都选 SSE。原因不复杂：

大模型推理的场景恰好是单向的：请求已经在 POST body 里发完了，之后只需要服务端持续往下推。用双向协议是浪费，而且多一层握手就多一层出故障的地方。

所以选 SSE 是对的。但"选对了协议"和"这条流能稳定跑"是两件事。

二、坑一：中间层把你的流缓冲了

现象：本地开发流式正常，部署到测试环境后，前端等十几秒，所有内容一次性出现——和不用流式没区别。

根因：链路上有某个组件在攒数据。按出现频率排：

怎么快速定位：跳过所有中间层直连后端。

# 直连后端，观察是否逐条到达
curl -N -H "Accept: text/event-stream" http://127.0.0.1:8080/chat

-N 是关键参数，关掉 curl 自己的缓冲。如果直连是流式的、走 nginx 就不是，问题 100% 在中间层。

然后逐层往回加：

curl -N https://your-domain.com/chat      # 过 nginx 了吗

一个容易漏的点：X-Accel-Buffering: no 是 nginx 专属的响应头，应用层加上就能覆盖 nginx 配置，比改 nginx.conf 更省事（不用动运维）。但如果前面还有一层不认这个头的网关，就得改那层的配置。

三、坑二：中文偶尔切成半个字，变成乱码

现象：流式输出英文永远正常，中文偶尔出现 �，或者某两个字变成方块。

根因：UTF-8 一个汉字占 3 字节，网络分片边界不保证落在字符边界上。第 N 个 chunk 可能正好切在一个汉字的中间：

完整字节流：  E4 BD A0  E5 A5 BD      （你好）
chunk 1：    E4 BD A0  E5            ← 后半截是"好"的第 1 个字节
chunk 2：    A5 BD

每个 chunk 单独 new TextDecoder().decode(value)，第一个 chunk 末尾那个孤立的 E5 无法解码，变成 �。

修法：TextDecoder 的 stream: true 选项，它会把不完整的字节序列留在内部缓冲，等下一个 chunk 拼上再输出。

const decoder = new TextDecoder('utf-8');
// ❌ 每个 chunk 独立解码
decoder.decode(value)

// ✅ streaming 模式：残留字节留到下次
decoder.decode(value, { stream: true })

还有一个收尾动作容易忘：流结束时，如果最后一个字节序列仍然不完整，需要用 decoder.decode()（不带参数）把内部残留冲出来。少了这一步，极少数情况下最后一个字会丢。

这个坑的隐蔽之处在于：它取决于网络分片，本地几乎不复现。判断方法是在处理逻辑里打一行日志，记录每个 chunk 的字节数和末尾几个字节的 hex，看是否出现单独的高位字节。

四、坑三：EventSource 带不了 Authorization 头

浏览器原生的 EventSource 很好用——自动重连、自动解析事件。但它有一个硬限制：不能设置自定义请求头。

// ❌ 做不到
const es = new EventSource(url, { headers: { Authorization: 'Bearer xxx' } });

标准 API 就没有这个参数。于是常见做法是把 token 塞进 query：

new EventSource(`/api/chat?token=${token}&q=...`)   // ⚠️ 有代价

代价有三个：

1. token 会进 nginx access log、浏览器历史、Referer
2. URL 长度有限制，prompt 稍长就爆
3. 日志脱敏要额外做规则

更稳的做法：用 fetch 自己读流。 代价是要自己解析 SSE 格式、自己实现重连，好处是完全可控——能带 header、能 POST 长 body、能做手动重连。

如果确实需要 EventSource 的自动重连，替代方案是用 Cookie 承载凭证（同源场景下 EventSource 会自动带上 Cookie），而不是把 token 放 query。

五、坑四：断线重连后，内容重复了一遍

现象：网络抖动一下，前端自动重连，用户看到已经输出过的段落又出现了一遍。

根因：SSE 协议本身提供了断点续传机制，但需要服务端配合：

id: 1
data: {"delta":"你"}

id: 2
data: {"delta":"好"}

浏览器重连时会带上 Last-Event-ID: 2 请求头。服务端必须读这个头，从第 3 条开始发。 如果你没实现，服务端就从头发，客户端就重复渲染。

// 服务端：读 Last-Event-ID 续传
const lastId = Number(req.headers['last-event-id'] || 0);
for (const chunk of chunks.slice(lastId)) {
  res.write(`id: ${chunk.id}\ndata: ${JSON.stringify(chunk.data)}\n\n`);
}

注意：fetch 自己实现的重连不会自动带 Last-Event-ID，需要你在客户端记录最后一个 id 并手动放进请求头。

如果续传实现不了怎么办：退一步，客户端在重连时清空已渲染内容重新开始，比拼接出乱序文本要好。至少用户看到的是完整的一段，而不是"你好你好"。

六、坑五：每个 token 都触发重渲染，页面卡死

现象：流式时 CPU 飙高、页面滚动卡顿，token 越快越卡。

根因：模型每秒能吐几十个 token，如果每个 token 都 setState，React 会做几十次渲染和 diff。渲染成本远高于网络成本。

修法一：按帧合并。 一个屏幕刷新周期（约 16ms）内到达的所有 token 合并成一次渲染：

let pending = '';
let scheduled = false;

function push(text) {
  pending += text;
  if (scheduled) return;
  scheduled = true;
  requestAnimationFrame(() => {
    scheduled = false;
    const chunk = pending;
    pending = '';
    appendToUI(chunk);        // 一帧只渲染一次
  });
}

修法二：服务端背压。 如果下游消费慢，服务端不该无限往连接里写。Node 的 res.write() 返回值就是信号：

if (!res.write(data)) {
  // 缓冲已满，等 drain 再继续，否则内存会涨
  await new Promise(r => res.once('drain', r));
}

不做背压的后果：慢客户端会导致服务端内存持续增长，极端情况下进程被 OOM 杀掉。这类问题在压测时通常发现不了——压测客户端都读得很快。

七、完整的 SSE 客户端解析器

上面五个坑都能落在一段代码里。下面这个版本处理了：字节边界、\r\n、注释心跳、多行 data、收尾冲刷。

/**
 * 逐事件读取 SSE 流
 * @param {string} url
 * @param {object} body   POST 的 JSON body
 * @param {AbortSignal} signal
 */
export async function* readSSE(url, body, signal) {
  const res = await fetch(url, {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      'Accept': 'text/event-stream',
    },
    body: JSON.stringify(body),
    signal,
  });
  if (!res.ok) throw new Error(`HTTP ${res.status}`);

  const reader = res.body.getReader();
  const decoder = new TextDecoder('utf-8');
  let buf = '';

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;

    // stream:true 让不完整的多字节字符留在 decoder 内部
    buf += decoder.decode(value, { stream: true });
    buf = buf.replace(/\r\n/g, '\n');          // 统一换行，兼容 \r\n\r\n 分隔

    let idx;
    while ((idx = buf.indexOf('\n\n')) >= 0) {
      const raw = buf.slice(0, idx);
      buf = buf.slice(idx + 2);
      const evt = parseEvent(raw);
      if (evt) yield evt;
    }
  }

  // 冲刷 decoder 残留 + 处理最后一个可能没有空行结尾的事件
  buf += decoder.decode();
  const tail = parseEvent(buf.trim());
  if (tail) yield tail;
}

function parseEvent(raw) {
  if (!raw.trim()) return null;
  let event = 'message';
  let id;
  const dataLines = [];

  for (const line of raw.split('\n')) {
    if (!line || line.startsWith(':')) continue;   // 空行 / 注释（心跳）
    const i = line.indexOf(':');
    const field = i === -1 ? line : line.slice(0, i);
    let val = i === -1 ? '' : line.slice(i + 1);
    if (val.startsWith(' ')) val = val.slice(1);   // 规范：冒号后可有一个空格

    if (field === 'data') dataLines.push(val);
    else if (field === 'event') event = val;
    else if (field === 'id') id = val;
    // retry 字段按需求忽略
  }

  if (!dataLines.length) return null;
  return { event, id, data: dataLines.join('\n') };
}

调用侧配合帧合并：

const controller = new AbortController();
let buffer = '';
let rafId = 0;

function flush() {
  rafId = 0;
  if (!buffer) return;
  const text = buffer;
  buffer = '';
  document.querySelector('#out').append(text);
}

for await (const evt of readSSE('/api/chat', { messages, stream: true }, controller.signal)) {
  if (evt.event === 'done') break;
  const obj = JSON.parse(evt.data);
  if (obj.delta) {
    buffer += obj.delta;
    if (!rafId) rafId = requestAnimationFrame(flush);
  }
}

做了什么

在一个内部问答页面上把流式链路重做了一遍，改动集中在四处：

1. 服务端加了 X-Accel-Buffering: no 和 Cache-Control: no-cache, no-transform，并把 text/event-stream 从 gzip 白名单里去掉
2. 客户端从 EventSource 换成 fetch + ReadableStream，凭证走请求头
3. TextDecoder 全部改成 stream: true，流结束时补一次 decode() 冲刷
4. 渲染改成 requestAnimationFrame 合并，服务端 res.write() 加了 drain 等待

改完之后体感变化明显的是第 4 点——之前 token 快的时候主线程被渲染占满，输入框都卡。

一些体会

流式的难点不在"怎么流"，在"流断了怎么办"。 本地开发时网络稳定、无代理，五个坑一个都碰不到。上线后它们会同时出现，而且症状很像：都是"内容不对"。

TextDecoder 的 stream: true 是必选项，不是可选优化。 这个坑的判定成本很低——打一行 chunk 长度日志就能看出来——但不知道它存在的话，会往"编码配置""字体"这些方向排查很久。

X-Accel-Buffering 比改 nginx.conf 划算。 应用层一个响应头解决，不用拉运维改配置、不用等发布窗口。

背压这件事，压测发现不了。 因为压测工具读得比生产客户端快得多。要验证背压是否生效，得故意做一个慢读客户端：

# 用 curl 的限速模拟慢客户端
curl -N --limit-rate 1k https://your-domain.com/chat

观察服务端内存是否持续增长。如果涨，说明没做背压。

最后一句判断标准：判断一条流式链路是否真的可用，不看它在办公室 WiFi 下多流畅，而看它在弱网 + 有代理 + 长时间输出这三个条件同时成立时的表现。