Spring 中 SSE 流式输出的多种实现方式详解

择要：Server-Sent Events (SSE) 是一种基于 HTTP 的服务器推送技能，在 AI 大模子流式相应场景中广泛应用。本文联合现实代码，具体先容 Spring 框架中实现 SSE 的三种主流方式。

什么是 SSE？

SSE（Server-Sent Events）是一种允许服务器向欣赏器推送实时数据的 Web 尺度技能。它基于 HTTP 协议，使用 text/event-stream 内容范例，具有以下特点：

• 单向通讯：仅支持服务器向客户端推送数据
  
• 基于 HTTP：无需特别协议，兼容现有根本办法
  
• 主动重连：内置断线重连机制
  
• 变乱 ID：支持变乱标识和规复
  

与 WebSocket 相比，SSE 更简单易用，得当服务器推送场景（如 AI 对话流式输出、实时关照、数据监控等）。
实现方式一：SseEmitter（Spring MVC 原生支持）

SseEmitter 是 Spring Web MVC 提供的 SSE 支持类，常用于处理处罚大模子的流式相应。
代码示例

1. @GetMapping("/streamEvents")
2. public SseEmitter streamServerEvents() {
3.     SseEmitter emitter = new SseEmitter(60_000L);
4.     Executors.newVirtualThreadPerTaskExecutor().submit(() -> {
5.         try {
6.             for (int i = 0; i < 50; i++) {
7.                 emitter.send("Event Data: " + i);
8.                 Thread.sleep(500);
9.             }
10.         } catch (IOException | InterruptedException e) {
11.             emitter.completeWithError(e);
12.         } finally {
13.             emitter.complete();
14.         }
15.     });
16.     return emitter;
17. }

复制代码

焦点要点

• 超时设置：new SseEmitter(60_000L) 设置 60 秒超时，克制毗连长时间占用
  
• 异步实行：使用假造线程（Java 21+）在配景发送数据，不壅闭主线程
  
• 非常处理处罚：通过 completeWithError() 处理处罚非常，complete() 正常竣事
  
• 实用场景：传统 Spring MVC 项目、简单流式推送
  

优缺点

长处缺点简单易用，无需额外依赖手动管理线程和生命周期兼容性好，Spring 4.2+ 支持错误处理处罚须要手动编码支持自界说变乱名称和 ID不得当复杂相应式场景实现方式二：StreamingResponseBody（机动控制相应流）

StreamingResponseBody 允许直接使用 OutputStream，得当自界说流式相应格式。
代码示例

1. @GetMapping("/streamResponse")
2. public ResponseEntity<StreamingResponseBody> streamChatResponse() {
3.    
4.     StreamingResponseBody body = outputStream -> {
5.         for (int i = 0; i < 50; i++) {
6.             String data = "Chunk: " + i + "\n";
7.             outputStream.write(data.getBytes(StandardCharsets.UTF_8));
8.             outputStream.flush();
9.             try {
10.                 Thread.sleep(500); // 模拟处理延迟
11.             } catch (InterruptedException e) {
12.                 throw new RuntimeException(e);
13.             }
14.         }
15.     };
17.     return ResponseEntity.ok()
18.             .header(HttpHeaders.CONTENT_TYPE, MediaType.TEXT_EVENT_STREAM_VALUE)
19.             .body(body);
20. }

复制代码

焦点要点

• 直接写流：通过 outputStream.write() 直接写入字节数据
  
• 手动革新：每次写入后调用 flush() 确保数据立即发送
  
• Content-Type：需手动设置 text/event-stream 标识 SSE 流
  
• 实用场景：须要风雅控制输特别式、文件流式下载
  

优缺点

长处缺点完全控制输特别式须要手动处理处罚全部细节得当大数据量传输不符合 SSE 尺度格式（需自行实现）可用于非 SSE 场景缺乏变乱 ID、重试等 SSE 特性留意事项

这种方式严酷来说不是尺度 SSE，而是平常的 HTTP 流式相应。如果要实现尺度 SSE，须要按照以下格式输出：

1. data: {"message": "Hello"}
2. id: 1
3. event: customEvent
4. retry: 3000

复制代码

实现方式三：Flux（相应式编程 - Spring WebFlux）

Flux 是 Spring WebFlux（基于 Project Reactor）提供的相应式范例，得当高并发的流式调用场景。
代码示例

1. @GetMapping(value = "/reactiveStream")
2. public Flux<String> reactiveDataStream() {
3.     return Flux.interval(Duration.ofSeconds(1))
4.             .take(20)
5.             .map(seq -> "Reactive Event: " + seq);
6. }

复制代码

焦点要点

• 声明式编程：通过链式调用界说数据流，无需手动管理线程
  
• 背压支持：主动处理处罚斲丧者速率慢于生产者的情况
  
• 非壅闭：基于变乱循环，资源占用更少
  
• 实用场景：高并发场景、相应式微服务、实时数据流处理处罚
  

进阶示例：联合大模子 API

1. @GetMapping("/aiStream")
2. public Flux<String> streamAIResponse(@RequestParam String question) {
3.     return webClient.post()
4.             .uri("https://api.openai.com/v1/chat/completions")
5.             .bodyValue(buildRequest(question))
6.             .retrieve()
7.             .bodyToFlux(String.class)
8.             .map(this::parseSSE);
9. }

复制代码

优缺点

长处缺点自然支持相应式流学习曲线陡峭主动背压和资源管理须要引入 WebFlux 依赖组合使用符丰富与传统 MVC 不兼容三种方式对比

特性SseEmitterStreamingResponseBodyFlux编程模子下令式下令式相应式线程管理手动手动主动SSE 尺度支持✅ 完备支持⚠️ 需手动实现✅ 完备支持背压支持❌❌✅实用框架Spring MVCSpring MVCSpring WebFlux学习资本低中高保举场景简单推送自界说流式输出高并发相应式应用现实应用：调用大模子流式 API

以下是一个完备的示例，展示怎样转发大模子的 SSE 相应：

1. @RequestMapping("/mockStreamResponse")
2. public String mockStreamResponse() {
3.     String requestBody = """
4.             {
5.                 "model": "qwen-plus",
6.                 "messages": [
7.                     {
8.                         "role": "system",
9.                         "content": "You are a professional technical writer."
10.                     },
11.                     {
12.                         "role": "user",
13.                         "content": "请简要说明Python在数据科学中的优势"
14.                     }
15.                 ],
16.                 "stream": true
17.             }
18.             """;
19.    
20.     HttpClient client = HttpClient.newHttpClient();
21.     HttpRequest request = HttpRequest.newBuilder()
22.             .uri(URI.create(API_URL))
23.             .header("Content-Type", "application/json")
24.             .header("Authorization", "Bearer " + API_KEY)
25.             .header("X-DashScope-SSE", "enable")
26.             .POST(HttpRequest.BodyPublishers.ofString(requestBody))
27.             .build();
29.     try {
30.         HttpResponse<String> response = client.send(
31.                 request, HttpResponse.BodyHandlers.ofString());
32.         return response.body();
33.     } catch (IOException | InterruptedException e) {
34.         throw new RuntimeException(e);
35.     }
36. }

复制代码

关键设置

• X-DashScope-SSE: enable：启用阿里云 DashScope 的 SSE 支持
  
• stream: true：哀求体中开启流式模式
  
• 直接返回原始 SSE 数据，由前端剖析
  

前端调用示例

1. // 方式一：使用 EventSource（推荐）
2. const eventSource = new EventSource('/streamEvents');
4. eventSource.onmessage = (event) => {
5.     console.log('Received:', event.data);
6. };
8. eventSource.onerror = (error) => {
9.     console.error('Connection error:', error);
10.     eventSource.close();
11. };
13. // 方式二：使用 fetch（更灵活）
14. const response = await fetch('/reactiveStream');
15. const reader = response.body.getReader();
17. while (true) {
18.     const { done, value } = await reader.read();
19.     if (done) break;
20.     console.log('Chunk:', new TextDecoder().decode(value));
21. }

复制代码

最佳实践

1. 公道设置超时时间

1. // 根据业务场景调整超时
2. SseEmitter emitter = new SseEmitter(5 * 60 * 1000L); // 5分钟

复制代码

2. 处理处罚客户端断开毗连

1. emitter.onCompletion(() -> {
2.     log.info("SSE connection completed");
3. });
5. emitter.onTimeout(() -> {
6.     log.warn("SSE connection timeout");
7. });

复制代码

3. 使用结构化数据

1. // 发送 JSON 格式数据
2. emitter.send(SseEmitter.event()
3.     .name("message")
4.     .data(Map.of("content", "Hello", "timestamp", System.currentTimeMillis())));

复制代码

4. 监控和日志

1. emitter.onError((ex) -> {
2.     log.error("SSE error: {}", ex.getMessage(), ex);
3.     metrics.increment("sse.errors");
4. });

复制代码

总结

Spring  提供了多种方式实现 SSE 流式输出，选择时应思量：

• 项目技能栈：MVC 照旧 WebFlux
  
• 功能需求：是否须要背压、变乱 ID 等特性
  
• 团队熟悉度：相应式编程的学习资本
  
• 性能要求：高并发场景保举 WebFlux + Flux
  

在 Spring 项目中，保举使用 Flux 或 SseEmitter，具体选择取决于你使用的是 Spring WebFlux 照旧 Spring Web MVC。
参考资源：

• Spring Framework 官方文档 - SSE
  
• MDN - Server-Sent Events
  
• Project Reactor 文档