什么是 SseEmitter？如何使用 SseEmitter 实现 AI 流式响应？

标签：后端、编程导航

🏛 业务架构师

SseEmitter 是 Spring MVC 提供的异步响应句柄：Controller 方法返回 SseEmitter，请求线程立即释放（不占用容器线程），业务线程随模型生成逐块 emitter.send(...)，完成后 complete()。

实现 AI 流式响应的关键代码结构：

Controller 返回 SseEmitter（超时设 0 或长值）；
订阅模型流：chatClient.stream() 返回的 Flux 逐 token 订阅，每块封装成 SSE 事件（区分事件类型：token / 工具调用状态 / 完成 / 错误）——事件类型设计是前端体验的契约（前端按类型渲染打字机、工具图标、结束态）；
收尾与异常：完成时 complete()、异常时带错误事件后 completeWithError()、客户端断开时 onCompletion/onTimeout 清理资源（取消上游模型订阅——用户关页面了还让模型继续算，就是白烧钱）。

生产级的四个坑（都是真实教训）：

异步配置缺失：Tomcat 线程池不匹配会导致连接挂起——异步支持与超时体系要整体配置；
网关/代理超时：Nginx 默认 60s 掐断长连接——proxy_read_timeout 调大 + 应用层心跳（每 15 秒发注释行）；
一次只能一个消费者：SseEmitter 非线程安全，多线程乱序 send 会炸——严格单线程顺序写（reactive 的串行订阅天然满足）；
断线重连的续传：标准 SSE 有 Last-Event-ID，但模型流难以从中间续——断线即结束本轮，前端保留已显示内容并提示重试，别追求假续传。

一句话：SseEmitter 把"流式 AI"接进 Spring 生态的成本压到最低，但生产化的功夫全在超时、心跳、资源清理这些细节纪律上。

🏗 系统架构师

SseEmitter 是 Spring MVC 提供的 SSE 推送抽象：Controller 返回 SseEmitter，容器保持连接，业务线程随时调用 emitter.send() 推送事件，complete() 结束。

用 SseEmitter 实现 AI 流式响应的完整链路：

java

@GetMapping(value = "/chat", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter chat(@RequestParam String q) {
    SseEmitter emitter = new SseEmitter(60_000L); // 超时时间
    chatService.streamChat(q, emitter);            // 异步处理
    return emitter;                                // 立即返回，连接挂起
}

// service 内（独立线程/响应式订阅，不阻塞容器线程）
public void streamChat(String q, SseEmitter emitter) {
    chatClient.prompt().user(q).stream()           // Spring AI 流式调用
        .doOnNext(token -> {
            try { emitter.send(SseEmitter.event().data(token)); }
            catch (IOException e) { /* 客户端断开 */ }
        })
        .doOnComplete(() -> emitter.complete())
        .doOnError(e -> emitter.completeWithError(e))
        .subscribe();
}

必须处理的工程细节（生产分水岭）：

异步执行：生成逻辑放独立线程池（@Async 或 WebClient 调模型），Controller 线程立即返回——否则占用容器线程；
超时与心跳：设 emitter 超时；LLM 首 token 前（可能 5s+）定期发注释行心跳防中间层断连；
客户端断开：onCompletion/onTimeout 回调里取消上游 LLM 调用（dispose 订阅），省 token 钱与连接资源；
异常时推送结构化错误事件再 complete，前端能区分正常结束与失败；
Nginx/网关关闭响应缓冲（proxy_buffering off / X-Accel-Buffering: no），否则"流式"变"一次性"。

协议格式：event 名（message/done/error）+ data 分片，前端按事件类型渲染打字机与结束态。

💻 全栈开发专家

SseEmitter = Spring MVC 提供的 SSE 推送句柄：Controller 创建后立即返回（释放 Tomcat 线程），异步线程持续往 emitter 写事件。

java

@GetMapping(value = "/chat/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter chat(@RequestParam String message,
                       @RequestParam String conversationId) {
    SseEmitter emitter = new SseEmitter(120_000L);        // 超时 2 分钟
    chatService.streamChat(conversationId, message, emitter);  // 异步
    return emitter;
}

// 服务层：订阅模型的 Flux，逐 token 发送
public void streamChat(String convId, String msg, SseEmitter emitter) {
    chatClient.prompt().user(msg)
        .stream()                                         // Flux<ChatResponse>
        .doOnNext(chunk -> {
            try {
                emitter.send(SseEmitter.event()
                    .data(chunk.getResult().getOutput().getText()));
            } catch (IOException e) { /* 客户端断开 */ }
        })
        .doOnComplete(() -> emitter.complete())
        .doOnError(e -> emitter.completeWithError(e))
        .subscribe();
}

踩过的四个坑（生产必遇）：

Nginx/网关缓冲：响应被缓冲导致"半天一次性吐出"——proxy_buffering off、X-Accel-Buffering: no；
超时设置：Nginx proxy_read_timeout、网关、SseEmitter 三层超时要打通（否则 60s 被掐）；
连接清理：onCompletion/onTimeout/onError 回调里释放资源、终止上游 Flux 订阅（用户关页面后别再白调模型烧钱）；
心跳：长思考期间没有 token 输出，中间层判死——每 15s 发一个注释事件 :\n\n 保活。

什么是 SseEmitter？如何使用 SseEmitter 实现 AI 流式响应？ ​

🏛 业务架构师 ​

🏗 系统架构师 ​

💻 全栈开发专家 ​

什么是 SseEmitter？如何使用 SseEmitter 实现 AI 流式响应？

🏛 业务架构师

🏗 系统架构师

💻 全栈开发专家