SpringBoot + ResponseBodyEmitter 实时异步流式推送，优雅！

ChatGPT 的火爆，让流式输出技术迅速走进大众视野。在那段时间里，许多热爱钻研技术的小伙伴纷纷开始学习和实践 SSE 异步处理。

我当时也写过相关文章，今天，咱们换一种更为简便的方式来实现流式输出，那就是 ​​ResponseBodyEmitter​​。

其实，​​ResponseBodyEmitter​​ 并非新技术，早在 Spring Framework 4.2 版本就已被引入。直到最近，我们在开发一个滚动日志输出功能时，才深入了解到它的强大之处。

ResponseBodyEmitter 的作用

相较于 SSE 技术，​​ResponseBodyEmitter​​ 更加简单易用。它主要用于处理异步的 HTTP 响应，其核心优势在于 ​​允许逐步将数据发送到客户端，而非一次性发送所有内容​​。这一特性使得它在需要长时间处理或进行流式传输的场景中表现出色。需要注意的是，​​ResponseBodyEmitter​​ 本质上是一个接口。

使用场景

1. 长轮询：服务器在有数据时会立即响应客户端请求，若暂无数据，则保持连接开放，等待数据到来。
2. **服务器推送事件 (SSE)**：服务器能够持续不断地向客户端推送各类事件，实现实时交互。
3. 流式传输：可逐步发送大量数据，像文件下载或者实时数据流传输等场景都适用。
4. 异步处理：在处理耗时任务时，能逐步返回处理结果，避免客户端长时间等待，提升用户体验。

业务场景举例

在实际业务中，​​ResponseBodyEmitter​​ 有着广泛的应用，比如进度条的实时更新、实时聊天功能、股票价格的实时更新、系统日志的流式输出以及 AI 的流式响应等。

实时日志流实战

接下来，我们通过一个简单的实时日志流功能，来深入了解 ​​ResponseBodyEmitter​​ 的使用。假设我们有一个应用程序，需要实时查看服务器的日志，以便快速定位和解决问题。

创建控制器

首先，我们在 Spring Boot 应用中创建一个控制器，借助 ​​ResponseBodyEmitter​​ 实现实时日志流。

import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitter;
@RestController
@RequestMapping("/api/log")
publicclass LogController {@GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)public ResponseBodyEmitter streamLogs() {ResponseBodyEmitter emitter = new ResponseBodyEmitter();// 开启异步线程处理数据并发送new Thread(() -> {try {while (true) {String logEntry = getLatestLogEntry();if (logEntry != null) {emitter.send(logEntry);}// 每秒检查一次日志更新Thread.sleep(1000); }} catch (Exception e) {// 出现异常时结束响应并传递错误信息emitter.completeWithError(e); }}).start();return emitter;}private String getLatestLogEntry() {// 模拟从日志文件中获取最新日志条目return"2025-02-12 12:00:00 - INFO: User logged in successfully.";}
}

运行效果

当我们启动这个应用程序，并访问 ​​/api/log/stream​​ 路径时，就能看到一个实时更新的日志流。服务器会每秒向客户端推送一条新的日志条目，客户端会将其显示在页面上，效果如下：

运行效果

ResponseBodyEmitter 的核心方法

• ​​send(Object data)​​：向客户端发送数据，该方法可以多次调用，实现数据的逐步发送。
• ​​complete()​​：用于结束响应流，表示数据已经全部发送完毕。
• ​​onTimeout(Runnable callback)​​：设置超时回调函数，当连接超时时，会执行该回调。
• ​​onCompletion(Runnable callback)​​：设置完成回调函数，当数据发送完成后，会执行该回调。

ResponseBodyEmitter 工作原理

异步数据生成与推送

在传统的 HTTP 请求 - 响应模式中，服务器通常需要等待整个响应数据生成完成后，才会将其一次性发送给客户端。而 ​​ResponseBodyEmitter​​ 打破了这种模式，它允许服务端在任务执行过程中异步地生成响应数据。

当有部分数据准备好时，就可以立即调用 ​​send()​​ 方法将这些数据推送给客户端，而无需等待整个任务完成。这就好比一场接力赛，每完成一段赛程（生成一部分数据），就马上将接力棒（数据）传递给客户端，大大提高了数据传输的实时性。

分块传输机制

​​ResponseBodyEmitter​​ 采用了 HTTP 的分块编码（Chunked Encoding）方式来传输数据。在传统的 HTTP 响应中，通常需要在响应头中明确指定 ​​Content-Length​​，表示整个响应数据的长度。但在分块传输中，服务器不会提前设置 ​​Content-Length​​，而是将数据分成多个独立的块，每个块都有自己的长度标识。

客户端在接收到数据块后，可以立即对其进行处理，而不必等待整个响应数据接收完毕。这种方式使得数据可以边生成边传输，减少了客户端的等待时间，提高了用户体验。

连接生命周期管理

为了确保资源的合理使用，​​ResponseBodyEmitter​​ 提供了对连接生命周期的有效管理。当所有数据都发送完毕后，需要调用 ​​complete()​​ 方法来明确告知客户端响应结束，关闭连接。如果在数据传输过程中出现异常，可以调用 ​​completeWithError()​​ 方法，结束响应并向客户端传递错误信息。

这样可以避免连接长时间保持开放，造成资源浪费。

注意事项

1. 客户端支持：虽然大多数浏览器和 HTTP 客户端库都支持分块传输，但某些老旧的客户端可能存在兼容性问题。
2. 超时设置：为避免长连接长时间占用资源，可以为 ResponseBodyEmitter 设置超时时间，示例代码如下：

emitter.onTimeout(() -> emitter.complete());

1. 线程安全：ResponseBodyEmitter 的 send() 方法是线程安全的，但在使用时需要注意控制任务线程的生命周期，避免出现资源泄漏。
2. 连接关闭：务必确保在任务结束时调用 complete() 或 completeWithError() 方法，否则可能导致连接无法正常关闭，造成资源浪费。

与 Streaming 和 SSE 的对比

• Streaming：直接通过 OutputStream 向客户端写入数据，灵活性较高，但需要手动处理流的关闭，增加了开发的复杂度。
• Server-Sent Events (SSE)：基于 text/event-stream 协议，适用于服务端事件推送场景，但要求客户端支持 SSE 协议。
• ResponseBodyEmitter：通用性更强，适用于任何支持 HTTP 的客户端，并且易于与 Spring 框架集成，是一种更为便捷的流式传输解决方案。

在处理类似 AI 这种响应式的流式输出场景时，相较于 SSE，​​ResponseBodyEmitter​​ 作为 Spring 提供的轻量级流式传输解决方案，在 HTTP 协议兼容性方面表现更优。

小结

​​ResponseBodyEmitter​​ 是 Spring 框架提供的轻量级流式传输解决方案，它能够显著提升高并发和实时性场景下的用户体验。通过 ​​ResponseBodyEmitter​​，我们可以轻松实现服务器向客户端的实时数据推送。

无论是进度条的实时更新、实时聊天、股票价格的实时监控还是系统日志的流式输出，​​ResponseBodyEmitter​​ 都能帮助我们构建更加动态和互动的应用程序。