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

Spring 实现 3 种异步流式接口，干掉接口超时烦恼

news 2025/9/18 12:44:45

大家好，我是小富～

如何处理比较耗时的接口？

这题我熟，直接上异步接口，使用 Callable、WebAsyncTask 和 DeferredResult、CompletableFuture等均可实现。

但这些方法有局限性，处理结果仅返回单个值。在某些场景下，如果需要接口异步处理的同时，还持续不断地向客户端响应处理结果，这些方法就不够看了。

Spring 框架提供了多种工具支持异步流式接口，如 ResponseBodyEmitter、SseEmitter 和 StreamingResponseBody。这些工具的用法简单，接口中直接返回相应的对象或泛型响应实体 ResponseEntity<xxxx>，如此这些接口就是异步的，且执行耗时操作亦不会阻塞 Servlet 的请求线程，不影响系统的响应能力。

下面将逐一介绍每个工具的使用及其应用场景。

ResponseBodyEmitter

ResponseBodyEmitter适应适合于需要动态生成内容并逐步发送给客户端的场景，例如：文件上传进度、实时日志等，可以在任务执行过程中逐步向客户端发送更新。

举个例子，经常用GPT你会发现当你提问后，得到的答案并不是一次性响应呈现的，而是逐步动态显示。这样做的好处是，让你感觉它在认真思考，交互体验比直接返回完整答案更为生动和自然。

使用ResponseBodyEmitter来实现下这个效果，创建 ResponseBodyEmitter 发送器对象，模拟耗时操作逐步调用 send 方法发送消息。

注意：ResponseBodyEmitter 的超时时间，如果设置为 0 或 -1，则表示连接不会超时；如果不设置，到达默认的超时时间后连接会自动断开。其他两种工具也是同样的用法，后边不在赘述了

@GetMapping("/bodyEmitter")
public ResponseBodyEmitter handle() {// 创建一个ResponseBodyEmitter，-1代表不超时ResponseBodyEmitter emitter = new ResponseBodyEmitter(-1L);// 异步执行耗时操作CompletableFuture.runAsync(() -> {try {// 模拟耗时操作for (int i = 0; i < 10000; i++) {System.out.println("bodyEmitter " + i);// 发送数据emitter.send("bodyEmitter " + i + " @ " + new Date() + "\n");Thread.sleep(2000);}// 完成emitter.complete();} catch (Exception e) {// 发生异常时结束接口emitter.completeWithError(e);}});return emitter;
}

实现代码非常简单。通过模拟每2秒响应一次结果，请求接口时可以看到页面数据在动态生成。效果与 GPT 回答基本一致。

SseEmitter

SseEmitter 是 ResponseBodyEmitter 的一个子类，它同样能够实现动态内容生成，不过主要将它用在服务器向客户端推送实时数据，如实时消息推送、状态更新等场景。在我之前的一篇文章我有 7种实现web实时消息推送的方案中详细介绍了 Server-Sent Events (SSE) 技术，感兴趣的可以回顾下。

SSE在服务器和客户端之间打开一个单向通道，服务端响应的不再是一次性的数据包而是text/event-stream类型的数据流信息，在有数据变更时从服务器流式传输到客户端。

整体的实现思路有点类似于在线视频播放，视频流会连续不断的推送到浏览器，你也可以理解成，客户端在完成一次用时很长（网络不畅）的下载。

客户端JS实现，通过一次 HTTP 请求建立连接后，等待接收消息。此时，服务端为每个连接创建一个 SseEmitter 对象，通过这个通道向客户端发送消息。

<body>
<div id="content" style="text-align: center;"><h1>SSE 接收服务端事件消息数据</h1><div id="message">等待连接...</div>
</div>
<script>let source = null;let userId = 7777function setMessageInnerHTML(message) {const messageDiv = document.getElementById("message");const newParagraph = document.createElement("p");newParagraph.textContent = message;messageDiv.appendChild(newParagraph);}if (window.EventSource) {// 建立连接source = new EventSource('http://127.0.0.1:9033/subSseEmitter/'+userId);setMessageInnerHTML("连接用户=" + userId);/*** 连接一旦建立，就会触发open事件* 另一种写法：source.onopen = function (event) {}*/source.addEventListener('open', function (e) {setMessageInnerHTML("建立连接。。。");}, false);/*** 客户端收到服务器发来的数据* 另一种写法：source.onmessage = function (event) {}*/source.addEventListener('message', function (e) {setMessageInnerHTML(e.data);});} else {setMessageInnerHTML("你的浏览器不支持SSE");}
</script>
</body>

在服务端，我们将 SseEmitter 发送器对象进行持久化，以便在消息产生时直接取出对应的 SseEmitter 发送器，并调用 send 方法进行推送。

private static final Map<String, SseEmitter> EMITTER_MAP = new ConcurrentHashMap<>();@GetMapping("/subSseEmitter/{userId}")
public SseEmitter sseEmitter(@PathVariable String userId) {log.info("sseEmitter: {}", userId);SseEmitter emitterTmp = new SseEmitter(-1L);EMITTER_MAP.put(userId, emitterTmp);CompletableFuture.runAsync(() -> {try {SseEmitter.SseEventBuilder event = SseEmitter.event().data("sseEmitter" + userId + " @ " + LocalTime.now()).id(String.valueOf(userId)).name("sseEmitter");emitterTmp.send(event);} catch (Exception ex) {emitterTmp.completeWithError(ex);}});return emitterTmp;
}@GetMapping("/sendSseMsg/{userId}")
public void sseEmitter(@PathVariable String userId, String msg) throws IOException {SseEmitter sseEmitter = EMITTER_MAP.get(userId);if (sseEmitter == null) {return;}sseEmitter.send(msg);
}

接下来向 userId=7777 的用户发送消息，127.0.0.1:9033/sendSseMsg/7777?msg=欢迎关注–>程序员小富，该消息可以在页面上实时展示。

而且SSE有一点比较好，客户端与服务端一旦建立连接，即便服务端发生重启，也可以做到自动重连。

StreamingResponseBody

StreamingResponseBody 与其他响应处理方式略有不同，主要用于处理大数据量或持续数据流的传输，支持将数据直接写入OutputStream。

例如，当我们需要下载一个超大文件时，使用 StreamingResponseBody 可以避免将文件数据一次性加载到内存中，而是持续不断的把文件流发送给客户端，从而解决下载大文件时常见的内存溢出问题。

接口实现直接返回 StreamingResponseBody 对象，将数据写入输出流并刷新，调用一次flush就会向客户端写入一次数据。

@GetMapping("/streamingResponse")
public ResponseEntity<StreamingResponseBody> handleRbe() {StreamingResponseBody stream = out -> {String message = "streamingResponse";for (int i = 0; i < 1000; i++) {try {out.write(((message + i) + "\r\n").getBytes());out.write("\r\n".getBytes());//调用一次flush就会像前端写入一次数据out.flush();TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}}};return ResponseEntity.ok().contentType(MediaType.TEXT_HTML).body(stream);
}

demo这里输出的是简单的文本流，如果是下载文件那么转换成文件流效果是一样的。

总结

这篇介绍三种实现异步流式接口的工具，算是 Spring 知识点的扫盲。使用起来比较简单，没有什么难点，但它们在实际业务中的应用场景还是很多的，通过这些工具，可以有效提高系统的性能和响应能力。

文中 Demo Github 地址：https://github.com/chengxy-nds/Springboot-Notebook/tree/master/springboot101/%E9%80%9A%E7%94%A8%E5%8A%9F%E8%83%BD/springboot-streaming

Spring 实现 3 种异步流式接口，干掉接口超时烦恼

ResponseBodyEmitter

SseEmitter

StreamingResponseBody

总结

相关文章：

Spring 实现 3 种异步流式接口，干掉接口超时烦恼

字节 HLLM 论文阅读

Chromium html＜iframe＞对应c++接口定义

Vue详细入门（语法【三】）

快速构建SpringBoot项目

架构设计笔记-14-云原生架构设计理论与实践

leetcode hot100 之【LeetCode 206. 反转链表】 java实现

基于Spring Cloud的电商系统设计与实现——用户与商品模块的研究（上）

Spring Boot + Vue 前后端分离项目总结：解决 CORS 和 404 问题

JVM篇（学习预热 - JVM正式展开 - （实战课程学习总结））（持续更新迭代）

WebGL编程指南 - 入门续

EPS导出DWG存在地物缺失或者没有编码属性的情况

跨境业务收款难？Zoho Books来帮忙

深入解析 Harris 角点检测算法：从孔径问题到响应函数的完整推导

抖音视频制作怎么暂停画面,抖音视频怎么让它有暂停的效果

Android GPIO方式解码红外数据

基于SpringBoot+Vue的益农智慧服务平台【提供源码+答辩PPT+参考文档+项目部署】

基于springboot的在线考试与学习交流网页

JS异步编程进阶（二）：rxjs与Vue、React、Angular框架集成及跨框架状态管理实现原理

nginx web代理

51c自动驾驶~合集58

Qt/C++开发监控GB28181系统/取流协议/同时支持udp/tcp被动/tcp主动

树莓派超全系列教程文档--(62)使用rpicam-app通过网络流式传输视频

基于uniapp+WebSocket实现聊天对话、消息监听、消息推送、聊天室等功能，多端兼容

Python爬虫实战：研究feedparser库相关技术

多模态商品数据接口：融合图像、语音与文字的下一代商品详情体验

vue3 字体颜色设置的多种方式

【JavaSE】绘图与事件入门学习笔记

网站指纹识别

现有的 Redis 分布式锁库（如 Redisson）提供了哪些便利？