跳到主要内容

SSE 消息粘连与截断详解

SSE（Server-Sent Events）看起来很简单：后端不断 write，前端不断接收。但线上最常见的两个问题就是：

  1. 消息粘连：一次读取拿到多条消息，边界混在一起。
  2. 消息截断：一次读取只拿到半条消息，甚至把一个 UTF-8 字符切断。

如果你把“每次读取到的 chunk 就当成一条完整消息”来处理，基本一定会踩坑。

一、先明确：SSE 的“消息边界”是什么​

SSE 的响应头一般是：

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

SSE 不是按 TCP 包分消息，而是按协议文本分帧。一条完整事件由多行字段组成，以一个空行结束：

id: 101
event: message
data: {"type":"notice","text":"hello"}

关键点：

  1. event、id、retry、data 都是字段行。
  2. 一条消息结束标记是 \n\n（或 \r\n\r\n）。
  3. 多行 data: 要拼接成同一条消息体。

二、什么是“消息粘连”和“消息截断”​

2.1 消息粘连​

本来推了两条事件，前端一次 read() 却拿到：

data: {"n":1}

data: {"n":2}

这不是协议错误，而是正常网络行为。因为 TCP 是字节流，不保证一条 write 对应一次 read。

2.2 消息截断​

一条事件可能被拆成两次读取：

第一次：

data: {"msg":"你

第二次：

好，SSE"}

如果你第一段就 JSON.parse，必报错。更隐蔽的是 UTF-8 多字节字符被切开，会出现乱码或替换字符。

三、根因拆解（为什么一定会发生）​

3.1 TCP 是字节流，不是消息队列​

TCP 只保证字节顺序，不保证应用层消息边界。你调用 3 次 res.write，客户端可能 1 次、2 次或更多次才能读完。

3.2 中间层缓冲会改变分片节奏​

反向代理、网关、CDN、压缩器都可能缓冲数据，再批量吐给客户端，导致 chunk 边界与事件边界更不一致。

3.3 编码层面的截断​

UTF-8 中文通常是 3 字节，一个字符可能跨 chunk。若解码器没用流式模式，会产生乱码或丢字。

四、前端正确解析方案（核心）​

目标：永远不要按 chunk 当消息，要按 SSE 空行分帧。

4.1 解析流程​

  1. 用 TextDecoder('utf-8') 流式解码（stream: true）。
  2. 把解码文本追加到 buffer。
  3. 把 \r\n 统一成 \n。
  4. 循环查找 \n\n，每次取出一帧完整事件。
  5. 逐行解析字段并组装 {id, event, data, retry}。

4.2 Fetch Stream 解析示例​

type SSEEvent = {
  id?: string;
  event?: string;
  data: string;
  retry?: number;
};

function parseSSEFrame(frame: string): SSEEvent | null {
  const lines = frame.split('\n');
  const dataLines: string[] = [];
  let id: string | undefined;
  let event: string | undefined;
  let retry: number | undefined;

  for (const line of lines) {
    if (!line || line.startsWith(':')) continue; // 空行或注释心跳
    const idx = line.indexOf(':');
    const field = idx === -1 ? line : line.slice(0, idx);
    let value = idx === -1 ? '' : line.slice(idx + 1);
    if (value.startsWith(' ')) value = value.slice(1);

    if (field === 'data') dataLines.push(value);
    else if (field === 'id') id = value;
    else if (field === 'event') event = value;
    else if (field === 'retry') retry = Number(value);
  }

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

export async function consumeSSE(url: string, onMessage: (e: SSEEvent) => void) {
  const res = await fetch(url, {
    headers: {Accept: 'text/event-stream'},
  });
  if (!res.body) throw new Error('SSE stream is empty');

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

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

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

    let splitIndex = buffer.indexOf('\n\n');
    while (splitIndex !== -1) {
      const frame = buffer.slice(0, splitIndex);
      buffer = buffer.slice(splitIndex + 2);
      const evt = parseSSEFrame(frame);
      if (evt) onMessage(evt);
      splitIndex = buffer.indexOf('\n\n');
    }
  }

  // 尝试处理流结束时残留的最后一帧
  const tail = buffer.trim();
  if (tail) {
    const evt = parseSSEFrame(tail);
    if (evt) onMessage(evt);
  }
}
实战建议

浏览器环境能用原生 EventSource 时优先用它；只有在你需要自定义 POST、鉴权头、Abort 控制、统一流式处理时，才用 fetch + ReadableStream 自己解析。

五、服务端如何降低粘连与截断带来的风险​

5.1 严格按 SSE 格式输出​

function writeSSE(res: import('http').ServerResponse, data: unknown, event = 'message', id?: string) {
  if (id) res.write(`id: ${id}\n`);
  res.write(`event: ${event}\n`);
  res.write(`data: ${JSON.stringify(data)}\n\n`);
}

5.2 发送心跳防止连接被中间层闲置关闭​

setInterval(() => {
  res.write(': heartbeat\n\n'); // 注释行，客户端可忽略
}, 15000);

5.3 关闭代理缓冲（Nginx 示例）​

location /sse {
    proxy_pass http://backend;
    proxy_http_version 1.1;
    proxy_set_header Connection '';
    proxy_buffering off;
    chunked_transfer_encoding on;
    add_header X-Accel-Buffering no;
    add_header Cache-Control no-cache;
}

5.4 断线重连与去重​

  1. 服务端给每条事件带 id。
  2. 客户端重连时发送 Last-Event-ID 或自己带上最后消费的 id。
  3. 服务端按 id 补发，避免丢消息或重复消费。

六、排障清单（线上可直接用）​

七、面试高频问答​

Q1：SSE 消息粘连是后端写错了吗？​

答：不一定。大部分粘连是正常网络行为。TCP 没有消息边界，多个事件被一次读取拿到是常态。正确做法是客户端按 SSE 协议分帧，而不是按读取 chunk 分帧。

Q2：SSE 截断和乱码为什么常一起出现？​

答：因为 UTF-8 是多字节编码，字符可能跨 chunk。若解码器不用流式解码，半个字符会被错误处理，表现为乱码或替换符，同时业务 JSON 也可能被截断。

Q3：EventSource 还会有粘连问题吗？​

答：底层网络层面仍然会有分片，但 EventSource 已帮你做了协议解析，所以业务层通常感知不到粘连。问题更多出现在自己用 fetch 读取流并手写解析时。

Q4：为什么 SSE 必须以空行结束一条消息？​

答：这是 SSE 协议的事件边界标记。没有空行，客户端无法判断一条事件何时结束，就会出现消息拼接错误或一直等待。

Q5：如何保证断线重连后不丢消息？​

答：使用事件 id。客户端记录最后消费的 id，重连时带上该 id，请求服务端补发缺失区间。业务端同时要做幂等处理，防止重复消费。

Q6：为什么线上比本地更容易出现粘连/截断问题？​

答：线上通常多了网关、反向代理、CDN、WAF、压缩链路，这些中间层会改变分片节奏和缓冲策略。本地直连后端时问题不明显，但线上会被放大。

Q7：SSE 和 WebSocket 在消息边界上有什么本质差异？​

答：WebSocket 是消息帧协议，边界由协议层保证；SSE 基于 HTTP 文本流，边界由应用层空行约定。SSE 解析更依赖客户端分帧实现是否正确。

Q8：排查 SSE 解析错误时优先看哪三件事？​

答：

  1. 客户端是否“缓冲后按 \n\n 分帧”；
  2. TextDecoder 是否使用流式解码；
  3. 代理层是否关闭缓冲并设置了心跳保活。