1. 项目概述为什么Java开发者需要关注AI流式输出最近在做一个AI对话功能的后端产品经理提了个需求用户问一个问题AI的回答要像真人打字一样一个字一个字地“流”出来而不是等AI全部生成完了再一次性返回。这个需求听起来简单但真做起来发现里面门道不少。传统的HTTP请求-响应模式是“一问一答”服务器处理完所有逻辑生成完整的响应体一次性返回给客户端。但在AI生成文本、实时数据推送、长任务执行进度汇报这些场景下这种模式就捉襟见肘了。用户会盯着空白页面干等体验很差如果生成时间过长还可能遇到请求超时。这就是“流式输出”要解决的问题。它的核心思想是把一个大的、耗时的响应拆分成多个小的数据块Chunk像水流一样持续不断地从服务器发送到客户端。对于Java后端开发者尤其是使用Spring Boot生态的实现流式输出主要有两种主流技术选择Server-Sent Events (SSE)和WebSocket。SSE是基于HTTP的单向通信特别适合服务器向客户端主动推送数据的场景比如新闻更新、股票价格、以及我们这里讨论的AI文本流式生成。而WebSocket是双向全双工通信更适合聊天室、在线协作等需要高频双向交互的场景。选择SSE来实现AI流式输出对于大多数Java应用来说是一个更轻量、更符合HTTP语义、也更容易上手的选择。它不需要像WebSocket那样引入额外的协议升级和复杂的连接管理利用普通的HTTP连接即可并且天然支持自动重连。接下来我们就深入拆解一下在Spring Boot应用中如何从零开始构建一个稳定、高效的AI流式输出接口。2. 核心技术原理与方案选型2.1 SSE协议深度解析SSE本质上是一个简单的协议它规范了服务器如何通过一个持久的HTTP连接向客户端发送一系列的事件流。我们来看一个最原始的SSE响应是什么样子HTTP/1.1 200 OK Content-Type: text/event-stream Cache-Control: no-cache Connection: keep-alive data: 这是第一段数据\n\n data: 这是第二段数据\n\n event: customEvent data: 这是一个自定义事件的数据\n\n : 这是一条注释\n id: 123\n data: 这条数据关联ID 123\n\n关键点在于响应头Content-Type: text/event-stream它告诉浏览器或客户端“接下来我要发送的是一个事件流请你用SSE的规则来解析。” 消息体由多行文本构成每一行以字段名开头后跟冒号和空格然后是字段值。核心字段有data: 表示数据行。一个事件可以包含多个data行它们会被连接成一个字符串用换行符分隔。两个连续的换行符\n\n标志着一个事件的结束客户端会触发onmessage回调。event: 指定事件类型。默认是message。客户端可以监听特定类型的事件。id: 事件ID。用于设置客户端lastEventId属性在连接断开重连时浏览器会自动在请求头中带上Last-Event-ID服务器可以据此决定从何处继续。retry: 指定重连时间毫秒。在Spring Boot中我们不需要手动拼接这些字符串。Spring Framework提供了对响应式编程范式的支持其核心是Reactive Streams规范。Flux是其中代表“0到N个元素的异步序列”的组件。当我们返回一个FluxString或FluxServerSentEvent并指定媒体类型为TEXT_EVENT_STREAM时Spring会帮我们自动将Flux中的每个元素按照SSE格式封装成一个个事件并通过HTTP响应流发送出去。这大大简化了开发。2.2 为什么是Spring Boot Flux而不是传统Servlet在传统的Servlet API包括Spring MVC的Controller中虽然也可以通过HttpServletResponse.getWriter()手动写入并刷新的方式模拟流式输出但这种方式对线程的占用不友好需要自己处理背压客户端处理不过来时服务器如何应对并且与响应式编程模型不搭。Spring WebFlux基于Project Reactor是Spring 5引入的响应式Web框架。它的核心优势在于非阻塞I/O和函数式编程模型。在处理流式输出时Flux可以优雅地表示一个数据流。当我们将AI大模型例如通过HTTP客户端调用OpenAI API或本地部署的模型返回的流式响应映射为一个Flux时整个数据通路就变成了响应式的从模型输出的字节流到Spring WebFlux的Flux数据流再到通过SSE协议输出的网络流。这个过程是异步非阻塞的一个工作线程可以处理大量并发连接资源利用率高非常适合高并发的流式推送场景。注意使用WebFlux并不意味着你必须将整个应用重构为响应式。你可以仅在需要流式输出的Controller中使用Flux作为返回值其他部分依然使用传统的Spring MVCRestController。Spring Boot能够很好地支持这种混合模式。2.3 与WebSocket的对比选型为了更清晰地做出技术选型我们通过一个表格来对比SSE和WebSocket在AI流式输出场景下的优劣特性Server-Sent Events (SSE)WebSocket通信方向单向(服务器 - 客户端)双向全双工协议基础HTTP(长连接)独立的WS/WSS协议(基于HTTP升级)连接管理简单HTTP自带。断开后浏览器自动重连。复杂需自行实现心跳、重连逻辑。数据格式文本事件流格式。二进制数据需编码如Base64。原生支持文本和二进制帧。浏览器兼容除IE/Edge旧版外现代浏览器支持良好。支持广泛。服务端实现(Spring)简单返回Flux注解GetMapping(produces MediaType.TEXT_EVENT_STREAM_VALUE)。相对复杂需配置WebSocketHandler处理会话。适用场景服务器向客户端推送实时通知、股票行情、AI文本流、日志流。双向实时交互在线聊天、协作编辑、多人在线游戏。与现有架构集成无缝集成现有HTTP安全层Spring Security、负载均衡、监控。可能需要额外处理例如在代理层支持WS协议。结论对于AI对话、内容生成这类典型的“客户端提问服务器持续推送回答”的场景SSE是更合适、更轻量的选择。它复用现有HTTP基础设施开发复杂度低并且自动重连机制提供了更好的鲁棒性。只有当你的应用同时需要客户端频繁向服务器发送数据如实时交互绘图AI时才需要考虑WebSocket。3. 实战构建Spring Boot AI流式输出接口3.1 基础环境与项目搭建首先我们创建一个标准的Spring Boot项目。如果你使用Spring Initializr需要选择以下依赖Spring Web(或Spring Reactive Web): 如果选择响应式栈直接选Reactive Web它会包含WebFlux。如果选择传统Servlet栈选Web但部分流式特性支持会弱一些。本文以更主流的WebFlux为例。Lombok: 减少样板代码可选但推荐。Spring Boot DevTools: 开发热部署。你的pom.xml关键依赖会类似这样dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-webflux/artifactId /dependency dependency groupIdorg.projectlombok/groupId artifactIdlombok/artifactId optionaltrue/optional /dependency3.2 核心Controller实现我们来创建一个最简单的流式输出接口模拟AI逐字生成的效果。import org.springframework.http.MediaType; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RestController; import reactor.core.publisher.Flux; import java.time.Duration; RestController public class StreamAiController { GetMapping(value /ai/stream-simple, produces MediaType.TEXT_EVENT_STREAM_VALUE) public FluxString streamAiResponseSimple(RequestParam String question) { // 模拟一个AI生成的长回答 String fullAnswer 你好这是一个关于[ question ]的流式回答。我将逐词展示流式输出的效果。; String[] words fullAnswer.split(); // 创建一个Flux每隔200毫秒发射一个词 return Flux.fromArray(words) .delayElements(Duration.ofMillis(200)) .map(word - data: word \n\n); // 手动拼接SSE格式 } }代码解析GetMapping(produces MediaType.TEXT_EVENT_STREAM_VALUE)这是关键注解声明该接口 produces “text/event-stream” 类型的响应。返回值FluxString每个String元素都将被作为一个独立的数据块发送。注意这里我们手动拼接了data: word \n\n这是因为Spring只会对返回类型为FluxServerSentEvent或设置了特定编码器的对象进行自动封装。对于FluxString它会直接将字符串写入响应体。我们手动加上SSE前缀和双换行让前端能正确解析。delayElements用于模拟AI生成每个词时的延迟让流式效果更明显。测试启动应用在浏览器中访问http://localhost:8080/ai/stream-simple?questionSpringBoot。你应该会看到字符一个接一个地显示出来。你也可以使用curl命令测试curl -N http://localhost:8080/ai/stream-simple?questiontest。-N参数用于禁用缓冲实时显示数据。3.3 集成真实AI大模型以OpenAI API为例上面的例子是模拟数据。现在我们来集成一个真实的流式AI接口例如OpenAI的Chat Completions API同样适用于国内大部分兼容OpenAI格式的大模型平台。首先添加一个HTTP客户端依赖如Spring的WebClient它是响应式的与WebFlux天生契合!-- WebClient已包含在spring-boot-starter-webflux中无需单独引入 --然后创建一个Service来处理AI调用import org.springframework.http.HttpHeaders; import org.springframework.http.MediaType; import org.springframework.stereotype.Service; import org.springframework.web.reactive.function.client.WebClient; import reactor.core.publisher.Flux; import java.nio.charset.StandardCharsets; Service public class OpenAiStreamService { private final WebClient webClient; // 假设你的API Key和Base URL通过配置注入 public OpenAiStreamService(Value(${ai.openai.api-key}) String apiKey, Value(${ai.openai.base-url}) String baseUrl) { this.webClient WebClient.builder() .baseUrl(baseUrl) .defaultHeader(HttpHeaders.AUTHORIZATION, Bearer apiKey) .defaultHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE) .build(); } public FluxString streamChatCompletion(String prompt) { // 构建请求体注意要设置 stream: true MapString, Object requestBody Map.of( model, gpt-3.5-turbo, messages, List.of(Map.of(role, user, content, prompt)), stream, true, temperature, 0.7 ); return webClient.post() .uri(/v1/chat/completions) .bodyValue(requestBody) .accept(MediaType.TEXT_EVENT_STREAM) // 接受流式响应 .retrieve() .bodyToFlux(String.class) // 将响应体转换为FluxString .takeUntil(s - s.contains([DONE])) // 遇到结束标志停止 .filter(s - s.startsWith(data: ) !s.contains([DONE])) // 过滤有效数据行 .map(s - { // 解析SSE格式的data行提取JSON String json s.substring(6).trim(); // 去掉data: if (json.isEmpty()) return ; // 这里需要解析JSON提取 choices[0].delta.content // 使用Jackson或JsonPath简单演示用字符串处理 if (json.contains(\content\:)) { // 简化处理实际应用应用JSON解析器 int start json.indexOf(\content\:\) 11; int end json.indexOf(\, start); if (start 10 end start) { return json.substring(start, end); } } return ; }) .filter(content - !content.isEmpty()); } }接着改造我们的Controller使用这个Serviceimport org.springframework.http.MediaType; import org.springframework.web.bind.annotation.*; import reactor.core.publisher.Flux; RestController RequestMapping(/ai) public class AiStreamController { private final OpenAiStreamService aiStreamService; public AiStreamController(OpenAiStreamService aiStreamService) { this.aiStreamService aiStreamService; } GetMapping(value /chat/stream, produces MediaType.TEXT_EVENT_STREAM_VALUE) public FluxServerSentEventString streamChat(RequestParam String message) { return aiStreamService.streamChatCompletion(message) .map(content - ServerSentEvent.builder(content).build()) .onErrorResume(e - { // 错误处理返回一个错误事件 return Flux.just(ServerSentEvent.builder(【服务异常】 e.getMessage()).build()); }); } }关键改进使用ServerSentEventT作为Flux的元素类型。Spring会自动将其转换为正确的SSE格式包括data:字段。你还可以方便地设置event,id,retry等属性。在Service中我们使用WebClient以流式方式(bodyToFlux)调用AI接口并将返回的字节流实时转换为FluxString。这里的数据处理逻辑JSON解析需要根据AI接口返回的具体格式进行调整。OpenAI的流式返回是多个SSE格式的JSON片段。增加了简单的错误处理(onErrorResume)确保即使后端调用失败前端也能收到一个明确的错误信息事件而不是连接突然中断。3.4 前端对接示例一个简单的前端HTML页面使用EventSourceAPI来接收SSE流!DOCTYPE html html head titleAI流式对话测试/title /head body input typetext idquestion placeholder输入你的问题... / button onclickstartStream()发送/button button onclickstopStream()停止/button br/br/ div idresponse stylewhite-space: pre-wrap; border:1px solid #ccc; min-height:200px; padding:10px;/div script let eventSource null; function startStream() { const question document.getElementById(question).value; if (!question) return; const responseDiv document.getElementById(response); responseDiv.innerHTML ; // 清空之前的内容 // 关闭之前的连接如果有 if (eventSource) { eventSource.close(); } // 创建新的EventSource连接 const url http://localhost:8080/ai/chat/stream?message${encodeURIComponent(question)}; eventSource new EventSource(url); eventSource.onmessage function(event) { // 接收到的数据是ServerSentEvent的data部分 responseDiv.innerHTML event.data; // 自动滚动到底部 responseDiv.scrollTop responseDiv.scrollHeight; }; eventSource.onerror function(error) { console.error(EventSource failed:, error); responseDiv.innerHTML \n\n【连接已关闭或出错】; eventSource.close(); }; } function stopStream() { if (eventSource) { eventSource.close(); eventSource null; document.getElementById(response).innerHTML \n\n【已手动停止】; } } /script /body /html4. 高级话题与生产级考量4.1 连接管理、超时与心跳在生产环境中SSE连接可能因为网络不稳定、代理超时、负载均衡器空闲连接断开等原因中断。虽然浏览器会自动重连但我们需要在服务端做好连接管理和保活。1. 设置响应超时与心跳Spring WebFlux默认没有为SSE连接设置特定的超时。长时间空闲的连接可能被中间件如Nginx、云负载均衡器切断。解决方案是定期发送“心跳”事件注释或空事件来保持连接活跃。public FluxServerSentEventString streamChatWithHeartbeat(RequestParam String message) { // 主要的AI响应流 FluxServerSentEventString dataStream aiStreamService.streamChatCompletion(message) .map(content - ServerSentEvent.builder(content).build()); // 心跳流每15秒发送一个注释行: heartbeat\n\n FluxServerSentEventString heartbeatStream Flux.interval(Duration.ofSeconds(15)) .map(tick - ServerSentEvent.Stringbuilder().comment(heartbeat).build()); // 合并两个流如果AI流结束整个流就结束 return dataStream .mergeWith(heartbeatStream) .timeout(Duration.ofMinutes(5)) // 设置总超时时间 .onErrorResume(TimeoutException.class, e - { return Flux.just(ServerSentEvent.builder(【会话超时】).build()); }); }2. 连接标识与状态管理对于需要关联用户会话的场景你需要在连接建立时生成一个唯一ID如UUID并将其与用户信息关联起来存储到ConcurrentHashMap或Redis中。当连接断开onComplete或onError信号时清理资源。Component public class SseConnectionManager { private final ConcurrentHashMapString, SinkServerSentEvent? connections new ConcurrentHashMap(); public void addConnection(String connectionId, SinkServerSentEvent? sink) { connections.put(connectionId, sink); } public void removeConnection(String connectionId) { connections.remove(connectionId); } public OptionalSinkServerSentEvent? getSink(String connectionId) { return Optional.ofNullable(connections.get(connectionId)); } } // 在Controller中使用Sinks.many()来创建可主动推送的流 GetMapping(value /connect, produces MediaType.TEXT_EVENT_STREAM_VALUE) public FluxServerSentEventString connect(RequestParam String userId) { String connectionId userId _ UUID.randomUUID(); Sinks.ManyServerSentEventString sink Sinks.many().unicast().onBackpressureBuffer(); connectionManager.addConnection(connectionId, sink.asFlux()); return sink.asFlux() .doOnCancel(() - connectionManager.removeConnection(connectionId)) .doOnError(e - connectionManager.removeConnection(connectionId)); }4.2 背压Backpressure处理背压是响应式编程中的核心概念指的是下游消费者处理速度跟不上上游生产者发射数据的速度时需要一种反馈机制来调节上游的发射速率。在AI流式输出中如果网络状况差或前端页面卡顿可能导致数据积压。Project Reactor的Flux内置了背压支持。WebClient在接收流式响应时默认会使用反压信号来告诉远程服务器“慢一点”。在大多数情况下你不需要手动处理。但如果你自己生成流如从数据库分页读取需要注意// 错误示例快速发射无视下游 Flux.range(1, 1000000) .delayElements(Duration.ofMillis(1)) // 延迟很小发射很快 .map(i - data: Item i \n\n); // 更好做法使用limitRate或onBackpressureBuffer等操作符调节 Flux.range(1, 1000000) .delayElements(Duration.ofMillis(1)) .limitRate(100) // 每次向下游请求最多100个元素 .map(i - data: Item i \n\n);对于SSE浏览器客户端的EventSource实现通常会处理背压但如果你的自定义客户端处理不过来服务端积压的数据可能会消耗大量内存。监控Flux的缓冲区大小是必要的。4.3 安全与权限控制整合Spring Security将SSE端点暴露在外网必须考虑安全问题。你需要确保只有认证的用户才能建立连接并且连接只能访问该用户权限内的数据。假设你使用了Spring Security可以像保护普通API一样保护SSE端点import org.springframework.context.annotation.Bean; import org.springframework.security.config.annotation.web.reactive.EnableWebFluxSecurity; import org.springframework.security.config.web.server.ServerHttpSecurity; import org.springframework.security.web.server.SecurityWebFilterChain; EnableWebFluxSecurity public class SecurityConfig { Bean public SecurityWebFilterChain springSecurityFilterChain(ServerHttpSecurity http) { http .authorizeExchange() .pathMatchers(/ai/chat/stream).authenticated() // SSE端点需要认证 .anyExchange().permitAll() .and() .httpBasic() // 可以使用HTTP Basic或更常见的JWT、OAuth2 .and() .csrf().disable(); // 对于SSE/WebSocket通常需要禁用CSRF return http.build(); } }在Controller中你可以通过AuthenticationPrincipal注入当前用户信息用于过滤数据流GetMapping(value /chat/stream, produces MediaType.TEXT_EVENT_STREAM_VALUE) public FluxServerSentEventString streamChat( RequestParam String message, AuthenticationPrincipal Jwt jwt) { // 假设使用JWT String userId jwt.getSubject(); // 确保AI回答的内容只与当前用户相关 return aiStreamService.streamChatCompletionForUser(message, userId) .map(content - ServerSentEvent.builder(content).build()); }一个重要陷阱在yudao-cloud等开源项目中曾出现过Flux流式输出与Spring Security的权限控制上下文丢失的问题。这是因为响应式编程中安全上下文可能不在同一个线程上传播。解决方案是使用ReactiveSecurityContextHolder来获取上下文return ReactiveSecurityContextHolder.getContext() .map(SecurityContext::getAuthentication) .flatMapMany(auth - { String userId auth.getName(); // 在正确的上下文中调用你的业务流 return aiStreamService.streamChatCompletionForUser(message, userId); }) .map(content - ServerSentEvent.builder(content).build());4.4 性能监控与问题排查监控指标活跃连接数监控当前服务器维持的SSE连接数量评估内存和文件描述符压力。数据发送速率与延迟监控从AI模型获取到第一个Token的延迟TTFT和后续Token的生成速率TPS。错误率连接异常断开、AI服务调用失败的比例。问题排查清单前端收不到数据检查浏览器控制台Network标签查看SSE连接是否成功建立状态码200查看响应头是否为text/event-stream。使用curl -N命令测试服务端是否正常输出。连接频繁断开检查Nginx等代理服务器的proxy_read_timeout、keepalive_timeout配置确保其大于你的心跳间隔和预期会话时长。在服务端添加心跳保活。内存泄漏确保Flux流在完成或出错时相关的资源如WebClient连接、数据库连接被正确释放。使用doOnCancel、doFinally回调进行清理。监控JVM堆内存和直接内存使用情况。数据乱码或格式错误确保服务端返回的数据严格遵循SSE格式以data:开头以两个换行符结尾。中文字符确保UTF-8编码。前端EventSource对格式要求很严格。5. 常见问题与避坑指南在实际开发和线上运维中我踩过不少坑这里总结几个最典型的1. 响应头被修改或压缩问题某些网关、代理或安全中间件可能会修改响应头或者对响应体进行压缩如gzip这会破坏SSE流。 解决明确配置你的反向代理如Nginx对于特定路径如/ai/stream/*不进行缓冲和压缩。location /ai/stream/ { proxy_pass http://backend; proxy_buffering off; # 关键关闭代理缓冲 proxy_cache off; proxy_set_header Connection ; proxy_http_version 1.1; chunked_transfer_encoding off; # 对于SSE有时也需要关闭分块编码 proxy_read_timeout 3600s; # 设置长的读超时 }2.Flux流提前结束或卡住问题在复杂的业务链中某个操作符可能抛出未处理的异常或者WebClient调用阻塞导致整个流静默结束或挂起。 解决为每个可能出错的步骤添加onErrorResume或onErrorContinue提供降级处理如返回错误信息事件。使用timeout操作符为异步操作设置超时避免无限期等待。使用doOnError记录日志便于排查。3. 多实例部署下的连接状态问题在Kubernetes或负载均衡后有多台服务实例时用户的SSE连接可能连接到A实例但后续的业务触发消息如后台任务完成通知可能来自B实例导致消息无法推送。 解决不要将连接状态保存在单机内存中。使用Redis的Pub/Sub功能或消息队列如Kafka、RabbitMQ作为中间层。当用户连接建立时订阅一个以其用户ID或会话ID命名的频道。任何需要向该用户推送消息的服务都向对应的频道发布消息。SSE服务实例收到消息后再通过本地保存的连接句柄推送给前端。4. 浏览器兼容性与连接限制问题浏览器对同一个域名下的并发HTTP连接数有限制通常是6个。如果页面同时打开多个SSE连接可能会阻塞其他资源的加载。 解决对于同一个应用尽量复用同一个SSE连接通过不同的事件类型event字段来区分不同的业务流。如果必须多个流考虑使用HTTP/2它支持多路复用可以缓解连接数限制。同时在连接不再需要时务必调用eventSource.close()主动关闭。5. 处理“Java: OutOfMemoryError: Insufficient memory”问题在流式处理大量数据时如果背压处理不当或者Flux中缓存了过多的未发送元素可能导致内存溢出。 解决使用onBackpressureBuffer时务必指定一个合理的缓冲区大小和溢出策略如ERROR或DROP。对于从数据库等源头读取大量数据并流式输出的场景使用分页查询并将每页数据作为一个Flux元素发射而不是一次性加载到内存。监控JVM内存特别是直接内存Direct Memory因为NettyWebFlux底层会大量使用。确保JVM参数中-XX:MaxDirectMemorySize设置得足够大。流式输出为Java AI应用带来了更流畅、更实时的用户体验。从简单的Flux.interval模拟到集成复杂的AI大模型再到处理生产环境下的连接、安全、性能问题每一步都需要仔细考量。核心在于理解SSE协议的本质并善用Spring WebFlux提供的响应式工具。希望这篇从原理到实战的剖析能帮助你在下一个项目中游刃有余地实现优雅的流式交互。