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

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

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

协议本身不复杂，事件大致长这样：

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

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

关键点：

• 响应头是 Content-Type: text/event-stream
• 一条事件通常以空行 \\n\\n 结束
• 业务数据多放在 data: 后面（常见再包一层 JSON）

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

一句话对比

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

方式一：EventSource

基本用法

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

es.onmessage = (event) => {
// event.data 就是 data: 后面的字符串
const payload = JSON.parse(event.data);
console.log(payload);
};

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

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

如果服务端用了命名事件（event: progress），可以这样听：

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

优点

• 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 一条事件以 \\n\\n 结束，按段切最稳：

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 });
const segments = buffer.split('\\n\\n');
buffer = segments.pop() || '';

for (const segment of segments) {
const trimmed = segment.trim();
if (!trimmed.startsWith('data:')) continue;

const dataPart = trimmed.replace(/^data:\\s*/, '');
if (dataPart) onDataLine(dataPart);
}
}
} finally {
reader.releaseLock?.();
}
}

业务侧：把 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;
}
} catch {
// 忽略半包/脏行
}
}
);

取消流：AbortController

const abortController = new AbortController();

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

把 signal 传给 fetch，并在 reader.read() 循环里检查 signal.aborted，就能干净停掉。

优点

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

代价

• 要自己处理粘包、半包、解码（下一节展开）
• 没有浏览器那种「断了自动重连」，需要的话得自己补

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

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 一条事件以空行 \\n\\n 结束，所以：

每次 read 到一块字节
→ decode 成文本，追加到 buffer
→ 用 '\\n\\n' split
→ 最后一段多半是「还没收完的半包」，塞回 buffer
→ 前面那些完整段，再提取 data: 交给业务

对应代码核心就三行：

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

图示：

buffer 当前内容：
┌─────────────────────────────────────────────┐
│ data: {"event":"start"}\\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 把流缓冲住
});

// 可选：先写一行注释心跳，帮部分代理保持连接
res.write(': keep-alive\\n\\n');

// 推一条业务事件
res.write(`data: ${JSON.stringify({ event: 'content', data: '你好' })}\\n\\n`);

// 结束
res.end();

客户端断开时记得停掉后续写入：

req.on('close', () => {
aborted = true;
});

为什么常说「拿原生 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 跳过包装，例如看到已经是 text/event-stream 就原样放过：

// 伪代码：全局拦截器里
if (contentType.includes('text/event-stream')) {
return next.handle(); // 不要 map 成 { code, msg, data }
}
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
• 方式一 EventSource：简单、能自动重连，但基本限于 GET，难带自定义头
• 方式二 fetch + ReadableStream：可 POST、可带 Token、可 Abort，解析要自己写
• 粘包/半包靠 buffer + \\n\\n 分隔；解码用 TextDecoder({ stream: true })
• SSE 不要走普通 JSON 信封拦截器：自己 writeHead + 持续 write
• 选哪种，看你的接口要不要鉴权和请求体，大多数业务流式接口会选第二种