SSE 学习

一、行业现状结论

1. 主流公有大模型（OpenAI / 智谱 GLM / 通义千问 / 文心一言 / Claude）

100% 统一采用text/event-stream（SSE 标准协议）

• 请求头：Content-Type: application/json（POST 传参）
• 响应头：Content-Type: text/event-stream（流式分片返回）
• 分片格式：data: 单条完整JSON\n\n，结尾data: [DONE]\n\n标记结束

2.text/plain仅用于自研内部流式接口

不会出现在标准大模型开放 API，仅两种场景使用：

1. 自己封装中转服务，不遵循 SSE 标准，用换行分割 JSON 片段；
2. 老旧自研 AI 服务，无标准化事件、id、retry 机制。

核心选型差异

二、Demo 1：标准大模型 SSE（text/event-stream）完整示例

以智谱 GLM 流式接口为例，分三层：客户端调用大模型、Java 中转服务 SseEmitter 转发、前端 EventSource 接收

1）curl 请求大模型原生接口（SSE 响应）

curl --location 'https://open.bigmodel.cn/api/paas/v4/chat/completions' \ --header 'Authorization: Bearer 你的API_KEY' \ --header 'Content-Type: application/json' \ --data '{ "model": "glm-5.2", "messages": [{"role": "user", "content": "写一段Java SSE代码"}], "stream": true }'

接口返回流（响应头Content-Type: text/event-stream）

data: {"id":"chatcmpl-xxx","choices":[{"delta":{"content":"好的，下面给你演示"},"index":0}]} data: {"id":"chatcmpl-xxx","choices":[{"delta":{"content":"SseEmitter实现代码："},"index":0}]} data: [DONE]

2）Java WebClient 消费大模型 SSE 流（后端中转）

import org.springframework.web.reactive.function.client.WebClient; import reactor.core.publisher.Flux; public Flux<String> getGlmStream(String prompt) { String reqJson = """ { "model": "glm-5.2", "messages": [{"role":"user","content":"%s"}], "stream": true } """.formatted(prompt); return WebClient.create("https://open.bigmodel.cn") .post() .uri("/api/paas/v4/chat/completions") .header("Authorization", "Bearer " + API_KEY) .contentType(org.springframework.http.MediaType.APPLICATION_JSON) .bodyValue(reqJson) // 声明接收SSE流 .accept(org.springframework.http.MediaType.TEXT_EVENT_STREAM) .retrieve() .bodyToFlux(String.class); }

3）SpringMVC SseEmitter 转发给前端

@GetMapping("/ai/sse/chat") public SseEmitter chatStream(String prompt) { SseEmitter emitter = new SseEmitter(60000L); Flux<String> glmFlux = getGlmStream(prompt); glmFlux.subscribe( chunk -> { // 直接将大模型data分片推给前端 emitter.send(SseEmitter.event().data(chunk)); }, emitter::completeWithError, emitter::complete ); return emitter; }

4）前端 EventSource 原生监听 SSE

const source = new EventSource("/ai/sse/chat?prompt=写Java代码"); source.onmessage = (e) => { const raw = e.data; if (raw === "[DONE]") { source.close(); return; } const res = JSON.parse(raw); const text = res.choices[0].delta.content; document.getElementById("ans").innerText += text; }; source.onerror = () => console.log("连接断开，浏览器自动重连");

三、Demo 2：手写SseEmitter

SseEmitter只是 SpringMVC 对 SSE 协议、异步输出、连接生命周期的封装，底层本质就是操作HttpServletResponse输出text/event-stream格式文本。 手动手写分两种：

1. 同步手写（简单 Demo，缺点会占用 Tomcat 工作线程）
2. 异步手写（生产可用，不阻塞线程，复刻 SseEmitter 异步逻辑）

前置知识

SSE 响应固定三件套响应头，缺一不可：

// 1. 媒体类型固定SSE resp.setContentType("text/event-stream;charset=UTF-8"); // 2. 禁用缓存 resp.setHeader("Cache-Control", "no-cache"); // 3. 长连接保持 resp.setHeader("Connection", "keep-alive");

分片格式：data: 内容\n\n，双换行代表一条消息结束。

方案 1：同步手写 SSE（简易 Demo，不推荐高并发）

缺点：整个流式过程占用 Tomcat 同步线程，长连接会耗尽线程池

import jakarta.servlet.http.HttpServletResponse; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RestController; import java.io.IOException; import java.io.PrintWriter; @RestController public class RawSseController { // 对接大模型流式接口，同步手写SSE返回前端 @GetMapping("/raw/sync/sse") public void syncSse(@RequestParam String prompt, HttpServletResponse resp) throws IOException, InterruptedException { // 固定SSE响应头 resp.setContentType("text/event-stream;charset=UTF-8"); resp.setHeader("Cache-Control", "no-cache"); resp.setHeader("Connection", "keep-alive"); resp.flushBuffer(); PrintWriter out = resp.getWriter(); // 模拟调用智谱GLM获取流式分片（替换成真实WebClient调用逻辑） String[] chunks = { "{\"delta\":{\"content\":\"手写SSE第一块\"}}", "{\"delta\":{\"content\":\"手写SSE第二块\"}}", "[DONE]" }; for (String chunk : chunks) { // SSE标准格式 data: xxx\n\n out.write("data: " + chunk + "\n\n"); out.flush(); // 强制刷出缓冲区，前端立刻收到 Thread.sleep(500); } out.flush(); out.close(); } }

前端不变，依旧用 EventSource

const es = new EventSource("/raw/sync/sse?prompt=test"); es.onmessage = e => { console.log("收到分片：", e.data); };

方案 2：异步手写 SSE（生产级，复刻 SseEmitter 异步能力）

解决同步方案阻塞 Tomcat 线程的核心问题：

1. 开启异步上下文startAsync()
2. 业务逻辑丢入独立线程池执行
3. 监听连接断开事件，释放资源，避免内存泄漏

@GetMapping("/raw/async/sse") public void asyncSse(@RequestParam String prompt, HttpServletResponse resp) throws IOException { // 1. SSE标准响应头 resp.setContentType("text/event-stream;charset=UTF-8"); resp.setHeader("Cache-Control", "no-cache"); resp.setHeader("Connection", "keep-alive"); resp.flushBuffer(); // 2. 开启Servlet异步上下文，释放Tomcat主线程 AsyncContext asyncCtx = resp.startAsync(); asyncCtx.setTimeout(60000); // 60秒超时，和SseEmitter默认行为对齐 PrintWriter out = resp.getWriter(); // 3. 监听连接关闭/超时事件，做资源清理 asyncCtx.addListener(new AsyncListener() { @Override public void onComplete(AsyncEvent event) throws IOException { out.close(); } @Override public void onTimeout(AsyncEvent event) throws IOException { out.write("data: {\"done\":true}\n\n"); out.flush(); out.close(); asyncCtx.complete(); } @Override public void onError(AsyncEvent event) throws IOException { out.close(); } @Override public void onStartAsync(AsyncEvent event) {} }); // 4. 独立线程执行大模型流式拉取，不占用Tomcat线程池 new Thread(() -> { try { // 这里替换 WebClient 调用智谱GLM流式接口 String[] chunks = { "{\"delta\":{\"content\":\"异步手写SSE分片1\"}}", "{\"delta\":{\"content\":\"异步手写SSE分片2\"}}", "[DONE]" }; for (String c : chunks) { out.write("data: " + c + "\n\n"); out.flush(); Thread.sleep(400); } } catch (Exception e) { e.printStackTrace(); } finally { // 流式全部输出完毕，关闭异步上下文 asyncCtx.complete(); } }).start(); }

核心优势（和 SseEmitter 对齐）

1. Tomcat 工作线程立刻释放，能支撑大量并发长连接；
2. 自动处理超时、前端主动关闭页面、网络异常断开；
3. 完全不依赖SseEmitter任何 API，纯原生 Servlet 实现。

四、Demo 3：自研 text/plain 分行流式（无 SSE 协议）

仅内部中转服务使用，响应头Content-Type: text/plain，每条 JSON 用单换行\n分隔，无data:前缀。

1）后端 Spring 接口输出 text/plain 流

@GetMapping(value = "/ai/plain/stream", produces = MediaType.TEXT_PLAIN_VALUE) public void plainStream(String prompt, HttpServletResponse resp) throws IOException { resp.setContentType("text/plain;charset=UTF-8"); resp.setHeader("Transfer-Encoding", "chunked"); PrintWriter out = resp.getWriter(); // 模拟大模型分片，每行一条完整JSON out.println("{\"content\":\"第一段代码\",\"done\":false}"); out.flush(); Thread.sleep(300); out.println("{\"content\":\"第二段代码\",\"done\":false}"); out.flush(); Thread.sleep(300); out.println("{\"content\":\"\",\"done\":true}"); out.flush(); out.close(); }

2）前端 fetch + ReadableStream 手动解析（无 EventSource 可用）

async function fetchPlainStream() { const res = await fetch(`/ai/plain/stream?prompt=测试`, { method: "GET" }); const reader = res.body.getReader(); const decoder = new TextDecoder("utf-8"); let buffer = ""; while (true) { const { done, value } = await reader.read(); if (done) break; buffer += decoder.decode(value); // 按换行切割分片 const lines = buffer.split("\n"); buffer = lines.pop(); for (const line of lines) { if (!line.trim()) continue; const json = JSON.parse(line); document.getElementById("ans").innerText += json.content; if (json.done) reader.cancel(); } } } fetchPlainStream();

五、关键补充区分（容易混淆的点）

1. 请求头永远是 application/json两种流式方案，客户端 POST 上传参数时，Content-Type: application/json不变；区别只在服务返回的响应头。
2. 为什么大厂不用 text/plain

• 需要前端手写复杂流切割逻辑，EventSource 原生能力全部作废；
• 没有断线自动重连、断点续传，网络抖动体验极差；
• 无统一结束标记、事件类型，多模型对接要重复写适配代码。

1. 什么时候才用 text/plain 仅小型内部工具、单机 Demo，不想引入 SSE 规范、不需要稳定长连接的场景；线上面向用户的 AI 问答一律使用text/event-stream。