SSE流式解析与异步执行
上一篇拆解了 doChat 同步调用的完整流程——校验供应商、构建请求体、发送 HTTP POST、解析 JSON 响应、提取 choices[0].message.content，九个步骤一路走下来，很直接。
但在文章最后提到，AbstractOpenAIStyleChatClient 还有一个 doStreamChat 模板方法。同步调用的逻辑很好理解：发一个 HTTP POST，等响应回来，解析 JSON，返回 String。流式调用就不一样了——响应不是一次到位的，而是一行一行通过 SSE 协议推送过来的。读取过程是阻塞的（readUtf8Line() 会阻塞等待下一行数据到达），不能在 Tomcat 请求线程上跑，得提交到专用线程池异步执行。而且用户随时可能点停止生成，得有一套取消机制能中断正在进行的流式读取。
这一篇打开流式调用的黑盒，看看 doStreamChat 内部是怎么把 SSE 解析、异步执行、回调推送、取消控制串在一起的。
doStreamChat vs doChat：同步与流式的分叉点
1. doStreamChat 模板方法
先看 doStreamChat 的完整代码：
protected StreamCancellationHandle doStreamChat(ChatRequest request, StreamCallback callback, ModelTarget target) {
AIModelProperties.ProviderConfig provider = HttpResponseHelper.requireProvider(target, provider());
if (requiresApiKey()) {
HttpResponseHelper.requireApiKey(provider, provider());
}
JsonObject reqBody = buildRequestBody(request, target, true);
Request streamRequest = newAuthorizedRequest(provider, target)
.post(RequestBody.create(reqBody.toString(), HttpMediaTypes.JSON))
.addHeader("Accept", "text/event-stream")
.build();
Call call = httpClient.newCall(streamRequest);
boolean reasoningEnabled = isReasoningEnabledForStream(request);
return StreamAsyncExecutor.submit(
modelStreamExecutor,
call,
callback,
cancelled -> doStream(call, callback, cancelled, reasoningEnabled)
);
}
和上一篇的 doChat 放在一起对比，前半部分几乎一模一样：校验供应商配置 → 校验 API Key → buildRequestBody 构建请求体 → newAuthorizedRequest 构建 HTTP 请求。但从构建请求体开始，三个关键差异出现了：
| 维度 | doChat（同步） | doStreamChat（流式） |
|---|---|---|
| 请求体 | buildRequestBody(request, target, false) | buildRequestBody(request, target, true)，JSON 中多了 "stream": true |
| Accept 头 | 无特殊设置 | .addHeader("Accept", "text/event-stream") |
| HTTP 执行 | 当前线程同步 httpClient.newCall(request).execute() | 创建 Call 后交给 StreamAsyncExecutor.submit() 异步执行 |
| 响应解析 | HttpResponseHelper.parseJson() 一次性解析完整 JSON | OpenAIStyleSseParser.parseLine() 逐行解析 SSE |
| 结果返回 | 直接 return String | 通过 StreamCallback 回调推送 |
| 取消控制 | 无（同步阻塞等待） | 返回 StreamCancellationHandle，调用方可随时取消 |
| 错误处理 | 抛 ModelClientException，由 executeWithFallback 捕获 | callback.onError() 通知调用方 |
分叉点就在 httpClient.newCall(streamRequest) 这一行——同步调用直接 .execute() 等结果，流式调用把 Call 对象交给异步执行器，自己立即返回一个取消句柄。
2. 为什么流式读取必须异步
这个问题值得展开说说。
OkHttp 的流式响应通过 BufferedSource.readUtf8Line() 逐行读取。这个方法是阻塞调用——如果服务端还没推送新数据，readUtf8Line() 会一直等着，直到新的一行到达或者连接关闭。
大模型生成一个完整回答可能需要几十秒甚至几分钟（生成长文本时），在这段时间里，线程一直被占着。如果把流式读取放在 Tomcat 的请求处理线程上，一个流式请求就占住一个 Tomcat 线程。Tomcat 线程池默认 200 个线程，假设同时有 50 个用户在流式对话，就有 50 个线程被占住。剩下的线程要处理其他所有 HTTP 请求——页面渲染、API 调用、健康检查等等。并发一上来，线程池就满了，新请求排队等待，整个系统响应变慢。
所以必须把流式读取提交到专用线程池 modelStreamExecutor 上异步执行。Tomcat 请求线程只负责构建 HTTP 请求和创建 Call 对象，然后立即返回取消句柄，不等待流式传输完成。真正的阻塞读取在专用线程池上进行，和 Tomcat 线程池互不干扰。
StreamCallback：回调体系
1. 接口定义
public interface StreamCallback {
/**
* 接收一次增量内容（Delta Token 或部分片段）
*/
void onContent(String content);
/**
* 接收思考过程增量内容（如果模型支持）
* 默认空实现，未支持思考的场景可以忽略
*/
default void onThinking(String content) {
}
/**
* 整个推理流程结束（全部内容推送完毕）
*/
void onComplete();
/**
* 流式推送过程中出现异常
*/
void onError(Throwable error);
}
四个方法分工明确：
onContent(String content)——增量内容回调。模型每生成一段内容（可能是一个字、一个词、一句话，取决于模型的分片策略），就通过这个方法推一次。调用次数不定，由模型输出节奏决定onThinking(String content)——思考过程回调。深度思考模型（比如开启了enable_thinking的 Qwen）会在生成正式回答之前输出推理过程，通过这个方法推送。注意它是default空实现onComplete()——流结束通知。所有内容推送完毕后调用一次，通知上层做收尾动作（停止 loading 动画、滚动到底部等）onError(Throwable error)——异常通知。网络中断、解析异常、供应商报错等场景触发
2. 调用时序
正常流程的回调顺序：
onThinking("用户在问候") ← 可选，深度思考模型才有
onThinking("，我应该友好回应") ← 可选
onContent("你")
onContent("好")
onContent("！")
onComplete() ← 正好一次
异常流程：
onContent("你")
onContent("好")
onError(IOException) ← 任意时刻，一次
onComplete() 和 onError() 是互斥的终态——正常结束走 onComplete()，异常结束走 onError()，不会两个都调用。
2.1 onThinking 的默认空实现
onThinking 被声明为 default 方法，方法体为空。这个设计的意图是：大多数场景不关心模型的思考过程，只关心最终生成的内容。如果 onThinking 是一个抽象方法，所有 StreamCallback 的实现类都得写一个空的 onThinking 方法，哪怕根本用不到。default 空实现让不需要思考内容的调用方零成本忽略它——只实现 onContent、onComplete、onError 三个方法就够了。
3. 和同步调用返回值的对比
同步调用 doChat 的返回方式是直接 return String——调用方同步阻塞等待整个回答生成完毕，然后一次性拿到完整文本。
流式调用通过 StreamCallback 异步推送——调用方不阻塞等待，增量内容逐步到达。这种模式让前端可以实现逐字输出的打字机效果：收到一个 onContent("你") 就在页面上渲染一个字，不用等整个回答生成完才显示。用户体验上，从发问到看到第一个字的延迟（首字延迟，Time to First Token）大幅缩短。
OpenAIStyleSseParser：SSE 行解析
1. 为什么需要专门的解析器
大模型的流式响应走 SSE（Server-Sent Events）协议，但实际上用的是一个简化子集——只用了 data: 字段，没有 event:、id:、retry: 等标准 SSE 字段。而且 data: 后面的 payload 是 JSON，和非流式响应的结构有区别：
- 非流式响应：内容在
choices[0].message.content - 流式响应：增量内容在
choices[0].delta.content - 流结束标记
[DONE]是 OpenAI 协议特有的约定，不在 SSE 标准里
所以需要一个专门的解析器来处理这些 OpenAI 协议特有的约定。OpenAIStyleSseParser 就是这个角色——它不是一个通用的 SSE 协议解析器，而是一个针对 OpenAI 兼容接口优化的 data-only SSE 消费器。
2. parseLine 完整代码
@NoArgsConstructor(access = lombok.AccessLevel.PRIVATE)
public final class OpenAIStyleSseParser {
private static final String DATA_PREFIX = "data:";
private static final String DONE_MARKER = "[DONE]";
static ParsedEvent parseLine(String line, Gson gson, boolean reasoningEnabled) {
if (line == null || line.isBlank()) {
return ParsedEvent.empty();
}
String payload = line.trim();
if (payload.startsWith(DATA_PREFIX)) {
payload = payload.substring(DATA_PREFIX.length()).trim();
}
if (DONE_MARKER.equalsIgnoreCase(payload)) {
return ParsedEvent.done();
}
JsonObject obj = gson.fromJson(payload, JsonObject.class);
JsonArray choices = obj.getAsJsonArray("choices");
if (choices == null || choices.isEmpty()) {
return ParsedEvent.empty();
}
JsonObject choice0 = choices.get(0).getAsJsonObject();
String content = extractText(choice0, "content");
String reasoning = reasoningEnabled ? extractText(choice0, "reasoning_content") : null;
boolean completed = hasFinishReason(choice0);
return new ParsedEvent(content, reasoning, completed);
}
private static boolean hasFinishReason(JsonObject choice) {
if (choice == null || !choice.has("finish_reason")) {
return false;
}
JsonElement finishReason = choice.get("finish_reason");
return finishReason != null && !finishReason.isJsonNull();
}
private static String extractText(JsonObject choice, String fieldName) {
if (choice == null) {
return null;
}
if (choice.has("delta") && choice.get("delta").isJsonObject()) {
JsonObject delta = choice.getAsJsonObject("delta");
if (delta.has(fieldName)) {
JsonElement value = delta.get(fieldName);
if (value != null && !value.isJsonNull()) {
return value.getAsString();
}
}
}
if (choice.has("message") && choice.get("message").isJsonObject()) {
JsonObject message = choice.getAsJsonObject("message");
if (message.has(fieldName)) {
JsonElement value = message.get(fieldName);
if (value != null && !value.isJsonNull()) {
return value.getAsString();
}
}
}
return null;
}
record ParsedEvent(String content, String reasoning, boolean completed) {
static ParsedEvent empty() {
return new ParsedEvent(null, null, false);
}
static ParsedEvent done() {
return new ParsedEvent(null, null, true);
}
boolean hasContent() {
return content != null && !content.isEmpty();
}
boolean hasReasoning() {
return reasoning != null && !reasoning.isEmpty();
}
}
}
parseLine 接收一行 SSE 文本，返回一个 ParsedEvent。解析流程分五步：
- 空行过滤：
null或空白行直接返回ParsedEvent.empty() - 剥离
data:前缀：SSE 每行数据都以data:开头，去掉前缀拿到真正的 payload - 识别
[DONE]标记：OpenAI 协议约定的流结束标记，返回ParsedEvent.done() - 解析 JSON：把 payload 解析为
JsonObject，提取choices[0] - 提取字段：从
choice0中提取content（增量内容）和reasoning_content（思考内容，如果启用），检查finish_reason判断是否完成
整个类是 private 构造的工具类（@NoArgsConstructor(access = PRIVATE)），所有方法都是 static，不维护状态——每行独立解析，无上下文依赖。