【简历进阶篇】SSE流式输出Spring WebFlux

个人主页代码不加冰欢迎来访作者简介java后端学习者❄️个人专栏LeetCode刷题日记 苍穹外卖日记SSM框架深入JavaWeb✨命运的结局尽可永在不屈的挑战却不可须臾或缺前言大家好我是代码不加冰这几天生病了浑身都疼然后又出去了一天多耽误了挺多然后今天上午看了一会项目这么给大家分享一下今天在项目里学到的一个比较重要的知识吧。摘要本文介绍了SSEServer-Sent Events技术在Java项目中的实现与应用。SSE是一种基于HTTP的服务器向客户端单向推送数据的技术适用于ChatGPT等大语言模型的流式输出场景。文章详细阐述了SSE与传统HTTP请求的区别以及与WebSocket的对比并提供了符合企业开发规范的SpringBoot实现方案包括Controller层、Service层接口定义和实现类的代码示例。此外文章还探讨了WebFlux框架在SSE实现中的优势如高并发处理能力、非阻塞I/O操作和响应式编程模型。通过本文读者可以了解到如何在Java项目中高效地实现SSE技术以支持流式数据传输需求。SSEServer-Sent Events服务器发送事件是一种基于 HTTP 协议的服务器向客户端进行单向数据推送的技术。我们日常使用 的ChatGPT、Claude 等大语言模型时看到字是一个一个蹦出来的这种流式输出的效果底层最常用的技术就是 SSE。一。 什么是 SSE在传统的 HTTP 请求中客户端发送一个请求服务器返回一个响应连接随即关闭即一问一答模式。而 SSE 的工作原理是客户端发起一个标准的 HTTP 请求但请求头中明确要求Accept: text/event-stream声明需要事件流。服务器保持这个 HTTP 连接打开不关闭。每当服务器有新数据时就会通过这个连接以特定的文本格式data: xxx\n\n源源不断地写给客户端。SSE 与 WebSocket 的区别WebSocket全双工通信双向通道客户端和服务器可以互相随意发数据基于独立的ws://协议开发相对复杂。SSE单向通信服务器 rightarrow 客户端基于标准的HTTP 协议轻量级自带断线重连机制非常适合不需要客户端频繁写数据、只需要服务器推送的场景如 AI 聊天、股票行情、直播弹幕。二。SSE 的数据传输格式SSE 传输的是纯文本服务器返回的 Response Header 必须包含Content-Type: text/event-stream Cache-Control: no-cache Connection: keep-alive每次发送的数据必须遵循以下固定格式以\n\n结尾作为一条消息的结束data: 这是第一句完整的话\n\n data: 这是第二句话\n\n三。具体实现在实际的 Java 项目中特别是使用 Spring Boot 的企业级架构中Controller 层通常只负责接收请求和处理 HTTP 协议相关的逻辑。真正的核心业务比如对接 AI 大模型、处理数据、控制重试等都会封装在Service 层。由于 SSE 的特性是持续性输出在 Service 层处理时我们需要将SseEmitter或者是数据的推送逻辑优雅地延伸到 Service 中。下面是符合企业开发规范的Controller Service 层的代码逻辑实现1. Controller 层保持轻量只做转发Controller 只负责创建SseEmitter然后将其传递给 Service 层去异步执行业务最后立即返回。import org.springframework.beans.factory.annotation.Factory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.http.MediaType; import org.springframework.web.bind.annotation.*; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; RestController RequestMapping(/api/chat) public class ChatController { Autowired private ChatService chatService; /** * 核心通过 produces 声明这是一个流式响应接口 */ GetMapping(value /stream, produces MediaType.TEXT_EVENT_STREAM_VALUE) public SseEmitter streamChat(RequestParam String prompt) { // 1. 创建 Emitter 设置 1 分钟超时 SseEmitter emitter new SseEmitter(60_000L); // 2. 调用 Service 层执行具体的流式业务逻辑 chatService.executeStreamChat(prompt, emitter); // 3. 立即返回不阻塞 Tomcat 线程 return emitter; } }2. Service 层接口定义import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; public interface ChatService { /** * 执行流式对话业务 * param prompt 用户的提示词 * param emitter 用于推送数据的连接器 */ void executeStreamChat(String prompt, SseEmitter emitter); }3. Service 层实现类核心逻辑在 Service 层我们通常会使用 Spring 提供的Async注解来开启异步线程需要开启EnableAsync线程池或者手动使用线程池。这里我们演示如何在 Service 层规范地操作SseEmitter并处理连接中断等异常情况。import org.springframework.scheduling.annotation.Async; import org.springframework.stereotype.Service; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; import java.io.IOException; Service public class ChatServiceImpl implements ChatService { /** * 使用 Async 开启异步线程执行防止阻塞主线程 * 注意实际开发中建议在配置类中自定义线程池而不是使用 Spring 默认的 SimpleAsyncTaskExecutor */ Async(chatExecutor) Override public void executeStreamChat(String prompt, SseEmitter emitter) { // 1. 注册客户端中断或超时的回调用于清理 Service 层的资源 emitter.onCompletion(() - System.out.println(Service层流式交互结束释放资源。)); emitter.onTimeout(() - System.out.println(Service层流式交互超时。)); try { // 2. 模拟调用大模型LLM获取流式数据的过程 // 在实际项目中这里可能是 HttpClient 调用 ChatGPT 的 stream 接口或者 DeepSeek 的 SDK String mockAiResponse 这是从 Service 层返回的流式数据。Prompt 是 prompt; String[] chunks mockAiResponse.split(); // 拆分成单个字 for (String chunk : chunks) { // 3. 核心调用 emitter.send 推送数据 // 包装成标准的 data: {chunk} \n\n 结构 emitter.send(SseEmitter.event().data(chunk)); // 模拟网络延迟或 AI 吐字间隔 Thread.sleep(150); } // 4. 业务数据发送完毕发送结束标识并闭环结束 Emitter emitter.send(SseEmitter.event().name(complete).data([DONE])); emitter.complete(); } catch (IOException e) { // 5. 异常处理通常发生在客户端在“蹦字”过程中主动关闭了浏览器标签页导致管道破裂ClientAbortException System.err.println(客户端主动断开连接: e.getMessage()); emitter.completeWithError(e); } catch (InterruptedException e) { System.err.println(业务执行被中断: e.getMessage()); emitter.completeWithError(e); Thread.currentThread().interrupt(); } } }4. 为什么 Service 层要这么设计① 职责分离Separation of ConcernsController 仅仅充当“门卫”它实例化了SseEmitter通道并立刻把它交给 Service。Service 负责产生数据并往这个通道里“塞”数据。Controller 不感知数据是怎么来的、经历了什么复杂的过滤和清洗。② 强依赖Async或自定义线程池因为 Service 里含有Thread.sleep或者等待第三方 AI 接口返回的阻塞操作如果不加Async这个方法就会在 Controller 的当前线程Tomcat 线程里同步执行。这会导致return emitter被阻塞直到所有字蹦完才会返回。这就完全失去了 SSE 异步流式的意义退化成了普通的同步阻塞请求。③ 异常捕获的必要性IOException在长连接流式推送中客户端随时可能断开连接比如用户刷新了网页、关闭了聊天窗口、或者网络信号不好断网。如果客户端断开Service 层在执行emitter.send()时就会抛出IOException。如果不进行try-catch捕获线程就会异常崩溃且无法通过emitter.completeWithError(e)通知 Spring 清理该连接的上下文可能会造成内存泄漏。四。项目代码演示这段代码是一套非常企业级、严密且优雅的AI 流式对话SSE/WebSocket后端 Service 层实现。它基于 Spring WebFluxProject Reactor 响应式编程框架构建核心是通过FluxString实现非阻塞的字流Token Stream推送。我们采用由浅入深的逻辑将其拆解为四大核心层次进行详细分析整体调用链路与核心流程这段代码由 5 个方法组成它们各司其职形成了一条标准的管道链[入口 1] aiChatFlux(requestParam) - 自动提取 username ↓ [入口 2] aiChatFlux(requestParam, username) - 参数强校验 越权检查 ↓ [内部构建] aiChatFluxInternal - 创建 Flux 响应式管道切入自定义线程池 ↓ [核心编排] processChat - 统一调度第三方支持组件查历史、存消息、流式调用、收尾落库 ↓ [配置解析] resolveAiProperties - 精准获取/校验大模型配置核心方法1. 门面入口与鉴权Override public FluxString aiChatFlux(AiMessageReqDTO requestParam, String username) { if (requestParam null) { ... } // 健壮性防御 // ... 检查 sessionId 和 username 是否为空 // 关键点越权安全检查 aiConversationService.requireOwnedConversation(requestParam.getSessionId(), username); requestParam.setUserName(username); return aiChatFluxInternal(requestParam); }逻辑严密性这里引入了requireOwnedConversation。在多租户或多用户系统中这是极重要的纵向越权防护。它在长连接建立前强行校验当前操作人username确实是sessionId会话窗口的主人。防止黑客通过猜测sessionId偷窥他人对话或刷额度。2. 桥接响应式与多线程aiChatFluxInternalprivate FluxString aiChatFluxInternal(AiMessageReqDTO requestParam) { // ... 参数校验 return Flux.create(sink - { // 1. 初始化累加器用于拼接大模型吐出的完整回答 AIContentAccumulator accumulator new AIContentAccumulator(); // 2. 显式切换线程将耗时的会话生命周期丢入自定义线程池 threadPoolTaskExecutor.submit(() - processChat(sessionId, aiId, userMessage, sink, accumulator)); // 3. 监听流的生命周期 sink.onCancel(() - log.warn(AI chat flux cancelled, sessionId{}, sessionId)); sink.onDispose(() - log.info(AI chat flux disposed, sessionId{}, sessionId)); }); }Flux.create(sink - {...})这里是创建异步流的源头。sink数据汇是连接底层网络和当前业务的纽带。显式线程切换为什么用threadPoolTaskExecutor.submit而不用Async响应式编程中Async的兼容性较差且它是个黑盒。显式提交能清晰地将Netty 事件循环线程负责网络响应与大模型 IO 阻塞线程负责拉取 AI 数据并落库隔离开来这是 WebFlux 保持高并发能力的核心。生命周期钩子sink.onCancel极为关键。当用户关闭网页、刷新页面时前端连接断开Spring 会立刻触发onCancel。3. 模板化流式编排processChat这个方法使用了非常优美的模板方法模式。它把复杂的 AI 对话生命周期抽象成一个配置对象交给conversationStreamingSupport执行器。我们看它的具体编排参数historySupplier提供历史消息。大模型对话需要上下文这里传入了一个 Lambda 表达式框架会在合适的时候自动调用conversationMessageHistoryService.listAiHistory捞取历史记录。userMessageSaver异步保存用户发出的这一句话保证消息不丢失。streamExecutor大模型调用核心AiPropertiesDO aiProperties resolveAiProperties(aiId); AiChatHandler handler aiChatHandlerFactory.getHandler(aiProperties.getAiType()); handler.streamToSink(aiProperties, userMessage, historyMessages, sink, contentAccumulator);通过策略工厂aiChatHandlerFactory根据不同的 AI 类型如DOUBAO、DEEPSEEK、OPENAI拿到对应的处理器然后调用streamToSink往sink管道里面源源不断地写数据sink.next()。assistantMessageSaver落库收尾 注意这里的payload.reasoningContent()。这是极具前瞻性的设计专门兼容了类似DeepSeek-R1、OpenAI o1等带思维链Reasoning的模型。在 AI 完全吐完字后框架会把“思考过程”和“最终回答”一起持久化到数据库。状态机控制 (successHandler/errorHandler)if (!sink.isCancelled()) { sink.complete(); }在结束或报错时先判断!sink.isCancelled()。如果用户网页早就关了就不再向通道发数据防止触发Broken Pipe管道破裂异常。4. 配置防呆解析resolveAiPropertiesprivate AiPropertiesDO resolveAiProperties(Long aiId) { if (aiId null) { aiProperties aiPropertiesService.getDefaultDoubaoConfig(); // 默认保底机制豆包模型 // ... } else { // ... 严格校验逻辑删除(delFlag)与启用状态(isEnabled) } }设计亮点提供了“默认保底机制”。如果前端没传aiId直接走系统默认的配置代码中默认为豆包配置。如果传了则进行三段式严格校验存在性、逻辑删除标志、禁用标志设计亮点高内聚低耦合策略模式通过AiChatHandlerFactory动态获取 handlerService 层的核心流程processChat与具体的 AI 厂商完全解耦。无论是升级模型还是更换供应商都不需要动这套核心逻辑。算力控本Cancellation Aware结合了sink.onCancel并在成功/错误处理中检查isCancelled()。如果用户问了一个问题模型刚要长篇大论用户把窗口关了。此时系统会立刻停止后续调用避免无谓的第三方 AI Token 费用消耗。数据一致性状态全回溯用户消息先存、AI 消息后存、会话状态如消息序号messageSeq最后更新。由conversationStreamingSupport组件统一提供事务或顺序保障设计极其严密。支持深度思考模型Reasoning 支持超前的持久化设计将大模型的content和reasoningContent拆开存储满足了最新一代推理大模型的业务落地需求。完美的线程隔离利用threadPoolTaskExecutor承载了整个大模型交互生命周期包含多次 DB 读写和 HTTP 长连接 IO将阻塞完全隔离在特定的业务线程池中保护了微服务网关/核心控制器的基础吞吐量。五。WebFluxSpring WebFlux是 Spring 5于2017年发布引入的响应式、非阻塞式Web 框架。它是为了应对高并发、高吞吐量场景而设计的用来作为传统 Spring MVC 的互补或替代方案。1. 为什么需要 WebFlux在传统的Spring MVC中采用的是基于 Servlet 的“一个请求对应一个线程”Thread-per-request模型。传统 Spring MVC 的瓶颈当一个 HTTP 请求到达服务器时Tomcat 会从线程池中分配一个线程来处理这个请求。如果这个请求需要查询数据库DB、调用第三方 AI 接口HTTP或者读写文件I/O那么这个线程就会进入阻塞Blocking状态。尽管这个线程此时什么都不干只是在死等数据返回它依然会占用大约 1MB 的内存。当并发量极高时比如双十一秒杀、海量大模型流式对话线程池很快会被耗尽新的请求只能排队导致服务器响应变慢甚至瘫痪。为了解决传统 I/O 阻塞浪费线程资源的问题响应式堆栈 WebFlux 应运而生。2.WebFlux 的核心底层支撑WebFlux 能够实现高并发的核心在于它抛弃了传统的 Servlet 容器改用了一套全新的技术栈-------------------------------------------------- | Spring WebFlux | - 框架层 -------------------------------------------------- | Project Reactor | - 响应式流规范实现 (Flux/Mono) -------------------------------------------------- | Netty / Undertow / Servlet 3.1 | - 异步非阻塞底层引擎 --------------------------------------------------1. 异步非阻塞服务器NettyWebFlux 默认不使用 Tomcat而是使用Netty作为底层服务器。Netty 基于 NIONon-blocking I/O模型利用极少数的固定线程通常是 CPU 核心数的 2 倍就可以通过事件轮询机制Event Loop处理成千上万的并发连接。2. 响应式流规范Project ReactorWebFlux 的核心编程模型基于 Project Reactor它实现了 Reactive Streams 规范。在 WebFlux 的代码里你看不到直接返回的User或ListUser取而代之的是两个核心的响应式发布者PublisherMonoT代表0 个或 1 个异步元素的异步序列例如查询单个用户、保存单个对象。FluxT代表0 个到 N 个异步元素的异步序列例如查询列表、大模型的字流/Token流、SSE 推送。3. WebFlux 的工作原理我们用一个例子来对比Spring MVC和WebFlux的工作方式去餐馆吃饭模拟请求服务Spring MVC传统模式每个顾客请求配备一个专属服务员线程。服务员带你入座、听你点菜然后站在后厨门口死等厨师做好I/O阻塞再端给你。如果来了1000个顾客餐馆就必须雇佣1000个服务员成本极高。Spring WebFlux响应式模式整个餐馆只有 4 个服务员固定线程池。服务员 A 帮你点完菜把订单丢给后厨注册事件立刻转身去服务其他顾客。当你的菜做好后后厨敲铃通知触发回调事件服务员 B不一定是 A顺手把菜端给你。在 WebFlux 中线程绝不等待。当遇到数据库查询或远程调用时WebFlux 会发起异步请求并立即释放当前线程让它去处理其他请求。当数据准备好时底层通过事件驱动通知线程回来继续处理从而用极少的资源撬动了极高的并发。4. WebFlux vs Spring MVC特性维度Spring MVCSpring WebFlux底层服务器Tomcat, Jetty, Undertow (基于 Servlet)Netty, Undertow, Jetty (基于 Reactive)并发模型一个请求对应一个线程 (Thread-per-request)异步非阻塞事件循环 (Event Loop)编程范式命令式编程 (Imperative, 逐行同步执行)函数式、声明式编程 (Reactive, 管道流)数据处理一次性返回全部数据支持流式输出Reactive Stream / SSE适合场景传统微服务、强事务、复杂 CRUD 业务高并发网关、I/O 密集型、大模型流式对话、物联网依赖阻断任何一个环节阻塞整体就会发生阻塞要求全链路非阻塞如必须用 R2DBC 代替 JDBC六。WebFlux实现SSE的优势使用 Spring WebFlux 中的FluxString正如你前面展示的代码来实现 SSE相比于传统的SseEmitter基于 Servlet 阻塞堆栈具有以下的技术优势和架构收益1. 极致的资源利用率与吞吐量传统SseEmitter的痛点虽然SseEmitter允许异步发送数据但由于它底层基于传统的 Servlet 容器如 Tomcat每一个活动的 SSE 长连接依然会占用一个 Tomcat 异步线程的上下文。如果同一时间有 5000 个用户在跟 AI 聊天Tomcat 的线程池和内存很容易被卡死或耗尽。WebFluxFlux的优势WebFlux 底层基于 Netty采用事件循环Event Loop机制。它可以用极少数的固定线程通常是 CPU 核心数的 2 倍同时维持成千上万个 SSE 长连接。当 AI 还在“思考”或等待第三方大模型返回数据时Netty 的线程不会傻傻等待而是立刻去处理其他用户的请求。这让你的服务器在面对海量并发时内存和 CPU 依然保持极低的消耗。2. 全链路非阻塞生态结合流式聊天不仅包含“吐字”还包含查历史记录、保存消息、更新会话。如果你用传统的SseEmitter结合传统的 JDBC如 MyBatis / JPA每一个数据库的读写请求都是阻塞的都会拖慢线程。既然你已经用了Flux就可以非常丝滑地对接目前主流的全链路非阻塞组件如R2DBC响应式数据库驱动、WebClient响应式 HTTP 客户端、Reactive Redis。在这条链路上数据从数据库捞出来到通过 HTTP 传给大模型再到最后推给前端没有任何一个地方发生线程阻塞这是现代高性能微服务梦寐以求的闭环架构。3. 天然的生命周期控制在 AI 流式对话中最头疼的问题就是用户问了一个问题大模型刚要长篇大论用户嫌慢直接把网页关了。*传统SseEmitter模式后端很难在第一时间知道用户关了网页。后端会继续调用大模型、继续往缓冲区写数据直到缓冲区满了抛出IOException管道破裂时后端才后知后觉地停止。这导致大量的 AI Token 被白白浪费。WebFluxFlux模式我们代码中所写Flux.create提供了原生的生命周期钩子sink.onCancel(() - log.warn(AI chat flux cancelled...));一旦前端的 SSE 连接断开底层的网络通道会立刻通知 WebFlux从而触发onCancel回调。在你的processChat里只要判断sink.isCancelled()就能立刻终止第三方大模型的请求帮你死死卡住因为用户频繁刷新/关闭网页带来的 Token 成本损耗。4. 强大的流式算子使用Flux带来的不仅仅是性能提升更是开发效率的飞跃。Flux拥有极其丰富的响应式算子如map,filter,timeout,window,retry等可以让对字流的处理变得无比优雅。举个实际的企业级业务例子需求如果 AI 超过 5 秒没有吐出一个字就判定大模型超时并向前端发送一个默认的报错提示。如果用SseEmitter你需要自己写定时器手动管理每个连接的时间戳代码冗长且极其容易出错。如果用Flux只需要在你的 Flux 链条后面直接追加一行代码return aiChatFluxInternal(requestParam) .timeout(Duration.ofSeconds(5)) // 5秒不吐字自动触发超时 .onErrorReturn(DEFAULT_ERROR_CONTENT); // 超时后输出默认错误5. 高内聚的保护大模型吐字的速度和前端浏览器渲染、下载的速度不一定完全匹配。如果大模型疯狂吐字而用户的手机由于网络卡顿接收极慢就会导致大量的字在服务器内存中堆积。Flux内置了背压机制。当下游前端/网络通道处理不过来时它会向上游业务流/大模型接收端发出信号让上游放慢发送速度从而保护服务器的内存不会因为瞬时的数据堆积而导致 OOM内存溢出。总结选择用Flux来实现 SSE表面上看只是换了几行代码但底层实际上是从“多线程阻塞模型”进化到了“单线程/少线程的异步事件驱动模型”。