高级用法与面试题

TransformStream、流中断恢复、面试高频问题

8. 高级用法

8.1 TransformStream 管道

function createSSEParser() {
    let buffer = '';

    return new TransformStream({
        transform(chunk, controller) {
            buffer += chunk;
            const lines = buffer.split('\n');
            buffer = lines.pop() || '';

            for (const line of lines) {
                if (line.startsWith('data: ')) {
                    const data = line.slice(6).trim();
                    if (data && data !== '[DONE]') {
                        try {
                            const json = JSON.parse(data);
                            const content = json.choices?.[0]?.delta?.content;
                            if (content) controller.enqueue(content);
                        } catch {}
                    }
                }
            }
        }
    });
}

// 使用管道
const response = await fetch('/api/chat', { method: 'POST', body: '...' });

const textStream = response.body
    .pipeThrough(new TextDecoderStream())
    .pipeThrough(createSSEParser());

for await (const token of textStream) {
    console.log(token);
}

8.2 流中断恢复

async function streamWithResume(messages, onToken, lastContent = '') {
    const resumeMessages = [...messages];

    if (lastContent) {
        resumeMessages.push({ role: 'assistant', content: lastContent });
        resumeMessages.push({ role: 'user', content: '请继续' });
    }

    try {
        for await (const token of streamChat(resumeMessages)) {
            lastContent += token;
            onToken(token);
        }
    } catch (error) {
        if (confirm('连接中断，是否继续?')) {
            return streamWithResume(messages, onToken, lastContent);
        }
    }
}

---

9. 面试高频问题

Q1: SSE 的本质是什么？为什么说 POST 也能实现 SSE？

核心答案：SSE 的本质是服务端的响应格式，而不是客户端的请求方式。

从服务端视角看，SSE 只需要两个要素：

1. 响应头设置 Content-Type: text/event-stream
2. 按照 data: ...\n\n 格式输出流式数据

与请求方法无关：

• 客户端用 GET、POST 甚至 PUT 都可以
• 服务端只需调整"入口"（参数获取方式）
• 推流逻辑（"出口"）完全一致

为什么会有误解？ 因为浏览器原生的 EventSource API 被设计成只能发起 GET 请求，无法自定义 Headers。这是 API 的限制，不是 SSE 协议的限制。

Q2: SSE 和 WebSocket 的区别?

维度	SSE	WebSocket
方向	单向 (服务端→客户端)	双向
协议	HTTP	独立协议 (ws://)
自动重连	✅ 内置	❌ 需手动
二进制	❌	✅
适用场景	LLM、通知	聊天、协作

Q3: 为什么 LLM 用 Fetch Streaming 而不是 EventSource?

1. POST 支持：EventSource 只支持 GET，无法 POST 长文本 prompt
2. 自定义 Headers：EventSource 无法添加 Authorization 等认证信息
3. 取消控制：Fetch Streaming 支持 AbortController 精确取消
4. 灵活性：更灵活的流处理控制和错误处理

Q4: TextDecoder 的 stream: true 有什么作用?

处理多字节字符 (如 UTF-8 中文) 被拆分到不同 chunk 的情况。

// 中文"中"的 UTF-8 编码是 [228, 184, 173]
// 如果被拆分成两个 chunk：[228, 184] 和 [173]

// ❌ 错误：每次独立解码会产生乱码
decoder.decode([228, 184]); // "�"

// ✅ 正确：stream: true 保留不完整序列
decoder.decode([228, 184], { stream: true }); // "" (等待)
decoder.decode([173], { stream: true }); // "中"

Q5: 如何实现"停止生成"功能?

1. 创建 AbortController
2. 将 signal 传给 fetch
3. 用户点击停止时调用 controller.abort()
4. 捕获 AbortError，不作为错误处理

const controller = new AbortController();
fetch('/api/chat', { signal: controller.signal });

// 用户点击停止按钮
stopButton.onclick = () => controller.abort();

Q6: ReadableStream 的背压机制是什么?

当消费者 (reader.read()) 处理速度慢于生产者时，ReadableStream 自动暂停数据流入，等待消费者准备好，防止内存无限增长。

示例场景：

• 服务端以 1MB/s 推送数据
• 客户端只能以 100KB/s 处理
• 背压机制会让服务端放慢速度，避免积压

Q7: POST SSE 是否需要手动解析？有哪些现成的库？

不需要手动解析，有多种成熟的库可选：

使用场景	推荐库	特点
通用 POST SSE	@microsoft/fetch-event-source	轻量、支持重连、POST/Headers
LLM 全栈应用	Vercel AI SDK	React Hooks、状态管理、多模型
OpenAI 专用	OpenAI SDK	官方支持、async iterator
纯解析器	eventsource-parser	极致轻量（2KB）

最简单的方式（Vercel AI SDK）：

import { useChat } from 'ai/react';

function Chat() {
    const { messages, input, handleSubmit } = useChat();
    // ✅ 自动处理 POST、SSE 解析、状态管理
    return <form onSubmit={handleSubmit}>...</form>;
}

手动解析虽然灵活，但需要处理粘包/分包、错误重连等复杂逻辑，生产环境建议使用成熟库。

Q8: 如何在 Nginx 后正确代理 SSE?

location /api/chat/stream {
    proxy_pass http://backend;
    proxy_http_version 1.1;
    proxy_set_header Connection "";

    # 关键：关闭缓冲
    proxy_buffering off;
    proxy_cache off;

    # SSE 特殊处理
    chunked_transfer_encoding on;
}