当前位置：首页 > article >正文

SpringBoot + ResponseBodyEmitter 实时异步流式推送

article 2025/3/12 1:57:18

ResponseBodyEmitter 是 Spring Framework 提供的一个类，用于处理需要长时间运行的异步响应场景。它允许你在服务器端逐步发送数据到客户端，而不是一次性返回所有内容。这对于实时更新的应用场景非常有用，比如股票报价、聊天应用、社交媒体更新等。

`ResponseBodyEmitter` 的作用

异步响应：支持在HTTP请求生命周期内多次向客户端发送数据。
流式传输：适合需要持续传输大量数据的场景，如视频流、实时日志输出等。
非阻塞操作：允许后台线程执行耗时操作，并将结果逐步发送给客户端，而不阻塞主线程。

主要API 方法

以下是 ResponseBodyEmitter 中常用的一些方法：

send(Object data, MediaType mediaType)：发送数据到客户端。可以指定数据类型（MediaType），这样客户端就能正确解析收到的数据。例如，你可以发送字符串、字节数组或任何实现了序列化接口的对象。
```
emitter.send("Hello World", MediaType.TEXT_PLAIN);
```
complete()：当所有数据都已发送完毕时调用此方法，通知Spring框架响应已完成。
```
emitter.complete();
```
completeWithError(Throwable t)：如果在处理过程中发生错误，则可以使用此方法完成响应并附带错误信息。
```
emitter.completeWithError(new RuntimeException("An error occurred"));
```
setTimeout(long timeout)：设置超时时间，在该时间内如果没有数据发送到客户端，连接将被关闭。默认值为30秒。
```
emitter.setTimeout(60000); // 设置超时时间为60秒
```

使用示例

以下是一个简单的例子，展示了如何使用 ResponseBodyEmitter 实现一个异步数据推送服务：

import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitter;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@RestController
public class AsyncController {

    private final ExecutorService executor = Executors.newCachedThreadPool();

    @GetMapping(value = "/stream-events", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public ResponseBodyEmitter streamEvents() {
        ResponseBodyEmitter emitter = new ResponseBodyEmitter();

        executor.execute(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    String message = "Event " + i + " at " + System.currentTimeMillis();
                    emitter.send(message + "\n\n", MediaType.TEXT_PLAIN);
                    Thread.sleep(1000); // 模拟延迟
                }
                emitter.complete();
            } catch (Exception e) {
                emitter.completeWithError(e);
            }
        });

        return emitter;
    }
}

注意事项

线程管理：由于 ResponseBodyEmitter 支持长时间运行的操作，因此通常需要一个独立的线程池来处理这些任务，以避免阻塞主请求处理线程。
超时设置：合理设置超时时间，确保在网络不稳定或客户端断开连接的情况下能够及时释放资源。
错误处理：在实际应用中，务必对可能发生的异常进行妥善处理，通过 completeWithError 方法告知客户端发生了错误。

通过灵活运用 ResponseBodyEmitter，你可以构建出高效且响应迅速的Web应用程序，满足现代Web开发中的多种需求。

在Spring Boot应用中，使用ResponseBodyEmitter可以实现服务器向客户端实时异步流式推送数据。这非常适合于需要持续更新的场景，比如股票价格、社交媒体更新或任何需要即时反馈的应用。

如何使用

引入依赖：确保你的Spring Boot项目中包含了Spring Web依赖。通常，在创建Spring Boot项目时会自动包含这个依赖。

控制器方法配置：
在你的控制器类中，你可以定义一个处理请求的方法，并返回ResponseBodyEmitter类型。下面是一个简单的例子：

@RestController
public class StreamController {

    @GetMapping("/events")
    public ResponseBodyEmitter handle() {
        ResponseBodyEmitter emitter = new ResponseBodyEmitter();

        // 模拟后台数据处理和发送
        Executors.newSingleThreadExecutor().submit(() -> {
            try {
                for (int i = 0; i < 5; i++) {
                    emitter.send("Event " + i + " at " + new java.util.Date(), MediaType.TEXT_PLAIN);
                    Thread.sleep(1000); // 每隔一秒发送一次
                }
                emitter.complete();
            } catch (IOException | InterruptedException e) {
                emitter.completeWithError(e);
            }
        });

        return emitter;
    }
}

前端接收：前端可以通过AJAX或者Fetch API来发起请求并处理服务器端推送的数据。以下是一个使用JavaScript Fetch API的例子：

fetch('/events')
    .then(response => {
        const reader = response.body.getReader();
        const decoder = new TextDecoder('utf-8');

        function read() {
            reader.read().then(({ done, value }) => {
                if (done) {
                    console.log('Stream complete');
                    return;
                }
                console.log(decoder.decode(value, {stream: true}));
                read();
            });
        }

        read();
    });