1. 背景：SSE 数据格式 {背景}
在理解代码之前，必须先理解 SSE 协议的原始数据格式。
1.1 服务端发送的原始字节流
服务端发送:
POST /api/ai/chat → 服务端开始流式返回
原始字节流（服务端连续发送）:
┌─────────────────────────────────────────────┐
│ event: delta\n │
│ data: {"content":"你","done":false}\n │
│ \n │ ← 空行，表示一个事件结束
│ event: delta\n │
│ data: {"content":"好","done":false}\n │
│ \n │
│ event: delta\n │
│ data: {"content":"","done":true}\n │
│ \n │
└─────────────────────────────────────────────┘
1.2 SSE 协议规范
┌─────────────────────────────────────────────────┐
│ SSE 事件格式（RFC 规范）: │
│ │
│ [field]: [value]\n │
│ [field]: [value]\n │
│ \n │ ← 空行 = 事件分隔符
│ │
│ field 只有 4 种：data / event / id / retry │
│ │
│ 示例： │
│ event: delta\n ← event 字段 │
│ data: {"content":"你"}\n ← data 字段 │
│ \n ← 事件结束 │
└─────────────────────────────────────────────────┘
1.3 浏览器接收到的是什么
HTTP 响应体（ReadableStream<Uint8Array>）:
chunk 1: [101 118 101 110 116 ...] ← Uint8Array 字节数组
chunk 2: [100 97 116 97 58 ...]
chunk 3: [123 34 99 111 110 116 ...] ← 可能被任意截断！
注意：TCP 数据包的边界与 SSE 事件边界完全无关
这就是为什么需要 手动——将乱序、任意截断的字节流，解析成业务可用的结构化数据。
2. EventSource 的工作方式
EventSource 是浏览器内置的专用 API，专门用来消费 SSE。EventSource 是浏览器专为 SSE 封装的高级 API，能够自动解析data/event/id/retry、自动分行、自动解码，服务器可指定重连间隔，断网自动重试。
简单示例：
const es = new EventSource('/time');
es.onmessage = (e) => console.log('收到时间：', e.data);
es.onerror = (e) => console.log('出错了');
你只需要关心业务逻辑，连接管理和协议解析都由浏览器搞定。
但是有一个关键的缺点，不支持传递自定义请求头，只能用 GET。因此，对于大型项目，我们只能使用更加原生流式请求fetch + ReadableStream的方式。fetch 是通用流式传输，不仅能做 SSE，还能处理文件下载、视频流、自定义二进制流。
3. fetch + ReadableStream 的工作方式
首先来一段代码，下面是使用了EventSource实现的流式对话的一个方法，我们直接监听onmessage方法取到数据，这个数据已经是格式化后的数据，将这个数据追加到message中就可以实现流式的效果。现在，我们需要将它改造为fetch + ReadableStream的方式。
export const connectSSE = (message: string, options: SSEOptions) => {
let currentAttempt = 0;
let delay = 1000;
let timeoutId: NodeJS.Timeout | null = null;
let isClosed = false;
let eventSource: EventSource | null = null;
const closeConnection = () => {
isClosed = true;
if (eventSource) {
eventSource.close();
eventSource = null;
}
if (timeoutId) {
clearTimeout(timeoutId);
timeoutId = null;
}
};
const connect = () => {
eventSource = new EventSource(
`/api/ai/chat?message=${encodeURIComponent(message)}`,
);
eventSource.onmessage = (event) => {
try {
const data = JSON.parse(event.data);
if (data.done) {
eventSource?.close();
options.onDone();
closeConnection();
} else {
options.onChunk(data.content);
}
} catch (error) {
options.onError(new Error("解析SSE数据失败"));
}
};
eventSource.onerror = () => {
if (isClosed) return;
eventSource?.close();
options.onError(new Error("SSE连接错误"));
if (currentAttempt < DEFAULT_RETRY_CONFIG.maxRetries) {
currentAttempt++;
timeoutId = setTimeout(() => {
connect();
}, delay);
} else {
options.onError(new Error(`SSE连接失败，已重试${currentAttempt}次`));
closeConnection();
}
};
};
connect();
return closeConnection;
};
fetch 返回的 Response 对象中的 body 是一个 ReadableStream（可读流）。它只提供最原始的字节数据，不解析任何格式，你需要手动处理：
-
发起请求：可以自定义 method、headers、body，不限于 GET。
-
获取流：
response.body 是一个 ReadableStream<Uint8Array>。
-
读取数据：通过
getReader() 获得读取器，循环读取数据块（chunk）。
-
解析协议：需要自己将二进制 chunk 解码为字符串，按行分割，检测 SSE 格式（
data:、\n\n 等）。
-
处理重连：如果需要自动重连，必须自己实现逻辑。
-
取消连接：可以调用
reader.cancel() 或 abortController.abort()。
简单示例：
const response = await fetch('/time', { method: 'GET' });
const reader = response.body.getReader();
const decoder = new TextDecoder();
let buffer = '';
while (true) {
const { done, value } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
// 解析 buffer 中的 SSE 消息...
// 检测 data: ...\n\n，提取数据，然后清空已处理部分
}
上面的value就是最新返回的数据流，通过解码后就是我们的文本块了，应该怎么处理呢。
EventSource 方式：
Server SSE → EventSource 解析 → onmessage 回调
Fetch + Stream 方式：
Server SSE → ReadableStream
↓
TextDecoder（字节→文本）
↓
行缓冲（处理分割）
↓
SSE 行解析（data: 前缀）
↓
JSON.parse
↓
onChunk 回调
首先，解码后的值赋值给buffer，此时buffer是一个字符串，它可能含有多条语句，因此需要进行切割。buffer.split("\n")将buffer分割为含有多个语句的数组，需要注意的是数组的最后一个元素可能并不是以\n结束的因此先不要处理。
为了处理这种分块，我们需要维护一个累积的缓冲区（buffer 变量），将每次新收到的 chunk 追加到缓冲区末尾，然后再一起处理。
export const connectSSEFetch = async (
message: string,
options: SSEOptions
) => {
const abortController = new AbortController();
try {
const response = await fetch(
`/api/ai/chat?message=${encodeURIComponent(message)}`,
{ signal: abortController.signal }
);
const reader = response.body!.getReader();
const decoder = new TextDecoder();
let buffer = "";
while (true) {
const { done, value } = await reader.read();
if (done) {
options.onDone();
break;
}
buffer += decoder.decode(value, { stream: true });
const lines = buffer.split("\n");
buffer = lines.pop() || ""; // 保留最后一行
for (const line of lines) {
if (line.startsWith("data: ")) {
const data = JSON.parse(line.slice(6));
if (data.done) {
options.onDone();
return;
}
options.onChunk(data.content);
}
}
}
} catch (err) {
options.onError(err as Error);
}
return () => abortController.abort();
};
举个例子- 例如，当前缓冲区内容为：
data: Hello\n
data: Wor
这里 "data: Wor" 没有换行符结尾，它只是单词 "World" 的一部分，下一个 chunk 可能会带来 "ld\n\n"。 split('\n') 会得到：
["data: Hello", "data: Wor"]
其中 "data: Wor" 是最后一项，它不是以换行符结尾的完整行，而是一个不完整的行。如果现在处理这一行，就会得到错误的数据。
所以，我们不能处理最后一行，必须把它留到下一次，等后续数据到来拼完整后再处理。
到现在为止，逻辑还是比较清晰的，对比之下，Fetch只是多了一些流数据处理的逻辑，是不是可以把这部分抽取出来呢。
// 高阶函数：创建 SSE 行处理器
// buffer 和 decoder 在闭包中维持状态，跨 chunk 使用
const createSSELineProcessor = () => {
let buffer = "";
const decoder = new TextDecoder();
return {
// 处理单个数据 chunk
processChunk: (chunk: Uint8Array): boolean => {
// if (dataEndReceived) return false;
// 1. 转换字节为文本（stream: true 表示可能是多字节字符的中间部分）
buffer += decoder.decode(chunk, { stream: true });
// 2. 按换行符分割成行
const lines = buffer.split("\n");
// 3. 最后一行可能不完整，保留到下一个 chunk
buffer = lines.pop() || "";
// 4. 处理完整的行
for (const line of lines) {
// 空行是 SSE 中的消息分隔符，跳过
if (!line.trim()) continue;
const data = parseSSELine(line);
if (!data) continue; // 不是 data 行，跳过
if (data.done) {
options.onDone();
return false; // 返回 false 表示应该停止处理
}
// 处理数据块
options.onChunk(data.content);
}
return true; // 返回 true 表示继续处理
},
// 处理流结束时的剩余数据
flush: (): void => {
// 最终解码（处理任何待处理的多字节字符）
const finalText = decoder.decode();
if (finalText.trim()) {
const data = parseSSELine(finalText);
if (data && !data.done) {
options.onChunk(data.content);
}
}
},
};
};
这里我们对流数据处理逻辑进行了抽取，buffer作为闭包保存，使用的时候只需要p=createSSELineProcessor,然后调用p.processChunk即可。依旧是每轮判断有没有done,有的话就退出。
while (true) {
const { done, value } = await reader.read();
if (done) {
break;
}
const shouldContinue = processor.processChunk(value);
if (!shouldContinue) {
break;
}
}
}
还有两个细节，原来的函数有flush，这个函数是干嘛的，什么时候调用呢。
TextDecoder 是浏览器原生 API，用于将 二进制字节流（Uint8Array） 解码为文本字符串，它有两种解码模式，对应两种调用方式：
带参数 + stream: true（流式解码，核心用法）
decoder.decode(chunk, { stream: true })
-
作用：分片、连续解码二进制流（比如 SSE / 文件流，数据是一块块传输的）；
-
关键特性：如果遇到不完整的多字节字符（比如中文、 emoji，UTF-8 占 3 字节），解码器不会强行解码，而是把残留字节缓存在解码器内部，等待下一块数据拼接完整后再解码；
- 不会出现乱码，专门处理流式分片数据。
不带参数（刷新 / 收尾解码，最终调用）
decoder.decode()
-
官方定义：结束流式解码，刷新解码器内部的所有残留字节；
-
作用：把之前流式解码时缓存的未完成字节，一次性全部解码成字符串；
-
副作用：清空解码器的内部缓冲区，标志着整个解码流程结束。
SSE 流是分块传输的，当服务器关闭流、传输彻底结束时，会存在两个残留数据问题：
-
解码器内部残留：
TextDecoder 用 stream: true 缓存了不完整的多字节字符，没有新 chunk 了，必须手动调用 decode() 刷新；
-
缓冲区残留：闭包中的
buffer 可能还剩最后一行不完整文本，没有后续换行符触发处理。
如果不调用 flush()，这部分数据会直接丢失，flush() 是流传输结束后的收尾方法，专门处理最后残留的、未被解码 / 未被处理的数据，防止数据丢失。
还有一个细节，reader最后注意释放， cancel主动终止整个可读流（ReadableStream）。调用后，流会被关闭，后续无法再从该流读取任何数据，并且任何挂起的读请求都会立即完成或失败。 releaseLock释放当前读取器对流的锁定，但不关闭流。调用后，该读取器不再关联任何流，流可以被其他读取器（或异步迭代器）再次锁定。
比如
// 假设已经用 reader 读了一部分
const firstChunk = await reader.read();
// 处理 firstChunk...
// 释放锁，然后才能用 for await 或 XStream
reader.releaseLock();
// 现在可以安全地使用 for await
for await (const chunk of response.body) {
// ...
}
如果中间没有释放，机会报错，一个流被两个代码读取了。
再补充一点，mdn对于流的读取还提到了这样一个例子：
const response = await fetch("https://www.example.org");
let total = 0;
// Iterate response.body (a ReadableStream) asynchronously
for await (const chunk of response.body) {
// Do something with each chunk
// Here we just accumulate the size of the response.
total += chunk.length;
}
// Do something with the total
console.log(total);
for await...of 本质上就是异步迭代的语法糖，它与手动 while 循环实现的是完全相同的流式处理效果。ReadableStream从2024年开始逐步实现了异步可迭代协议。Chrome 124（2024年4月稳定版）正式增加了对ReadableStream的异步迭代支持。区别在于代码风格：for await...of 更简洁、更符合直觉，而手动循环则暴露了更多的底层控制细节。
手动 while 循环通常是这样写的：
const reader = stream.getReader();
while (true) {
const { done, value } = await reader.read();
if (done) break;
// 处理 value
}
现在还可以采用更加优雅的方式。
if (stream[Symbol.asyncIterator]) {
for await (const chunk of stream) {
const shouldContinue = processor.processChunk(chunk);
// 如果接收到 done 信号，继续消耗流直到结束
if (!shouldContinue ) {
break;
}
}
} else {
throw new Error("浏览器不支持 for await...of");
}
4. XStream 整体架构与逻辑链路
根据官网，@ant-design/x-sdk 提供了一系列的工具API，旨在帮助开发人员开箱即用的管理AI对话应用数据流，其中的XStream用于转换可读数据流。先看看用法
import { XStream } from '@ant-design/x';
async function request() {
const response = await fetch();
// .....
for await (const chunk of XStream({
readableStream: response.body,
})) {
console.log(chunk);
}
}
是不是跟上面的很像，其实就是做了进一步的底层封装和优化，逻辑都差不多，下方看看怎么实现的。
源代码
4.1 数据流转图
服务端 HTTP 响应
│
▼
ReadableStream<Uint8Array> ← 原始字节流（TCP 数据包）
│
│ .pipeThrough(decoderStream)
▼
ReadableStream<string> ← UTF-8 文本流（可能在字符中间截断）
│
│ .pipeThrough(splitStream('\n\n'))
▼
ReadableStream<string> ← SSE 事件字符串流（按 \n\n 分割）
│
│ .pipeThrough(splitPart('\n', ':'))
▼
ReadableStream<SSEOutput> ← 结构化 SSE 对象流
│
│ stream[Symbol.asyncIterator]
▼
AsyncGenerator<SSEOutput> ← 支持 for await...of 消费
4.2 实际数据转换示例
输入（Uint8Array 字节）:
[101, 118, 101, 110, 116, 58, 32, 100, 101, 108, 116, 97, ...]
─── decoderStream ───→
输入（string，可能不完整）:
"event: delta\ndata: {\"con" ← chunk 1
"tent\":\"你好\"}\n\n" ← chunk 2
─── splitStream('\n\n') ───→
输出（完整事件，按 \n\n 分割）:
"event: delta\ndata: {\"content\":\"你好\"}"
─── splitPart('\n', ':') ───→
输出（SSEOutput 对象）:
{
event: "delta",
data: "{\"content\":\"你好\"}"
}
5. 核心 API 讲解
5.1 ReadableStream
ReadableStream 是浏览器中表示"可读数据流"的原生 API。
// 创建一个自定义 ReadableStream
const stream = new ReadableStream<string>({
start(controller) {
// 流启动时调用
controller.enqueue('Hello'); // 推入数据
controller.enqueue(' World');
controller.close(); // 关闭流
},
cancel(reason) {
// 消费者取消时调用
console.log('流被取消:', reason);
}
});
// 消费方式1: getReader()
const reader = stream.getReader();
const { done, value } = await reader.read(); // { done: false, value: 'Hello' }
reader.releaseLock();
// 消费方式2: for await...of（需要 asyncIterator 支持）
for await (const chunk of stream) {
console.log(chunk); // 'Hello', ' World'
}
关键特性：
- 流只能被读取一次（single-consumer）
- 一次只能有一个 reader
-
enqueue 将数据推入内部队列
-
close 发出 done 信号
5.2 TransformStream
TransformStream 是可读流和可写流的组合，专门用来转换数据。
// 基本结构
const transform = new TransformStream<Input, Output>({
transform(chunk, controller) {
// 每次有 chunk 时调用
// controller.enqueue() 将转换后的数据推入输出流
// controller.error() 报错
},
flush(controller) {
// 流结束时调用，处理剩余缓冲数据
// 对应 TextDecoder 的 decode() 无参版本
},
start(controller) {
// 初始化时调用（可选）
}
});
// 使用示例：大写转换器
const upperCaseTransform = new TransformStream<string, string>({
transform(chunk, controller) {
controller.enqueue(chunk.toUpperCase());
}
});
// 消费
const writer = upperCaseTransform.writable.getWriter();
const reader = upperCaseTransform.readable.getReader();
writer.write('hello');
const { value } = await reader.read(); // 'HELLO'
与 上文 中 createSSELineProcessor 的对比:
// sse.ts：手动管理状态的处理器（命令式）
const createSSELineProcessor = () => {
let buffer = '';
const decoder = new TextDecoder();
return {
processChunk(chunk: Uint8Array): boolean {
buffer += decoder.decode(chunk, { stream: true });
// ... 手动处理 buffer 和行分割
return true;
},
flush(): void {
// 手动处理剩余数据
}
};
};
// XStream：使用 TransformStream（声明式，职责更清晰）
new TransformStream<string, string>({
transform(streamChunk, controller) {
// 只关注转换逻辑，不关注如何读写流
buffer += streamChunk;
// ...
},
flush(controller) {
// 框架自动调用，无需手动触发
}
});
5.3 pipeThrough
pipeThrough 将一个 ReadableStream 接入 TransformStream，返回新的 ReadableStream。
// 基本用法
const outputStream = inputStream.pipeThrough(transformStream);
// 链式调用（XStream 的精华）
const processedStream = rawStream
.pipeThrough(decoderStream) // Uint8Array → string
.pipeThrough(splitStream()) // string → SSE事件块
.pipeThrough(splitPart()); // SSE事件块 → {event, data}
// 类比：Linux 管道
// cat file.txt | grep "error" | awk '{print $1}'
// ↓ ↓ ↓
// ReadableStream pipeThrough pipeThrough
关键特性：
- 自动处理背压（backpressure）：下游消费慢时，自动暂停上游写入
- 惰性求值：不消费就不处理
- 返回新的 ReadableStream，原始流不可再用
// pipeThrough 内部等价于：
readable
.pipeTo(transform.writable) // 原流连接到 transform 的可写端
.then(/* 完成 */);
return transform.readable; // 返回 transform 的可读端
5.4 TextDecoderStream
专门用于字节→字符串转换的原生 TransformStream。
// 原生 TextDecoderStream（现代浏览器支持）
const nativeDecoder = new TextDecoderStream('utf-8');
// 等价于：
const polyfillDecoder = new TransformStream({
transform(chunk, controller) {
controller.enqueue(decoder.decode(chunk, { stream: true }));
},
flush(controller) {
controller.enqueue(decoder.decode()); // 处理剩余字节
}
});
// 使用场景：正确处理被切断的多字节字符
const chineseChar = '你'; // UTF-8: [E4 BD A0]（3字节）
// 不用 stream: true 的问题：
decode([0xE4]) // 错误：'?'（字节不完整）
decode([0xBD, 0xA0]) // 错误：'??'
// 用 stream: true 的正确处理：
decode([0xE4], { stream: true }) // ''（等待后续字节）
decode([0xBD, 0xA0], { stream: true }) // '你'（完整输出）
createDecoderStream 的兼容处理：
function createDecoderStream() {
// 优先使用原生 API（性能更好）
if (typeof TextDecoderStream !== 'undefined') {
return new TextDecoderStream();
}
// 降级到 polyfill（Safari 旧版等不支持的环境）
const decoder = new TextDecoder('utf-8');
return new TransformStream({ /* ... */ });
}
5.5 Symbol.asyncIterator
让对象支持 for await...of 语法。
// 基本用法
const asyncIterable = {
[Symbol.asyncIterator]: async function*() {
yield 1;
yield 2;
yield 3;
}
};
for await (const value of asyncIterable) {
console.log(value); // 1, 2, 3
}
// XStream 的实现：给 ReadableStream 附加异步迭代器
stream[Symbol.asyncIterator] = async function*() {
const reader = this.getReader(); // this = stream
while (true) {
const { done, value } = await reader.read();
if (done) break;
if (!value) continue;
yield value; // 每次 yield 一个已转换好的 SSEOutput
}
};
// 使用效果：
const xStream = XStream({ readableStream: response.body });
for await (const event of xStream) {
console.log(event); // { event: 'delta', data: '{"content":"你好"}' }
}
6. XStream 各函数详解
6.1 splitStream：SSE 事件块分割器
function splitStream(streamSeparator = '\n\n') {
let buffer = ''; // 闭包维持跨 chunk 状态
return new TransformStream<string, string>({
transform(streamChunk, controller) {
buffer += streamChunk;
const parts = buffer.split(streamSeparator); // 按 \n\n 分割
parts.slice(0, -1).forEach(part => {
if (isValidString(part)) controller.enqueue(part); // 推出完整事件
});
buffer = parts[parts.length - 1]; // 保留最后不完整的部分
},
flush(controller) {
if (isValidString(buffer)) controller.enqueue(buffer); // EOF 时推出最后数据
}
});
}
执行过程追踪：
输入 chunk 1: "event: delta\ndata: {\"con"
buffer = "event: delta\ndata: {\"con"
parts = ["event: delta\ndata: {\"con"] ← 只有1部分，无完整事件
enqueue: 无
buffer = "event: delta\ndata: {\"con"
输入 chunk 2: "tent\":\"你\"}\n\nevent: de"
buffer = "event: delta\ndata: {\"content\":\"你\"}\n\nevent: de"
parts = [
"event: delta\ndata: {\"content\":\"你\"}", ← 完整事件！
"event: de" ← 不完整
]
enqueue: "event: delta\ndata: {\"content\":\"你\"}" ✅
buffer = "event: de"
flush 时:
buffer = "event: de"（如有剩余）
enqueue: "event: de"
6.2 splitPart：键值对解析器 {#splitpart}
function splitPart(partSeparator = '\n', kvSeparator = ':') {
return new TransformStream<string, SSEOutput>({
transform(partChunk, controller) {
// 输入: "event: delta\ndata: {\"content\":\"你好\"}"
const lines = partChunk.split(partSeparator);
const sseEvent = lines.reduce<SSEOutput>((acc, line) => {
const separatorIndex = line.indexOf(kvSeparator); // 找第一个 ':'
if (separatorIndex === -1) return acc;
const key = line.slice(0, separatorIndex).trim();
if (!isValidString(key)) return acc; // 跳过注释行（: 开头）
const value = line.slice(separatorIndex + 1).trim();
return { ...acc, [key]: value };
}, {});
if (Object.keys(sseEvent).length === 0) return;
controller.enqueue(sseEvent);
}
});
}
细节：为什么用 indexOf 而不是 split(':')？
// 假设 data 内容中包含冒号：
const line = 'data: {"url":"https://example.com"}';
// ❌ split(':') 会把 URL 中的冒号也切割
line.split(':') // ['data', ' {"url"', '"https', '//example.com"}']
// ✅ indexOf 只找第一个冒号
const i = line.indexOf(':'); // 4
const key = line.slice(0, 4).trim(); // 'data'
const value = line.slice(5).trim(); // '{"url":"https://example.com"}'
执行过程追踪：
输入: "event: delta\ndata: {\"content\":\"你好\"}"
lines = [
"event: delta",
"data: {\"content\":\"你好\"}"
]
reduce 过程：
line "event: delta":
separatorIndex = 5
key = "event"
value = "delta"
acc = { event: "delta" }
line "data: {\"content\":\"你好\"}":
separatorIndex = 4
key = "data"
value = "{\"content\":\"你好\"}"
acc = { event: "delta", data: "{\"content\":\"你好\"}" }
enqueue: { event: "delta", data: "{\"content\":\"你好\"}" }
6.3 createDecoderStream：兼容性解码器 {#createdecoderstream}
function createDecoderStream() {
// 优先使用原生 API
if (typeof TextDecoderStream !== 'undefined') {
return new TextDecoderStream();
}
// 降级 polyfill：手动实现相同逻辑
const decoder = new TextDecoder('utf-8');
return new TransformStream({
transform(chunk, controller) {
controller.enqueue(decoder.decode(chunk, { stream: true }));
},
flush(controller) {
controller.enqueue(decoder.decode()); // 确保最后字节被正确输出
},
});
}
兼容性说明：
-
TextDecoderStream：Chrome 71+, Firefox 105+, Safari 14.1+
- polyfill：理论上覆盖所有支持
TransformStream 的浏览器
6.4 XStream 主函数 {#xstream-main}
function XStream<Output = SSEOutput>(options: XStreamOptions<Output>) {
const { readableStream, transformStream, streamSeparator, partSeparator, kvSeparator } = options;
const decoderStream = createDecoderStream();
// 构建管道链（两种模式）
const stream = (
transformStream
? // 模式A：自定义 transformStream（用户完全自控解析逻辑）
readableStream
.pipeThrough(decoderStream) // Uint8Array → string
.pipeThrough(transformStream) // string → Output（用户定义）
: // 模式B：默认 SSE 解析（三段式管道）
readableStream
.pipeThrough(decoderStream) // Uint8Array → string
.pipeThrough(splitStream(...)) // string → SSE事件块
.pipeThrough(splitPart(...)) // SSE事件块 → SSEOutput
) as XReadableStream<Output>;
// 给流对象附加 AsyncIterator，让其支持 for await...of
stream[Symbol.asyncIterator] = async function*() {
const reader = this.getReader();
while (true) {
const { done, value } = await reader.read();
if (done) break;
if (!value) continue;
yield value;
}
};
return stream;
}
自定义 TransformStream 的场景：
// 场景：服务端返回 JSON Lines 格式而非标准 SSE
// {"content":"Hello","type":"text"}\n
// {"content":"World","type":"text"}\n
const jsonLinesTransform = new TransformStream<string, MyOutput>({
transform(chunk, controller) {
// 手动处理每行 JSON
chunk.split('\n').filter(Boolean).forEach(line => {
try {
controller.enqueue(JSON.parse(line));
} catch {}
});
}
});
const stream = XStream({
readableStream: response.body,
transformStream: jsonLinesTransform, // 替换默认的 SSE 解析管道
});
7. 与 上文实现 的对比分析
7.1 架构对比
自己实现 (命令式) XStream (声明式/管道式)
───────────────────────── ──────────────────────────
connectSSE() XStream()
│ │
├─ EventSource (高层API) ├─ ReadableStream (底层)
│ └─ 自动处理SSE协议 │
│ │ .pipeThrough(decoderStream)
└─ connectSSEWithFetch() │ .pipeThrough(splitStream)
│ │ .pipeThrough(splitPart)
├─ fetch() │
├─ createSSELineProcessor() │ for await...of
│ ├─ buffer (闭包) │
│ ├─ decoder (闭包) └─ 返回流对象（延迟消费）
│ ├─ processChunk()
│ └─ flush()
└─ readStream()
7.2 数据处理对比
自己实现（命令式，紧耦合业务逻辑）：
// 解析和业务回调混在一起
const processChunk = (chunk: Uint8Array) => {
buffer += decoder.decode(chunk, { stream: true });
const lines = buffer.split('\n');
buffer = lines.pop() || '';
for (const line of lines) {
if (!line.trim()) continue;
const data = parseSSELine(line);
if (!data) continue;
if (data.done) {
options.onDone(); // ← 业务逻辑耦合在解析中
return false;
}
options.onChunk(data.content); // ← 业务回调耦合在解析中
}
};
XStream（声明式，关注点分离）：
// splitStream 只关心按 \n\n 分割
new TransformStream({ transform(chunk) { /* 只分割 */ } })
// splitPart 只关心解析键值对
new TransformStream({ transform(chunk) { /* 只解析 */ } })
// 业务逻辑完全在消费侧
for await (const event of stream) {
if (event.data) {
const data = JSON.parse(event.data);
if (data.done) onDone();
else onChunk(data.content);
}
}
7.3 buffer 处理对比
// sse.ts：手动管理 buffer（需要理解流式处理细节）
const createSSELineProcessor = () => {
let buffer = ''; // 跨 chunk 维持
const decoder = new TextDecoder(); // 手动创建
return {
processChunk(chunk: Uint8Array) {
buffer += decoder.decode(chunk, { stream: true }); // 手动解码
const lines = buffer.split('\n');
buffer = lines.pop() || ''; // 手动保留最后行
// ...
},
flush() {
decoder.decode(); // 手动 flush
}
};
};
// XStream：TransformStream 框架自动处理 flush
new TransformStream({
transform(chunk, controller) {
buffer += chunk;
// ...buffer 处理...
},
flush(controller) {
// 框架在流结束时自动调用这个方法，不需要手动触发
if (buffer) controller.enqueue(buffer);
}
});
7.4 主要差异表
| 维度 |
自己实现 |
XStream |
| 设计思想 |
命令式，手动控制流程 |
声明式，管道组合 |
| 关注点分离 |
解析与业务耦合 |
每层只做一件事 |
| 可组合性 |
低，硬编码解析逻辑 |
高，可替换任意管道环节 |
| 可测试性 |
需模拟整个连接 |
每个 TransformStream 可单独测试 |
| 背压处理 |
手动/无 |
pipeThrough 自动处理 |
| 重连逻辑 |
内置 |
不含（需外部处理） |
| 格式灵活性 |
固定 SSE 格式 |
支持自定义 transformStream |
| 学习曲线 |
低，容易理解 |
较高，需理解流 API |
| flush 触发 |
手动调用 |
框架自动调用 |
7.5 联系：相同的核心解题思路
两者都解决了同一个核心问题：TCP 数据包不按照 SSE 事件边界切割。
同一 buffer 策略：
sse.ts: XStream:
───────────────── ─────────────────────────────
buffer = '' let buffer = '' (在 splitStream 中)
buffer += decode(chunk) buffer += streamChunk
lines = buffer.split('\n') parts = buffer.split('\n\n')
buffer = lines.pop() buffer = parts[parts.length - 1]
7.6 结合使用建议
实际上，可以把 XStream 用于 sse.ts 中：
// 用 XStream 替代 createSSELineProcessor
export const connectSSEWithFetch = (message, options) => {
// ... 重连逻辑保留 ...
const connect = async () => {
const response = await fetch(url, { signal: abortController.signal });
// ✅ 用 XStream 处理解析，用 sse.ts 处理重连
const stream = XStream({ readableStream: response.body });
for await (const event of stream) {
if (!event.data) continue;
const data = JSON.parse(event.data);
if (data.done) {
options.onDone();
closeConnection();
break;
}
options.onChunk(data.content);
}
};
};
8. 总结
8.1 流 API 知识体系
浏览器 Streams API 体系：
ReadableStream 可读流（数据来源）
├─ .getReader() 获取 reader（锁定流）
│ ├─ .read() 逐块读取
│ └─ .releaseLock() 释放锁
├─ .pipeThrough(ts) 接入 TransformStream
└─ .pipeTo(ws) 接入 WritableStream
TransformStream<I, O> 转换流（数据中转站）
├─ .readable 可读端（输出侧）
├─ .writable 可写端（输入侧）
└─ new TransformStream({
transform(chunk, ctrl) { ctrl.enqueue(data) }
flush(ctrl) { /* EOF 处理 */ }
})
WritableStream 可写流（数据消费者）
管道链：
ReadableStream
.pipeThrough(TransformStream1) → ReadableStream
.pipeThrough(TransformStream2) → ReadableStream
.pipeTo(WritableStream) → Promise<void>
8.2 XStream 管道链的优雅之处
// 每一层只做一件事，清晰可维护
readableStream // Uint8Array（二进制）
.pipeThrough(createDecoderStream()) // → string（解码）
.pipeThrough(splitStream('\n\n')) // → string（按事件分割）
.pipeThrough(splitPart('\n', ':')) // → SSEOutput（解析键值）
// 想换成 JSON Lines 格式？只替换最后两层
.pipeThrough(jsonLinesTransform)
// 想加压缩？在最前面加一层
readableStream
.pipeThrough(decompressionStream) // → Uint8Array（解压）
.pipeThrough(createDecoderStream())
// ...
8.3 快速参考
// ① 创建最简单的 XStream 消费
const stream = XStream({ readableStream: response.body });
for await (const event of stream) {
console.log(event); // { event: 'delta', data: '...' }
}
// ② 自定义格式
const stream = XStream({
readableStream: response.body,
transformStream: myCustomTransform,
});
// ③ 自定义分隔符
const stream = XStream({
readableStream: response.body,
streamSeparator: '\r\n\r\n', // Windows 换行
kvSeparator: '=', // key=value 格式
});
// ④ 使用 splitStream/splitPart 单独测试
const splitTransform = splitStream('\n\n');
const writer = splitTransform.writable.getWriter();
const reader = splitTransform.readable.getReader();
writer.write("event: test\n\n");
const { value } = await reader.read(); // "event: test"
9. 感想
其实一步步做下来会发现，sse看起来只是一个协议，但是涉及到的前端知识挺多的，从浏览器自己封装的EventSource接口再到自己基于fetch封装更加灵活的接口，再到xStream的源码实现，其中有许多流的相关概念，同时还涉及到了闭包、断连重试等JS基础，总而言之，好好看看还是有不少收获的。
登录
