前端 Streaming 架构实战：从 SSE 到 Web Streams API 的全链路方案

1
// 最简单的 ReadableStream
2
const stream = new ReadableStream({
3
  start(controller) {
4
    controller.enqueue('Hello');
5
    controller.enqueue(' ');
6
    controller.enqueue('Streams');
7
    controller.close();
8
  }
9
});
10

11
// 消费它
12
const reader = stream.getReader();
13
while (true) {
14
  const { done, value } = await reader.read();
15
  if (done) break;
16
  console.log(value); // 'Hello', ' ', 'Streams'
17
}

1
// 一个将文本按行分割的 TransformStream
2
function createLineSplitter() {
3
  let buffer = '';
4
  return new TransformStream({
5
    transform(chunk, controller) {
6
      buffer += chunk;
7
      const lines = buffer.split('\n');
8
      // 最后一个可能是不完整的行，留在 buffer 里
9
      buffer = lines.pop();
10
      for (const line of lines) {
11
        if (line.trim()) {
12
          controller.enqueue(line);
13
        }
14
      }
15
    },
16
    flush(controller) {
17
      // 流结束时，把 buffer 里剩余的内容也推出去
18
      if (buffer.trim()) {
19
        controller.enqueue(buffer);
20
      }
21
    }
22
  });
23
}

1
const response = await fetch('/api/chat', {
2
  method: 'POST',
3
  headers: { 'Content-Type': 'application/json' },
4
  body: JSON.stringify({ prompt: '解释量子计算' })
5
});
6

7
const reader = response.body
8
  .pipeThrough(new TextDecoderStream())
9
  .getReader();
10

11
while (true) {
12
  const { done, value } = await reader.read();
13
  if (done) break;
14
  appendToUI(value);
15
}

1
function createSSEParser() {
2
  let buffer = '';
3
  return new TransformStream({
4
    transform(chunk, controller) {
5
      buffer += chunk;
6
      const events = buffer.split('\n\n');
7
      buffer = events.pop(); // 最后一个可能不完整
8

9
      for (const event of events) {
10
        const lines = event.split('\n');
11
        let data = '';
12
        let eventType = 'message';
13

14
        for (const line of lines) {
15
          if (line.startsWith('data: ')) {
16
            data += line.slice(6);
17
          } else if (line.startsWith('event: ')) {
18
            eventType = line.slice(7);
19
          }
20
        }
21

22
        if (data === '[DONE]') {
23
          controller.terminate();
24
          return;
25
        }
26

27
        if (data) {
28
          try {
29
            controller.enqueue({
30
              type: eventType,
31
              data: JSON.parse(data)
32
            });
33
          } catch {
34
            // 非 JSON 数据，原样传递
35
            controller.enqueue({ type: eventType, data });
36
          }
37
        }
38
      }
39
    },
40
    flush(controller) {
41
      // 处理 buffer 中剩余的数据
42
      if (buffer.trim()) {
43
        const lines = buffer.split('\n');
44
        let data = '';
45
        for (const line of lines) {
46
          if (line.startsWith('data: ')) data += line.slice(6);
47
        }
48
        if (data && data !== '[DONE]') {
49
          try {
50
            controller.enqueue({ type: 'message', data: JSON.parse(data) });
51
          } catch {
52
            controller.enqueue({ type: 'message', data });
53
          }
54
        }
55
      }
56
    }
57
  });
58
}

1
class StreamingChat {
2
  constructor(apiUrl, options = {}) {
3
    this.apiUrl = apiUrl;
4
    this.maxRetries = options.maxRetries ?? 3;
5
    this.timeout = options.timeout ?? 30000;
6
    this.controller = null;
7
  }
8

9
  async *stream(messages) {
10
    let retries = 0;
11

12
    while (retries <= this.maxRetries) {
13
      this.controller = new AbortController();
14
      const timeoutId = setTimeout(
15
        () => this.controller.abort('timeout'),
16
        this.timeout
17
      );
18

19
      try {
20
        const response = await fetch(this.apiUrl, {
21
          method: 'POST',
22
          headers: { 'Content-Type': 'application/json' },
23
          body: JSON.stringify({ messages, stream: true }),
24
          signal: this.controller.signal
25
        });
26

27
        if (!response.ok) {
28
          throw new Error(`HTTP ${response.status}: ${response.statusText}`);
29
        }
30

31
        clearTimeout(timeoutId);
32

33
        const eventStream = response.body
34
          .pipeThrough(new TextDecoderStream())
35
          .pipeThrough(createSSEParser());
36

37
        const reader = eventStream.getReader();
38

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

43
          const content = value.data?.choices?.[0]?.delta?.content;
44
          if (content) {
45
            yield content;
46
          }
47
        }
48

49
        return; // 成功完成，退出重试循环
50

51
      } catch (error) {
52
        clearTimeout(timeoutId);
53

54
        if (error.name === 'AbortError') {
55
          if (retries < this.maxRetries) {
56
            retries++;
57
            console.warn(`Stream timeout, retry ${retries}/${this.maxRetries}`);
58
            await this.delay(1000 * retries); // 指数退避
59
            continue;
60
          }
61
        }
62

63
        throw error;
64
      }
65
    }
66
  }
67

68
  cancel() {
69
    this.controller?.abort('user_cancel');
70
  }
71

72
  delay(ms) {
73
    return new Promise(resolve => setTimeout(resolve, ms));
74
  }
75
}

1
class StreamingMarkdownRenderer {
2
  constructor(container) {
3
    this.container = container;
4
    this.fullText = '';
5
    this.renderTimer = null;
6
    this.parser = null; // 你喜欢的 markdown 解析器
7
  }
8

9
  append(token) {
10
    this.fullText += token;
11
    // 节流渲染：不是每个 token 都触发 DOM 更新
12
    if (!this.renderTimer) {
13
      this.renderTimer = requestAnimationFrame(() => {
14
        this.render();
15
        this.renderTimer = null;
16
      });
17
    }
18
  }
19

20
  render() {
21
    // 策略：完整重新解析，但用 DOM diff 减少重绘
22
    const html = this.parseMarkdown(this.fullText);
23

24
    // 不要直接 innerHTML，用增量更新
25
    this.patchDOM(html);
26

27
    // 自动滚动到底部
28
    this.container.scrollTop = this.container.scrollHeight;
29
  }
30

31
  parseMarkdown(text) {
32
    // 对于未闭合的语法，追加闭合标记
33
    let safeText = text;
34

35
    // 处理未闭合的代码块
36
    const codeBlockCount = (safeText.match(/```/g) || []).length;
37
    if (codeBlockCount % 2 !== 0) {
38
      safeText += '\n```';
39
    }
40

41
    // 处理未闭合的加粗/斜体
42
    const boldCount = (safeText.match(/\*\*/g) || []).length;
43
    if (boldCount % 2 !== 0) {
44
      safeText += '**';
45
    }
46

47
    // 用你喜欢的库解析（marked, markdown-it, etc.）
48
    return marked.parse(safeText);
49
  }
50

51
  patchDOM(newHTML) {
52
    // 使用 morphdom 或类似库做 DOM diff
53
    morphdom(this.container, `<div>${newHTML}</div>`, {
54
      childrenOnly: true
55
    });
56
  }
57
}

1
async function streamUpload(file, uploadUrl) {
2
  const CHUNK_SIZE = 1024 * 1024; // 1MB per chunk
3
  let offset = 0;
4

5
  const fileStream = new ReadableStream({
6
    async pull(controller) {
7
      if (offset >= file.size) {
8
        controller.close();
9
        return;
10
      }
11

12
      const slice = file.slice(offset, offset + CHUNK_SIZE);
13
      const buffer = await slice.arrayBuffer();
14
      controller.enqueue(new Uint8Array(buffer));
15
      offset += CHUNK_SIZE;
16
    }
17
  });
18

19
  // 上传 transform：每个 chunk 变成一个 HTTP 请求
20
  const uploadTransform = new TransformStream({
21
    async transform(chunk, controller) {
22
      const formData = new FormData();
23
      formData.append('chunk', new Blob([chunk]));
24
      formData.append('offset', offset - CHUNK_SIZE);
25

26
      const res = await fetch(uploadUrl, {
27
        method: 'POST',
28
        body: formData
29
      });
30

31
      if (!res.ok) throw new Error(`Upload failed: ${res.status}`);
32

33
      controller.enqueue({
34
        uploaded: offset,
35
        total: file.size,
36
        progress: Math.min(100, (offset / file.size * 100).toFixed(1))
37
      });
38
    }
39
  });
40

41
  // 进度消费
42
  await fileStream
43
    .pipeThrough(uploadTransform)
44
    .pipeTo(new WritableStream({
45
      write({ progress }) {
46
        progressBar.style.width = `${progress}%`;
47
      }
48
    }));
49
}

1
class ResumableStream {
2
  constructor(url, options = {}) {
3
    this.url = url;
4
    this.lastEventId = options.lastEventId || null;
5
    this.collectedContent = '';
6
    this.onToken = options.onToken || (() => {});
7
    this.onError = options.onError || (() => {});
8
    this.onComplete = options.onComplete || (() => {});
9
  }
10

11
  async start() {
12
    const headers = { 'Content-Type': 'application/json' };
13

14
    // 如果有上次的位置，带上断点信息
15
    if (this.lastEventId) {
16
      headers['Last-Event-ID'] = this.lastEventId;
17
    }
18

19
    try {
20
      const response = await fetch(this.url, {
21
        method: 'POST',
22
        headers,
23
        body: JSON.stringify({
24
          resume_from: this.lastEventId,
25
          partial_content: this.collectedContent
26
        })
27
      });
28

29
      const reader = response.body
30
        .pipeThrough(new TextDecoderStream())
31
        .pipeThrough(createSSEParser())
32
        .getReader();
33

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

38
        if (value.data?.id) {
39
          this.lastEventId = value.data.id;
40
        }
41

42
        const content = value.data?.choices?.[0]?.delta?.content;
43
        if (content) {
44
          this.collectedContent += content;
45
          this.onToken(content);
46
        }
47
      }
48

49
      this.onComplete(this.collectedContent);
50

51
    } catch (error) {
52
      this.onError(error);
53
      // 自动重试
54
      if (this.lastEventId) {
55
        console.log('Stream interrupted, resuming...');
56
        await new Promise(r => setTimeout(r, 2000));
57
        return this.start(); // 递归重试
58
      }
59
      throw error;
60
    }
61
  }
62
}

1
async function chatWithFallback(messages, onToken) {
2
  try {
3
    // 先尝试流式
4
    const response = await fetch('/api/chat', {
5
      method: 'POST',
6
      headers: { 'Content-Type': 'application/json' },
7
      body: JSON.stringify({ messages, stream: true })
8
    });
9

10
    if (!response.body) {
11
      throw new Error('ReadableStream not supported');
12
    }
13

14
    // 流式消费...
15
    const reader = response.body.pipeThrough(new TextDecoderStream()).getReader();
16
    let fullText = '';
17
    while (true) {
18
      const { done, value } = await reader.read();
19
      if (done) break;
20
      fullText += value;
21
      onToken(value);
22
    }
23
    return fullText;
24

25
  } catch (streamError) {
26
    console.warn('Streaming failed, falling back to regular request', streamError);
27

28
    // 降级到普通请求
29
    const response = await fetch('/api/chat', {
30
      method: 'POST',
31
      headers: { 'Content-Type': 'application/json' },
32
      body: JSON.stringify({ messages, stream: false })
33
    });
34

35
    const data = await response.json();
36
    const content = data.choices[0].message.content;
37
    onToken(content); // 一次性输出
38
    return content;
39
  }
40
}