技术小册

主题

Java 流式输出与 SSE ​

一句话：SSE 是在普通 HTTP 响应上持续写 text/event-stream 帧的单向推送；它把 LLM 的逐 token 输出变成首字可见，代价是连接要长期保持、代理要关缓冲、断线要自己设计续传。

适用 Spring Boot 3.x / JDK 17+。多模型父子流、取消链和慢客户端编排见 15-SSE父子流与流编排；本文只讲 HTTP 传输与单条流的 Java 落地。

为什么需要 ​

LLM 生成是逐 token 的。非流式接口要等完整答案生成完才返回，首字延迟（TTFT）约等于全量生成时间；流式接口在第一个 token 到达时就写响应，用户感知的等待从「十秒」降到「几百毫秒」。

SSE 比 WebSocket 更适合 token 流：它只有服务端到客户端的单向通道，走普通 HTTP 请求，能复用 Cookie/Header 鉴权、网关、访问日志和压缩策略；浏览器原生 EventSource 自带重连和 Last-Event-ID。

核心机制 ​

协议帧 ​

响应头至少要有 Content-Type: text/event-stream; charset=UTF-8、Cache-Control: no-cache 和 X-Accel-Buffering: no。每个事件由若干字段行组成，空行表示一帧结束：

text
event: token
id: 42
data: {"delta":"你"}

: ping

event: done
data: [DONE]
  • data: 是负载，多行 data: 会用换行拼接；JSON 里不能直接出现未转义的换行。
  • event: 是事件类型，客户端按类型分发；不写就是默认 message。
  • id: 会写入 Last-Event-ID，断线重连时浏览器把它带回服务端。
  • retry: 建议重连间隔，单位毫秒；不写时浏览器通常按约 3 秒重连，规范允许实现调整。
  • 以 : 开头的是注释，常用来发心跳，客户端不会把它当事件。

用 curl -N 关闭缓冲，直接看原始帧：

bash
curl -N -H 'Accept: text/event-stream' http://localhost:8080/ai/stream
# : ping
# event: token
# id: 1
# data: {"delta":"你"}
# 空行后是下一帧；最后 event: done / data: [DONE]

与 WebSocket 的边界 ​

SSE 是普通 HTTP 响应上的单向下行，能复用 Cookie 鉴权、网关和访问日志，EventSource 自带 Last-Event-ID 重连；WebSocket 要 Upgrade、全双工、自己做心跳重连。token 流用不上双向能力时，SSE 运维成本更低；需要客户端频繁发消息、二进制流或实时协作时才换 WebSocket。

最小上手 ​

Spring MVC：SseEmitter ​

SseEmitter 基于 Servlet 异步请求。Controller 返回 emitter 后，请求线程被释放，后续 send 在业务线程池里执行；不能在请求线程里循环生成 token，否则会占满 Tomcat 工作线程。

java
@RestController
public class SseController {
    private final ExecutorService pool = Executors.newFixedThreadPool(8);
    @GetMapping(value = "/ai/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter stream() {
        SseEmitter emitter = new SseEmitter(300_000L); // 5 分钟超时
        emitter.onTimeout(() -> emitter.complete()); emitter.onError(e -> emitter.completeWithError(e));
        emitter.onCompletion(() -> { /* 取消上游 LLM 请求 */ });
        pool.execute(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    emitter.send(SseEmitter.event().id("" + i)
                            .event("token").data("token-" + i));
                    Thread.sleep(200); // 模拟逐 token；真实代码用 SDK 回调
                }
                emitter.send(SseEmitter.event().event("done").data("[DONE]"));
                emitter.complete();
            } catch (Exception e) {
                emitter.completeWithError(e); // 客户端断开时 send 会抛 IOException
            }
        });
        return emitter;
    }
}

默认 SseEmitter() 不显式设置 timeout，实际超时由 Spring MVC 的异步配置和 Servlet 容器决定（常见容器默认约 30 秒，具体以官方文档和容器配置为准）；长连接要么显式设更大的值，要么设 0（多数容器解释为不超时）并靠心跳保活。心跳用 SseEmitter.event().comment("ping") 定时发送，并在完成/异常时取消定时任务。同一 emitter 的 send 不是线程安全的：心跳和 token 发送必须串行化（单线程发送，或共用一把锁/一个队列），否则帧会交错甚至抛异常；完整取消链见 15。

WebFlux：Flux<ServerSentEvent> ​

WebFlux 的返回值本身就是流，背压由 Reactor 和 HTTP 层传递；不要在里面 block()。

java
@GetMapping(value = "/ai/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<ServerSentEvent<String>> stream() {
    return Flux.interval(Duration.ZERO, Duration.ofMillis(200)).take(10)
            .map(i -> ServerSentEvent.<String>builder()
                    .id(String.valueOf(i)).event("token").data("token-" + i).build())
            .timeout(Duration.ofSeconds(10))
            .onErrorResume(e -> Flux.just(ServerSentEvent.<String>builder()
                    .event("error").data("stream failed").build()));
}

timeout 约束的是相邻事件/完成信号之间的间隔；超时后 HTTP 200 已经发出，只能在流内发 event: error，不能再改成 500。

消费端 ​

WebClient 直接把 SSE 解码成对象：

java
WebClient client = WebClient.create("http://localhost:8080");
client.get().uri("/ai/stream").accept(MediaType.TEXT_EVENT_STREAM).retrieve()
        .bodyToFlux(new ParameterizedTypeReference<ServerSentEvent<String>>() {})
        .doOnNext(e -> System.out.println(e.event() + " " + e.data())).blockLast(); // 仅演示

JDK HttpClient 没有 SSE 解析器，但可以按行流式读取：

java
HttpClient client = HttpClient.newHttpClient();
HttpRequest req = HttpRequest.newBuilder(URI.create("http://localhost:8080/ai/stream"))
        .header("Accept", "text/event-stream").build();
HttpResponse<Stream<String>> resp = client.send(req, HttpResponse.BodyHandlers.ofLines());
resp.body().filter(l -> l.startsWith("data:")).forEach(l -> System.out.println(l.substring(5).trim()));

ofLines() 返回的 Stream 要关闭（try-with-resources），否则连接不会释放；空行和 event:/id: 需要自己解析。

生产动作 ​

  • 代理与网关：Nginx 关缓冲 proxy_buffering off; proxy_cache off; proxy_http_version 1.1; proxy_set_header Connection ""; proxy_read_timeout 300s;，并加 X-Accel-Buffering: no；云 LB/CDN 空闲超时常见 60/100 秒，以厂商文档为准，心跳要小于最小空闲超时（如 15 秒心跳配 60 秒超时）。proxy_read_timeout 默认 60 秒，必须大于心跳间隔。
  • 断开要取消上游：MVC 监听 onCompletion/onError/send 的 IOException，WebFlux 用 doOnCancel 或 Flux.create 的 onDispose；拿到 LLM SDK 的取消句柄并关闭流，否则客户端走了模型还在计费。
  • 错误在流内下发：首字节发出后 HTTP 状态码已经固定，业务错误用 event: error + data 表达，再关闭连接；不要在流开始后抛异常指望框架改状态码。
  • 多实例：独立请求不需要会话粘性；要支持 Last-Event-ID 续传，就需要共享的事件存储和统一的序号，日志用 traceId 串起网关、应用和模型调用。
  • MVC 线程隔离：生成任务放在有界线程池，池满时拒绝或排队，不要用无界队列；WebFlux 则避免在事件循环里做阻塞调用。

边界 ​

  • SSE 是单向的：客户端要发消息（如打断生成）另开 POST；不能在同一条 SSE 连接上双向通信。浏览器 EventSource 只支持 GET、不能自定义 Authorization 头；需要 Header 鉴权时用 fetch + ReadableStream 自行解析 SSE，或用 Cookie/短期签名 URL。
  • HTTP/1.1 浏览器连接数有限：同域通常最多 6 个并发连接，每个 SSE 占一个，开多个标签页会互相阻塞；HTTP/2 多路复用可以缓解，但代理和浏览器仍可能有上限。
  • 移动端与重连：后台挂起、切换网络都会断连；EventSource 自动重连不保证立刻发生，Last-Event-ID 只是提示，服务端不持久化事件就只能从头开始，重复事件要在客户端或业务层去重。
  • 慢客户端不能靠无界缓冲：SSE 没有浏览器侧的 request(n)，发送队列必须有界、有超时和取消，具体做法见 15。
  • 该用 WebSocket 的场景：双向频繁交互、二进制流、实时语音/协作编辑；只是把模型 token 推给浏览器，SSE 更简单。

讲清楚 ​

LLM 逐 token 生成，非流式会把首字延迟拖成全文生成时间，所以要把响应边生成边写。SSE 在普通 HTTP 响应上写 text/event-stream，用 data/event/id/retry 字段和空行分帧，用 : 注释发心跳，断线后浏览器带 Last-Event-ID 重连；它单向、复用现有鉴权和网关，比 WebSocket 更适合 token 流。

Java 侧 MVC 用 SseEmitter 把生成任务丢到有界线程池，WebFlux 用 Flux<ServerSentEvent> 让背压贯穿链路，消费端用 WebClient 或 JDK HttpClient 按行读。生产上四个坑：代理缓冲关掉、空闲超时大于心跳、客户端断开要取消模型调用、流开始后错误只能以事件下发。多模型合并、父子取消链和慢客户端队列见 15。

适用 Spring Boot 3.x / JDK 17+ | 初稿 2026-10

把一门技术讲透到既懂也会用

技术小册 · 持续更新