Spring 实现 3 种异步接口(苏三说技术2024年10月18日文章)

分享到：

作者：苏三说技术 | 更新时间：2024-10-20 02:55:01

大家好，我是苏三～如何处理比较耗时的接口？这题我熟，直接上异步接口，使用 Callable、WebAsyncTask 和 DeferredResult、CompletableFuture等均可实现。但这些方法有局限性，处理结果仅返回单个值。...

A股板块轮动加剧，跨年大妖来袭，这几只票主力已明显介入！微信搜索关注【研讯小组】公众号（可长按复制），回复666，领取代码！

大家好，我是苏三～

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

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

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

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

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

ResponseBodyEmitter

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

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

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

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

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

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

SseEmitter

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

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

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

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

<body> <divid="content"style="text-align:center;"> <h1>SSE接收服务端事件消息数据h1> <divid="message">等待连接...div> div> <script> letsource=null; letuserId=7777 functionsetMessageInnerHTML(message){ constmessageDiv=document.getElementById("message"); constnewParagraph=document.createElement("p"); newParagraph.textContent=message; messageDiv.appendChild(newParagraph); } if(window.EventSource){ //建立连接 source=newEventSource('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 方法进行推送。

privatestaticfinalMapEMITTER_MAP=newConcurrentHashMap<>(); @GetMapping("/subSseEmitter/{userId}") publicSseEmittersseEmitter(@PathVariableStringuserId){ log.info("sseEmitter:{}",userId); SseEmitteremitterTmp=newSseEmitter(-1L); EMITTER_MAP.put(userId,emitterTmp); CompletableFuture.runAsync(()->{ try{ SseEmitter.SseEventBuilderevent=SseEmitter.event() .data("sseEmitter"+userId+"@"+LocalTime.now()) .id(String.valueOf(userId)) .name("sseEmitter"); emitterTmp.send(event); }catch(Exceptionex){ emitterTmp.completeWithError(ex); } }); returnemitterTmp; } @GetMapping("/sendSseMsg/{userId}") publicvoidsseEmitter(@PathVariableStringuserId,Stringmsg)throwsIOException{ SseEmittersseEmitter=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") publicResponseEntityhandleRbe(){ StreamingResponseBodystream=out->{ Stringmessage="streamingResponse"; for(inti=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(InterruptedExceptione){ e.printStackTrace(); } } }; returnResponseEntity.ok().contentType(MediaType.TEXT_HTML).body(stream); }