SpringBoot手动实现流式输出方案整理以及SSE规范输出详解

背景：

最近做流式输出时，一直使用python实现的，应需求方的要求，需要通过java应用做一次封装并在java侧完成系统鉴权、模型鉴权等功能后才能真正去调用智能体应用，基于此调研java实现流式输出的几种方式，并完成与python服务对接的方案。

方案：

• 使用Servlet原生API实现流式输出
• 使用ResponseBodyEmitter实现异步流式输出
• 使用SseEmitter实现服务器发送事件(SSE)
• 使用WebFlux实现响应式流式输出
• 使用Spring MVC的StreamingResponseBody
• websockt

说一下我的业务场景，我原本的前后端适配已经按照SSE规范完成了功能，因此新写接口时也采用SSE规范，避免同一个系统中前端出现多种方式的调用，而且我的python微服务采用SSE规范，当时第一反应采用Feign去调用接口返回即可，但是使用后发现Openfeign支持这种调用不友好，因此接口对接这里采用的是WebClient。因此本文着重说一下SSE规范调用

一、SSE是什么

SSE (Server-Sent Events) 是一种基于HTTP的服务器向客户端推送数据的Web技术规范，它允许服务器单向地向客户端发送事件流。以下是SSE规范的全面解析：

1.基本概念

SSE是HTML5标准的一部分，主要特点包括：

• 单向通信：仅服务器→客户端方向

• 基于HTTP：使用普通HTTP连接

• 文本协议：事件以纯文本格式传输

• 自动重连：内置连接恢复机制

• 简单易用：比WebSocket更轻量级

2. 协议格式

SSE事件流是一个UTF-8编码的文本流，包含以下字段（每个字段以\n结尾）：

event: message\n
id: 123\n
retry: 5000\n
data: {\n
data: "name": "John",\n
data: "age": 30\n
data: }\n\n

• data: 有效载荷内容（可多行，每行需加"data: "前缀）

• event: 自定义事件类型（默认"message"）

• id: 事件ID（用于断线重连时定位）

• retry: 重连时间(毫秒)

服务器响为：

Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive 

3.客户端API

浏览器端JavaScript使用EventSource接口：

const eventSource = new EventSource('/sse-endpoint');// 监听默认事件
eventSource.onmessage = (e) => {console.log('Message:', e.data);
};// 监听自定义事件
eventSource.addEventListener('customEvent', (e) => {console.log('Custom event:', e.data);
});// 错误处理
eventSource.onerror = (e) => {console.error('SSE error:', e);
};

4.与相关技术的对比

特性	SSE	WebSocket	Long Polling
方向	单向(服务器→客户端)	双向	单向(轮询)
协议	HTTP	WS/WSS	HTTP
连接管理	自动重连	需手动处理	每次请求新建连接
数据格式	文本	二进制/文本	文本
复杂度	低	中	低

5. 适用场景

SSE特别适合：

• 实时通知（新闻、股价、天气）

• 日志流监控

• 进度报告（文件处理、任务执行）

• 社交媒体动态更新

• 需要简单实时功能但不需要双向通信的场景

虽然WebSocket更强大，但SSE仍有很多优势：

• 更简单的实现

• 自动利用HTTP/2的多路复用

• 不需要额外的协议升级

• 被所有现代浏览器支持（IE除外）

二、WebClient ‌

1.概念

WebClient ‌是 Spring Framework 5中引入的一个基于响应式编程模型的 HTTP客户端 ，主要用于执行HTTP请求。相比传统的 RestTemplate ，WebClient采用了 Reactor库 ，支持非阻塞式（异步）调用，能够充分利用多核CPU资源，特别适合高并发场景。

• WebClient 是 线程安全 的，适合作为单例Bean复用。

• 底层使用 连接池（默认基于Reactor Netty），减少重复创建连接的开销。

2.与OpenFeign比较

推荐方案：优先使用WebClient + Service分层架构
原因：WebClient原生支持响应式流处理，更适合SSE场景，而OpenFeign更适合普通REST调用

备选方案：使用OpenFeign（需要特殊配置）
注意：需要Spring Cloud 2020.0.3+版本和响应式Feign支持

特性	WebClient方案	OpenFeign方案
响应式支持	✅ 原生支持	⚠️ 需要特殊配置
代码复杂度	简单	较复杂
维护性	高	中
性能	高（非阻塞IO）	中等
连接池管理	自动	需要手动配置
适合场景	高并发流式处理	简单接口调用

3.与传统RestTemplate对比

特性	WebClient（推荐）	RestTemplate（过时）
协议支持	HTTP/1.1, HTTP/2, WebSocket	仅HTTP/1.1
编程模型	响应式（Reactive）	同步阻塞
性能	非阻塞IO，高并发	线程池阻塞模型
配置方式	Builder模式，灵活扩展	直接实例化

优势

• 响应式集成：完美兼容Spring WebFlux。

• 函数式API：链式调用更清晰。

• 更好的性能：基于Project Reactor和Netty。

三、代码实现

1.基础实现

@RestController
public class SseController {@GetMapping("/sse-stream")public SseEmitter streamSse() {SseEmitter emitter = new SseEmitter(30_000L); // 30秒超时CompletableFuture.runAsync(() -> {try {for (int i = 0; i < 100; i++) {SseEmitter.SseEventBuilder event = SseEmitter.event().data("SSE Event " + i).id(String.valueOf(i)).name("sse-event");emitter.send(event);Thread.sleep(100);}emitter.complete();} catch (Exception e) {emitter.completeWithError(e);}});return emitter;}
}

2.业务进阶

2.1 依赖配置：

在pom.xml中添加必要依赖：

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId>
</dependency>

2.2 WebClient配置：

• 使用WebClient创建HTTP客户端，支持响应式流处理

• 配置第三方SSE接口地址和必要的请求头（如认证信息）

    WebClient配置类

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.reactive.function.client.WebClient;@Configuration
public class WebClientConfig {@Beanpublic WebClient webClient() {return WebClient.builder().baseUrl("https://api.example.com").build();}
}

2.3Service层实现

@Service
public class WebClientSseService {@Autowiredprivate WebClient webClient;public Flux<String> streamEvents() {System.out.println("前置校验。。。。");Flux<String> resFlux = null;try{resFlux = webClient.get().uri("/stream").accept(MediaType.TEXT_EVENT_STREAM).retrieve().bodyToFlux(String.class).map(data -> {// 处理原始SSE数据#if (data.startsWith("data:")) {#return data.substring(5).trim();#}return data;});}catch (Exception exception){resFlux = Flux.just("{'status': 'Error', 'message': '"+exception.getMessage()+"'}");}return resFlux;}
}

2.4 Controller

// application-web模块
@RestController
public class DataStreamController {@PostMapping(value = "/stream",consumes = MediaType.MULTIPART_FORM_DATA_VALUE,produces = MediaType.TEXT_EVENT_STREAM_VALUE)public Flux<String> provideStream(@RequestBody StreamRequest request) {return dataProcessor.streamEvents(request);}@PostMapping(value = "/stream",produces = MediaType.TEXT_EVENT_STREAM_VALUE)public Flux<String> provideStream(@RequestParam(name = "file", required = false) MultipartFile file, @RequestParam Map<String, Object> jsonObject) {return dataProcessor.streamEvents(file, jsonObject);}
}

解释一下这两个参数：

consumes = MULTIPART_FORM_DATA_VALUE,

produces = TEXT_EVENT_STREAM_VALUE

consumes、produces 两个参数的作用与区别

参数	作用	示例值
consumes = MULTIPART_FORM_DATA_VALUE	声明接口接收的请求内容类型（客户端→服务端）	multipart/form-data
produces = TEXT_EVENT_STREAM_VALUE	声明接口返回的响应内容类型（服务端→客户端）	text/event-stream

为什么需要同时声明？

1. 输入输出分离原则：

   • 输入（consumes）：处理文件上传需要 multipart/form-data

   • 输出（produces）：SSE流式响应需要 text/event-stream

2. HTTP协议规范：

POST /upload HTTP/1.1
Content-Type: multipart/form-data  ← 对应consumes
Accept: text/event-stream         ← 对应produces

内容类型对照速查表

场景	客户端设置	服务端声明
文件上传+JSON响应	Content-Type: multipart/form-data	consumes = MULTIPART_FORM_DATA_VALUE
文件上传+SSE流响应	Accept: text/event-stream	produces = TEXT_EVENT_STREAM_VALUE
JSON上传+SSE流响应	Content-Type: application/json	consumes = APPLICATION_JSON_VALUE

根据需要自由选择。

2.5 这里对webclient做个扩展

如果上传的是文件可以用这个方式写body的内容

.contentType(MediaType.MULTIPART_FORM_DATA)

.body(BodyInserters.fromMultipartData(formData))

如果不同的json类型的body请求体可以这么写

.body(BodyInserters.fromValue(res)) 

注意这块的细节，我就是在这里写绕了很多

四、其他方案实现

1. 使用Servlet原生API实现流式输出

@RestController
public class StreamingController {@GetMapping("/stream1")public void stream1(HttpServletResponse response) throws IOException {response.setContentType("text/plain;charset=UTF-8");try (PrintWriter writer = response.getWriter()) {for (int i = 0; i < 100; i++) {writer.write("Data line " + i + "\n");writer.flush(); // 手动刷新缓冲区Thread.sleep(100); // 模拟延迟}} catch (InterruptedException e) {Thread.currentThread().interrupt();}}
}

2. 使用ResponseBodyEmitter实现异步流式输出

@RestController
public class StreamingController {@GetMapping("/stream2")public ResponseBodyEmitter stream2() {ResponseBodyEmitter emitter = new ResponseBodyEmitter();CompletableFuture.runAsync(() -> {try {for (int i = 0; i < 100; i++) {emitter.send("Data line " + i + "\n");Thread.sleep(100);}emitter.complete();} catch (Exception e) {emitter.completeWithError(e);}});return emitter;}
}

3. 使用SseEmitter实现服务器发送事件(SSE)

@RestController
public class SseController {@GetMapping("/sse-stream")public SseEmitter streamSse() {SseEmitter emitter = new SseEmitter(30_000L); // 30秒超时CompletableFuture.runAsync(() -> {try {for (int i = 0; i < 100; i++) {SseEmitter.SseEventBuilder event = SseEmitter.event().data("SSE Event " + i).id(String.valueOf(i)).name("sse-event");emitter.send(event);Thread.sleep(100);}emitter.complete();} catch (Exception e) {emitter.completeWithError(e);}});return emitter;}
}

4. 使用WebFlux实现响应式流式输出

@RestController
@RequestMapping("/reactive")
public class ReactiveStreamingController {@GetMapping("/stream")public Flux<String> streamData() {return Flux.interval(Duration.ofMillis(100)).map(sequence -> "Reactive data " + sequence + "\n").take(100); // 限制输出数量}@GetMapping(value = "/stream-file", produces = MediaType.TEXT_EVENT_STREAM_VALUE)public Flux<String> streamLargeFile() {return Flux.using(() -> Files.lines(Paths.get("large-file.txt")),Flux::fromStream,Stream::close);}
}

5. 使用Spring MVC的StreamingResponseBody

@RestController
public class StreamingResponseBodyController {@GetMapping("/stream3")public StreamingResponseBody stream3() {return outputStream -> {Writer writer = new BufferedWriter(new OutputStreamWriter(outputStream));for (int i = 0; i < 100; i++) {writer.write("Streaming line " + i + "\n");writer.flush();Thread.sleep(100);}};}
}

五、WebClient​​​​​​​配置扩展

1. 添加默认请求头

@Bean
public WebClient thirdPartyWebClient() {return WebClient.builder().baseUrl("https://api.example.com").defaultHeader("Authorization", "Bearer token123").defaultHeader("Accept", "application/json").build();
}

2. 配置超时

import java.time.Duration;@Bean
public WebClient thirdPartyWebClient() {return WebClient.builder().baseUrl("https://api.example.com").clientConnector(new ReactorClientHttpConnector(HttpClient.create().responseTimeout(Duration.ofSeconds(5))).build();
}

3. 添加拦截器

@Bean
public WebClient thirdPartyWebClient() {return WebClient.builder().baseUrl("https://api.example.com").filter((request, next) -> {System.out.println("Sending request to: " + request.url());return next.exchange(request);}).build();
}

常见问题

1: 能否创建多个WebClient Bean

​​​​​​​可以，但需要指定不同的Bean名称： 

@Bean("paymentClient")
public WebClient paymentClient() {return WebClient.builder().baseUrl("https://payment.api").build();
}@Bean("weatherClient")
public WebClient weatherClient() {return WebClient.builder().baseUrl("https://weather.api").build();
}

2: 如何测试WebClient

使用MockWebServer（OkHttp）模拟API：

@SpringBootTest
class ApiTest {@Autowiredprivate WebClient webClient;@Testvoid testGetUser() {MockWebServer server = new MockWebServer();server.enqueue(new MockResponse().setBody("{\"name\":\"John\"}").addHeader("Content-Type", "application/json"));server.start();webClient = webClient.mutate().baseUrl(server.url("/").toString()).build();Mono<User> user = webClient.get().uri("/users/1").retrieve().bodyToMono(User.class);StepVerifier.create(user).expectNextMatches(u -> u.getName().equals("John")).verifyComplete();server.shutdown();}
}