前端对接 SSE 的两种常见方式

LLM 流式输出、进度推送、长任务状态更新，后端经常会用 SSE（Server-Sent Events）：一条 HTTP 长连接，服务端持续往下推事件，客户端边收边渲染。

协议本身不复杂，线上常见长这样：

: ping

data: {"event":"content","data":"你好"}

data: {"event":"end","traceId":"abc"}

关键点：

• 响应头是 Content-Type: text/event-stream
• 一条事件以空行 \\n\\n 结束（HTTP 里也可能是 \\r\\n\\r\\n）
• data: 后面才是业务载荷（本文业务流再包一层 JSON）
• 以 : 开头的是 注释 / 心跳，不是业务事件，解析时直接丢掉

SSE 规范里也可以写原生 event: progress，浏览器 EventSource 能按名字监听。 本系列的业务生成接口走 fetch 自解析，把类型放进 JSON 的 event 字段——下一篇会讲为什么。连上之前，先把传输层拆对。

前端真正要选的，是怎么连上这条流。常规有两种：

一句话对比

业务 API 往往需要 Bearer Token + POST body，所以第二种更常见；监控面板、公开进度页，第一种更省事。

方式一：EventSource

基本用法

const es = new EventSource('/api/notifications/stream');

es.onmessage = (event) => {
// 只接收「未命名」事件（没有 event: 字段，或 event: message）
const payload = JSON.parse(event.data);
console.log(payload);
};

es.onerror = () => {
// 默认会自动重连；不需要时可 es.close()
console.error('SSE error');
};

// 主动断开
// es.close();

如果服务端用了命名事件（event: progress），onmessage 听不到，要按名字订阅：

es.addEventListener('progress', (event) => {
const payload = JSON.parse((event as MessageEvent).data);
console.log(payload);
});

原因很简单：浏览器把 SSE 的 event: 行当成「事件名」。

event: progress ← 这是名字，决定进哪个回调
data: {"percent":40} ← 这是数据，进 event.data

• 没有 event: 行（或写的是 event: message）→ 进 onmessage
• 写了 event: progress → 只进 addEventListener('progress')，onmessage 收不到

所以 EventSource 这边，类型写在协议的 event: 行里。 后面用 fetch 自己读流时，没有这套浏览器回调，类型放在 JSON 里更省事。下一篇会对比这两种写法。

优点

• API 短，上手快
• 断线自动重连（对「订阅型」推送很友好）
• 不用自己处理字节流和粘包；注释行也会被浏览器丢掉

限制（也是很多人最终换掉它的原因）

什么时候用它

• 不需要登录，或鉴权已靠 Cookie（同源自动带上）
• 接口本身就是 GET 订阅
• 你需要浏览器自带的断线重连

方式二：fetch + ReadableStream

思路：

发起带鉴权的 SSE 请求

async function requestAuthorizedSse(
url: string,
init: { method: string; body?: string; signal?: AbortSignal },
onDataLine: (dataLine: string) => void
) {
const token = localStorage.getItem('token') || '';

const response = await fetch(url, {
method: init.method,
headers: {
'Content-Type': 'application/json',
Authorization: `Bearer ${token}`,
},
body: init.body,
signal: init.signal,
});

if (!response.ok) {
const errorBody = await response.text().catch(() => '');
throw new Error(errorBody || `SSE request failed: ${response.statusText}`);
}

await readSseSegments(response, onDataLine, init.signal);
}

按「空行分段」解析（推荐）

SSE 一条事件以空行结束。段里可能有注释、也可能有多行字段，所以不要假设「整段都以 data: 开头」：

async function readSseSegments(
response: Response,
onDataLine: (dataLine: string) => void,
signal?: AbortSignal
) {
if (!response.body) {
throw new Error('SSE response has no body');
}

const reader = response.body.getReader();
const decoder = new TextDecoder();
let buffer = '';

try {
while (true) {
if (signal?.aborted) {
throw new DOMException('Aborted', 'AbortError');
}

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

buffer += decoder.decode(value, { stream: true }).replace(/\\r\\n/g, '\\n');
const segments = buffer.split('\\n\\n');
buffer = segments.pop() || '';

for (const segment of segments) {
for (const line of segment.split('\\n')) {
const trimmed = line.trim();
if (!trimmed || trimmed.startsWith(':')) continue; // 心跳 / 注释
if (!trimmed.startsWith('data:')) continue;
const dataPart = trimmed.slice('data:'.length).trimStart();
if (dataPart) onDataLine(dataPart);
}
}
}
} finally {
reader.releaseLock?.();
}
}

本系列约定：每条业务事件只有一行 data:，里面是完整 JSON。多行 data: 按规范要拼接，这里用不到。

业务侧：把 data 解析成事件

await requestAuthorizedSse(
'/api/generate/draft',
{
method: 'POST',
body: JSON.stringify({ projectId, task }),
signal: abortController.signal,
},
(dataLine) => {
try {
const data = JSON.parse(dataLine);
switch (data.event) {
case 'start':
console.log('开始', data.traceId);
break;
case 'content':
appendText(data.data); // 打字机效果
break;
case 'end':
finish(data);
break;
case 'error':
showError(data.data);
break;
default:
break; // ping 等非四件套，忽略即可
}
} catch {
// 忽略脏行；半包不该走到这里，靠的是上面的 buffer
}
}
);

取消流：AbortController

const abortController = new AbortController();

// 用户点「停止生成」
abortController.abort();

把 signal 传给 fetch，并在 reader.read() 循环里检查 signal.aborted，前端就能立刻停画。

这只取消了 HTTP 连接。服务端如果不听 req.close、不把同一把 AbortSignal 传给上游 LLM，模型仍会继续跑、token 照烧。下一篇会把这一层补全。

优点

• 支持 POST + JSON body（复杂生成任务很常见）
• 能带 Authorization 等自定义头
• 取消、超时、非 2xx 错误处理都更可控
• 和现有 API Client 风格容易统一

代价

• 要自己处理粘包、半包、解码（下一节展开）
• 没有浏览器那种「断了自动重连」，需要的话得自己补
• 心跳注释行要自己丢掉；漏掉会当脏数据去 JSON.parse

粘包、半包、解码到底怎么处理？

reader.read() 每次给你的不是「一条完整 SSE 事件」，而是一块块 字节（Uint8Array）。 网络怎么切包，你控制不了，所以会出现三种情况。

1）解码：字节 → 文本

TCP/HTTP 流里先是二进制。中文等多字节字符还可能被拆到两次 read() 中间。

const decoder = new TextDecoder();

// stream: true 很重要：告诉解码器「后面可能还有字节」
// 遇到半个汉字时先缓存，等下次凑齐再吐出完整字符
buffer += decoder.decode(value, { stream: true });

如果写成 decoder.decode(value)（默认 stream: false），半个 UTF-8 字符可能直接变成 `` 或乱码。

2）半包：一次 read() 不够一条事件

服务端本意推送：

data: {"event":"content","data":"你好"}\\n\\n

但第一次可能只收到：

data: {"event":"content","da

第二次才收到：

ta":"你好"}\\n\\n

如果每次 read() 立刻 JSON.parse，第一次必炸。

做法：先塞进 buffer，只处理已经完整的部分。

3）粘包：一次 read() 塞了多条事件

也可能一次就收到：

data: {"event":"start"}\\n\\n
data: {"event":"content","data":"你"}\\n\\n
data: {"event":"content","data":"好"}\\n\\n

如果只当一条处理，会漏事件或解析失败。

做法：用分隔符切开，循环处理每一段。

4）标准解法：缓冲区 + 分隔符

SSE 一条事件以空行结束，所以：

每次 read 到一块字节
→ decode 成文本，追加到 buffer（顺便把 \\r\\n 归一成 \\n）
→ 用 '\\n\\n' split
→ 最后一段多半是「还没收完的半包」，塞回 buffer
→ 前面那些完整段，逐行抽 data:；以 : 开头的心跳丢掉

对应代码核心就三行：

buffer += decoder.decode(value, { stream: true }).replace(/\\r\\n/g, '\\n');
const segments = buffer.split('\\n\\n');
buffer = segments.pop() || ''; // 半包留下，完整段拿去处理

图示：

buffer 当前内容：
┌─────────────────────────────────────────────┐
│ data: {"event":"start"}\\n\\n │ ← 完整，可处理
│ : ping\\n\\n │ ← 心跳，丢掉
│ data: {"event":"content","data":"你"}\\n\\n │ ← 完整，可处理
│ data: {"event":"cont │ ← 半包，留在 buffer
└─────────────────────────────────────────────┘
↑
segments.pop() 留着等下次

业务层 JSON.parse 再包一层 try/catch，是为了兜住脏数据； 真正防半包的，是上面的 buffer，不是 catch。

服务端要配合什么？

无论前端用哪种连法，服务端都要先把响应变成 SSE：

res.writeHead(200, {
'Content-Type': 'text/event-stream',
'Cache-Control': 'no-cache',
Connection: 'keep-alive',
'X-Accel-Buffering': 'no', // 避免 Nginx 把流缓冲住
});

const heartbeat = setInterval(() => {
if (!res.writableEnded) {
res.write(': ping\\n\\n');
}
}, 15_000);

req.on('close', () => {
clearInterval(heartbeat);
abortController.abort(); // 停写还不够，取消上游生成
});

res.write(`data: ${JSON.stringify({ event: 'content', data: '你好' })}\\n\\n`);
res.end();

心跳用 SSE 注释行（: ping）即可：目的是骗过 Nginx / 网关的空闲超时，不必进业务状态机。间隔按代理的 proxy_read_timeout 来，常见 15–30 秒。 JSON 里的换行交给 JSON.stringify，不要在协议外再手转一层 \\n。

res.write 在客户端慢时会把数据堆在 Node 内存里。生产里至少判断返回值，false 时等 drain 再继续写；下一篇给完整写法。

为什么常说「拿原生 res 自己写，别走普通 JSON 拦截器」？

普通接口的返回路径通常是：

Controller return { foo: 1 }
→ 拦截器 / 管道再包一层
→ 变成 { code: 0, msg: 'success', data: { foo: 1 } }
→ 框架一次性 JSON.stringify 后发给前端

SSE 要的是另一条路：

先写响应头 Content-Type: text/event-stream
→ 每隔一会儿 res.write('data: …\\n\\n')
→ 连接一直开着，最后再 res.end()

如果 SSE 也走「普通 JSON 拦截器」，常见会坏在三处：

所以 Nest 里常见写法是：

@Post('optimize/plan')
async optimizePlan(@Req() req, @Res() res) {
// @Res()：接管原生响应，框架不再替你自动 JSON.stringify
res.writeHead(200, { 'Content-Type': 'text/event-stream', /* … */ });
res.write(`data: ${JSON.stringify({ event: 'start' })}\\n\\n`);
// … 持续 write
res.end();
}

全局拦截器也要对 SSE 跳过包装。不要在 intercept() 开头读 Content-Type：那时 Handler 往往还没 writeHead，头是空的，跳过逻辑会失效。更稳的是按路径、自定义装饰器、或「该方法用 @Res() 接管了响应」来判断：

// 伪代码：按约定跳过，而不是赌响应头已经写好
const req = context.switchToHttp().getRequest();
if (isSseRoute(req)) {
return next.handle();
}
return next.handle().pipe(
map((data) => ({ code: 0, msg: 'success', data }))
);

一句话：

普通接口：框架帮你打包成 JSON 信封。 SSE：你自己按事件协议往响应里「一点一点写」，别让信封逻辑插手。

怎么选？

需要 POST body？或需要 Authorization Header？
├─ 是 → fetch + ReadableStream
└─ 否
├─ 需要自动重连的简单订阅 → EventSource
└─ 仍想统一客户端封装 → 也可以一律用 fetch

实战经验：

• 聊天/写作/长任务生成：几乎都是第二种
• 公告、公开看板、简单通知：第一种够用
• 团队若已有鉴权 API Client，优先第二种，少维护两套连接哲学

小结

• SSE 是服务端推事件的 HTTP 长连接，核心格式是 data: …\\n\\n；: ping 是心跳，不是业务事件
• 方式一 EventSource：简单、能自动重连，但基本限于 GET，难带自定义头
• 方式二 fetch + ReadableStream：可 POST、可带 Token、可 Abort，解析要自己写
• 粘包/半包靠 buffer + \\n\\n 分隔；解码用 TextDecoder({ stream: true })
• 前端 abort 只断 HTTP；上游 LLM 要服务端自己取消
• SSE 不要走普通 JSON 信封拦截器：自己 writeHead + 持续 write；跳过拦截器不要赌 Content-Type 已经写好
• 选哪种，看你的接口要不要鉴权和请求体，大多数业务流式接口会选第二种

系列导航

• 下一篇：SSE 事件设计与可中断的打字机效果