ResponseBodyEmitter与SseEmitter使用

背景

• 最近在接入阿里云百炼AI助手时，接触到ResponseBodyEmitter。
  
• 实现了流式返回，比较感兴趣，所以去了解了一下如何实现。

ResponseBodyEmitter

简介

• Spring 框架提供的通用流式传输接口，支持分块传输编码（Chunked Encoding），允许逐步向客户端发送数据块，异步推送数据，而非一次性响应。返回ResponseBodyEmitter灵活性强，也可以自己构造标准的SSE返回。

核心代码实现

后端代码

private final ExecutorService executorService = Executors.newFixedThreadPool(2000);private final List<String> replyData = Arrays.asList("我是", "您的AI助手", "有什么可以帮您", "我是", "您的AI助手", "有什么可以帮您");/*** 返回ResponseBodyEmitter灵活性强，也可以自己构造标准的SSE返回* @param response* @return*/@RequestMapping(value = "/responseBodyEmitter")@CrossOriginpublic ResponseBodyEmitter responseBodyEmitter(HttpServletResponse response) {ResponseBodyEmitter emitter = new ResponseBodyEmitter(180000L);executorService.execute(() -> {try {for (String value : replyData) {emitter.send(value.getBytes(java.nio.charset.StandardCharsets.UTF_8));Thread.sleep(1000);}emitter.complete();} catch (Exception e) {log.error("其他的请求聊天异常 {}", e);emitter.completeWithError(e);throw new RuntimeException(e);}});return emitter;}

前端代码

• 先来看百炼助手使用fetch
  

• 我是使用Nuxt 3 框架,也用fetch API调用

const fetchStream = async () => {const response = await fetch('http://127.0.0.1:8080/api/index/responseBodyEmitter'); // 替换为你的接口路径const reader = response.body.getReader();const decoder = new TextDecoder('utf-8');// 持续读取数据流while (true) {const { done, value } = await reader.read();if (done) {console.log('Stream completed');break;}const textChunk = decoder.decode(value, { stream: true });console.log('Received chunk:', textChunk);text.value += textChunk}
}onMounted(()=>{console.log("onMounted")// 返回ResponseBodyEmitter fetchStream()})

运行效果

SseEmitter

• 在了解ResponseBodyEmitter时又发现了SseEmitter。

简介

• ResponseBodyEmitter的子类，为​​Server-Sent Events（SSE）​​协议设计，基于text/event-stream格式实现服务器到客户端的单向推送。自带重连机制。

核心代码实现

后端代码

   private final List<String> replyData = Arrays.asList("我是", "您的AI助手", "有什么可以帮您", "我是", "您的AI助手", "有什么可以帮您");private final ExecutorService executorService = Executors.newFixedThreadPool(2000);@RequestMapping("/chat")@CrossOriginpublic SseEmitter chat(String query) {SseEmitter emitter = new SseEmitter(180000L);executorService.execute(() -> {try {for (int i = 0; i < replyData.size(); i++) {String value = replyData.get(i);emitter.send(value.getBytes(java.nio.charset.StandardCharsets.UTF_8));Thread.sleep(1000);}emitter.send(SseEmitter.event().name("end").data("[DONE]"));Thread.sleep(1000);emitter.complete();} catch (Exception e) {log.error("其他的请求聊天异常 {}", e);emitter.completeWithError(e);throw new RuntimeException(e);}});log.info("返回emitter");return emitter;}

前端代码

//接收后台消息
const  receiveMessage = () =>{let eventSource  = new EventSource('http://127.0.0.1:8080/api/index/chat');eventSource.onopen = (event) =>{console.log("onopen ",event); }//接收成功eventSource.onmessage = (event) => {console.log("onmessage ",event);text.value =  text.value + event.data;};eventSource.addEventListener('end', (event) => {console.log("服务器主动关闭连接");eventSource.close(); // 主动关闭连接});//接收失败eventSource.onerror = (error) => {console.error('SSE error:',eventSource.readyState, error);if (eventSource.readyState === EventSource.CLOSED) { console.log("正常关闭"); // 应在此过滤已关闭状态} else {console.error("真实错误:");}// 如果不close，会自动重连eventSource.close()};}onMounted(()=>{console.log("onMounted")// SseEmitter 的方式receiveMessage()})

运行效果

小程序

• 小程序也支持SSE，核心代码如下，我使用的是uni-app-x

小程序代码

let title = ref<string>("hello")onMounted(() => {console.log("onMounted")startStream()})const startStream = () => {const requestTask = wx.request({url: 'http://127.0.0.1:8080/api/index/chat', // 后端接口地址method: 'GET',// dataType: 'text',//responseType: 'stream', // 关键：设置响应类型为 streamenableChunked: true,success: (res) => {},fail: (err) => {console.error("请求失败:", err);}});const decoder = new TextDecoder('utf-8');requestTask.onChunkReceived(function (resp) {let data = decoder.decode(resp.data)console.log(data)title.value += data});}

运行效果

本文源码

• https://github.com/1030907690/emitter-test
• https://github.com/1030907690/emitter-test-mobile
• https://github.com/1030907690/response-body-emitter-web

总结

• ​​ResponseBodyEmitter​​适用于更灵活的流式传输​​场景，如大文件下载或兼容性要求高的实时日志。SseEmitter则是基于标准的SSE协议。

• ResponseBodyEmitter、SseEmitter与 DeferredResult、Callable同为异步处理，不同的是 DeferredResult、Callable只能发送一次数据。ResponseBodyEmitter、SseEmitter可以多次调用send发送多次。

• 想要了解DeferredResult和Callable，可以参考本人拙作 Spring MVC(Boot) Servlet 3.0异步处理，DeferredResult和Callable、Spring MVC(Boot) Servlet 3.0异步处理，DeferredResult和Callable（续篇）