流式输出的收益只有一条：首 token 时间。代价却是一长串——发出前做不了后处理，错误只能在流中途发现，连接生命周期从毫秒级拉长到几十秒；nginx、连接池、客户端取消三处任一处没配好，用户看到的就是"转圈很久，然后一次性吐出全文"。

---

1. 收益与代价的对照

维度	非流式	流式
感知延迟	等于完整生成时间	等于首 token 时间
总耗时	基本相同	基本相同
后处理	可以做过滤、校验、改写	已发出的内容改不了
错误发现	结束才知道	中途才知道，已发出的收不回
连接占用	短	长（几十秒到几分钟）
网关要求	无特殊要求	必须关闭缓冲、放宽读超时
可测试性	断言完整输出	断言增量与拼接结果

流式是感知层的优化，不是正确性的优化：需要输出前校验、脱敏、解析 JSON 的场景，要么放弃流式，要么把校验下沉到协议层。

---

2. SSE 协议要点

SSE 是一段 text/event-stream 纯文本，消息之间空行分隔，每行形如 字段: 值：

字段	作用	工程注意
data:	载荷，可多行	含换行须拆成多个 data: 行
event:	事件类型，默认 message	区分 content / done / error
id:	事件编号	重连时回传 Last-Event-ID
: 开头	注释行	常用作心跳，防止空闲断连

响应头三件套缺一不可：

Content-Type: text/event-stream; charset=utf-8
Cache-Control: no-cache
Connection: keep-alive

---

3. 两种实现

3.1 SseEmitter

适合 Servlet 栈、逻辑已是阻塞式调用的存量项目：

@RestController
public class ChatStreamController {

    private final ChatService chatService;
    public ChatStreamController(ChatService chatService) { this.chatService = chatService; }

    private static void send(SseEmitter em, String event, String payload) {
        try { em.send(SseEmitter.event().name(event).data(payload)); } catch (IOException ignored) { }
    }

    @GetMapping(value = "/api/chat/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter stream(@RequestParam String q, @RequestParam String traceId) {
        SseEmitter emitter = new SseEmitter(120_000L);   // 超时值必须显式给
        AtomicBoolean closed = new AtomicBoolean(false);

        emitter.onCompletion(() -> closed.set(true));
        emitter.onTimeout(() -> { closed.set(true); emitter.complete(); });
        emitter.onError(e -> closed.set(true));

        chatService.streamAnswer(q, traceId).subscribe(
            chunk -> { if (!closed.get()) send(emitter, "content", chunk); },   // 已断开则不再写
            err -> { if (!closed.get()) { send(emitter, "error", err.getMessage()); emitter.complete(); } },
            () -> { if (!closed.get()) { send(emitter, "done", "[DONE]"); emitter.complete(); } });
        return emitter;
    }
}

3.2 Flux

适合 WebFlux 栈，取消语义天然贯通——客户端断开时 Flux 被取消，doFinally 必定执行：

@GetMapping(value = "/api/chat/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<ServerSentEvent<String>> stream(@RequestParam String q,
                                            @RequestParam String traceId) {
    return chatService.streamAnswer(q, traceId)
            .map(token -> ServerSentEvent.<String>builder(token).event("content").build())
            .concatWith(Flux.just(ServerSentEvent.<String>builder("[DONE]").event("done").build()))
            .onErrorResume(err -> Flux.just(
                    ServerSentEvent.<String>builder("生成失败，请重试").event("error").build()))
            .doFinally(signal -> log.info("stream_end trace={} signal={}", traceId, signal));
}

对比项	SseEmitter	Flux<ServerSentEvent>
技术栈	Spring MVC	WebFlux
线程模型	需自行安排线程	事件循环，不阻塞
断开感知	靠回调，有延迟	Flux 取消，及时
背压	无	有
学习成本	低	中（要管 Subscription）
存量项目	改造量小	需整层响应式

---

4. nginx 侧必须改的三项

项	不配的症状	处理
proxy_buffering	内容攒在 nginx 缓冲里，结束才吐	proxy_buffering off;
X-Accel-Buffering	应用以为放开了，nginx 仍缓冲	响应头 no
proxy_read_timeout	默认值到点 504，用户看到半截	调到比最长生成更长

location /api/chat/stream {
    proxy_pass http://app_backend;
    proxy_http_version 1.1;
    proxy_set_header Connection '';
    proxy_buffering off;
    proxy_cache off;
    chunked_transfer_encoding on;
    proxy_read_timeout 300s;
    proxy_send_timeout 300s;
}

// 应用侧双保险：既是给 nginx 的信号，也方便排查是否被代理缓冲
response.setHeader("X-Accel-Buffering", "no");
response.setHeader("Cache-Control", "no-cache");

前面还有 CDN 或二层代理时同样要关缓冲，否则只改 nginx 仍是"一次性吐出"。

---

5. 断线与重连的语义

SSE 自带重连：浏览器在连接断开后带着 Last-Event-ID 重新发起同一个 GET。

场景	能否续传	说明
客户端抖动，服务端连接还在	一般不能	新请求即新一轮生成
服务端持久化了已发分片	可以	事件表 + 按 id 回放
生成结束后断开	不需要	客户端已有全文
多实例且无共享存储	不能	重连可能落到另一实例

给每个 content 事件加自增 id；服务端只有"存了事件表"时才有资格承诺续传：

long seq = 0;
for (String token : tokens) {
    yield ServerSentEvent.<String>builder(token).event("content").id(String.valueOf(++seq)).build();
}
// 事件 id 只是"可回放"的前提条件，落地还需要服务端存事件表

工程默认选择是"重来"：把这层语义写进前端交互（重新生成按钮），比做一个半可靠的续传省事得多。

---

6. 取消与资源释放

三处回调缺一处就泄漏：onCompletion、onTimeout、onError。要清理的东西固定是四样：

资源	不清理的后果
模型订阅 / 句柄	无人在读，token 仍计费
线程或连接	池被长连接占满
会话缓冲	内存随并发线性增长
埋点计时器	耗时统计被未结束的流污染

客户端断开的检测方式：

emitter.onError(e -> {
    subscription.dispose();                       // 停掉模型侧生成
    metrics.counter("ai.stream.cancelled", "reason", classify(e)).increment();
});
scheduler.scheduleAtFixedRate(() -> {              // 定时心跳：保活 + 尽早发现断开的客户端
    try { emitter.send(SseEmitter.event().comment("ping")); }
    catch (IOException e) { emitter.completeWithError(e); }
}, 15, 15, TimeUnit.SECONDS);

心跳两个作用：别让中间设备按空闲回收连接，以及尽早发现已断开的客户端。

---

7. 流式下的错误表达

流开始后 200 已经发出，后续出错只能靠事件类型表达：

事件	载荷	客户端处理
content	增量文本	追加渲染
done	结束标记	收尾、解除禁用
error	可读的错误摘要	保留已渲染部分 + 提示重试
meta	traceId、model、用量	记入会话

Flux<ServerSentEvent<String>> body = chatService.streamAnswer(q, traceId)
        .map(t -> ServerSentEvent.<String>builder(t).event("content").build())
        .concatWith(Flux.just(ServerSentEvent.<String>builder("[DONE]").event("done").build()))
        .onErrorResume(e -> {
            log.warn("stream_error trace={}", traceId, e);
            return Flux.just(ServerSentEvent.<String>builder("生成中断，请重试").event("error").build());
        });

前端约定：收到 error 不丢弃已渲染内容，只把状态从"生成中"改为"未完成"。

---

8. 首 token 延迟埋点

指标	起点	终点	用途
ai.stream.ttfb	请求进入	首个 content 发出	核心体验指标
ai.stream.duration	请求进入	done 发出	总耗时
ai.stream.gap_max	相邻 token 最大间隔	—	卡顿检测
ai.stream.cancelled	断开计数	—	判断是否前端超时过短
ai.stream.chunks	分片数量	—	偏少说明有缓冲

long start = System.nanoTime();
AtomicBoolean first = new AtomicBoolean(true);
return service.stream(q)
    .doOnNext(t -> { if (first.compareAndSet(true, false))
        timer("ai.stream.ttfb").record(System.nanoTime() - start, NANOSECONDS); })
    .doOnComplete(() -> timer("ai.stream.duration").record(System.nanoTime() - start, NANOSECONDS));
// 直连 curl -N 与经过 nginx 各打一次，差值就是代理链路引入的延迟

TTFB 要按链路分段看：客户端→CDN→nginx→应用→模型。只打应用内一段，会把网络排队算成模型慢。

---

9. 排错

现象	原因	处理
一次性吐出全部内容	nginx 缓冲未关，或上游没发 X-Accel-Buffering	两处都配；用 curl -N 直连验证
60 秒后 504	proxy_read_timeout 默认值偏短	调到比最长生成时间更长的值
并发一高就卡死	长连接占满 Tomcat 线程	换 WebFlux，或调大配额并给流设超时
客户端断开但账单还在涨	未释放订阅	在 onError/onTimeout 内 dispose()
前端收到事件但没有 event 名	未显式设置名字，落到默认 message	服务端显式 event("content")
中文乱码	未声明 charset	produces 里带 charset=utf-8

---

10. 边界

场景	更合适的选择	理由
输出必须是合法 JSON	非流式，或分块 JSON + 完整性校验	半截 JSON 解析不了
需要先脱敏再输出	服务端缓冲后输出	已发出内容无法收回
结果只有一两句	非流式	收益低于复杂度成本
批量离线任务	队列 + 回查	不必占长连接
多端同步同一回答	生成侧落库，端侧拉取	避免重复生成与计费

---

11. 自测

1. 流式优化的是哪个指标，哪个指标其实没变？
2. proxy_buffering off 与 X-Accel-Buffering: no 为什么两处都要配？
3. 客户端断线后，为什么"续传"在多数部署下做不到？
4. onCompletion、onTimeout、onError 三处都清理，能防住哪一类泄漏？
5. 流已开始后出错，为什么不能用 HTTP 500 表达？
6. TTFB 只打应用内一段，会把什么误算成模型慢？