如何处理比较耗时的接口这题我熟直接上异步接口使用Callable、WebAsyncTask和DeferredResult、CompletableFuture等均可实现。但这些方法有局限性处理结果仅返回单个值。在某些场景下如果需要接口异步处理的同时还持续不断地向客户端响应处理结果这些方法就不够看了。Spring 框架提供了多种工具支持异步流式接口如ResponseBodyEmitter、SseEmitter和StreamingResponseBody。这些工具的用法简单接口中直接返回相应的对象或泛型响应实体ResponseEntityxxxx如此这些接口就是异步的且执行耗时操作亦不会阻塞Servlet的请求线程不影响系统的响应能力。下面将逐一介绍每个工具的使用及其应用场景。ResponseBodyEmitterResponseBodyEmitter适用于要动态生成内容并逐步发送给客户端的场景例如文件上传进度、实时日志等可以在任务执行过程中逐步向客户端发送更新。举个例子经常用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 回答基本一致。SseEmitterSseEmitter是ResponseBodyEmitter的一个子类它同样能够实现动态内容生成不过主要将它用在服务器向客户端推送实时数据如实时消息推送、状态更新等场景。在我之前的一篇文章 我有 7种 实现web实时消息推送的方案 中详细介绍了Server-Sent Events (SSE)技术感兴趣的可以回顾下。SSE在服务器和客户端之间打开一个单向通道服务端响应的不再是一次性的数据包而是text/event-stream类型的数据流信息在有数据变更时从服务器流式传输到客户端。整体的实现思路有点类似于在线视频播放视频流会连续不断的推送到浏览器你也可以理解成客户端在完成一次用时很长网络不畅的下载。客户端JS实现通过一次 HTTP 请求建立连接后等待接收消息。此时服务端为每个连接创建一个SseEmitter对象通过这个通道向客户端发送消息。body div idcontent styletext-align: center; h1SSE 接收服务端事件消息数据/h1 div idmessage等待连接.../div /div script let source null; let userId 7777 function 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 MapString, 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); }接下来向userId7777的用户发送消息127.0.0.1:9033/sendSseMsg/7777?msg欢迎关注--程序员小富该消息可以在页面上实时展示。而且SSE有一点比较好客户端与服务端一旦建立连接即便服务端发生重启也可以做到自动重连。StreamingResponseBodyStreamingResponseBody与其他响应处理方式略有不同主要用于处理大数据量或持续数据流的传输支持将数据直接写入OutputStream。例如当我们需要下载一个超大文件时使用 StreamingResponseBody 可以避免将文件数据一次性加载到内存中而是持续不断的把文件流发送给客户端从而解决下载大文件时常见的内存溢出问题。接口实现直接返回 StreamingResponseBody 对象将数据写入输出流并刷新调用一次flush就会向客户端写入一次数据。GetMapping(/streamingResponse) public ResponseEntityStreamingResponseBody 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 实现 3 种异步流式接口，干掉接口超时烦恼
如何处理比较耗时的接口这题我熟直接上异步接口使用Callable、WebAsyncTask和DeferredResult、CompletableFuture等均可实现。但这些方法有局限性处理结果仅返回单个值。在某些场景下如果需要接口异步处理的同时还持续不断地向客户端响应处理结果这些方法就不够看了。Spring 框架提供了多种工具支持异步流式接口如ResponseBodyEmitter、SseEmitter和StreamingResponseBody。这些工具的用法简单接口中直接返回相应的对象或泛型响应实体ResponseEntityxxxx如此这些接口就是异步的且执行耗时操作亦不会阻塞Servlet的请求线程不影响系统的响应能力。下面将逐一介绍每个工具的使用及其应用场景。ResponseBodyEmitterResponseBodyEmitter适用于要动态生成内容并逐步发送给客户端的场景例如文件上传进度、实时日志等可以在任务执行过程中逐步向客户端发送更新。举个例子经常用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 回答基本一致。SseEmitterSseEmitter是ResponseBodyEmitter的一个子类它同样能够实现动态内容生成不过主要将它用在服务器向客户端推送实时数据如实时消息推送、状态更新等场景。在我之前的一篇文章 我有 7种 实现web实时消息推送的方案 中详细介绍了Server-Sent Events (SSE)技术感兴趣的可以回顾下。SSE在服务器和客户端之间打开一个单向通道服务端响应的不再是一次性的数据包而是text/event-stream类型的数据流信息在有数据变更时从服务器流式传输到客户端。整体的实现思路有点类似于在线视频播放视频流会连续不断的推送到浏览器你也可以理解成客户端在完成一次用时很长网络不畅的下载。客户端JS实现通过一次 HTTP 请求建立连接后等待接收消息。此时服务端为每个连接创建一个SseEmitter对象通过这个通道向客户端发送消息。body div idcontent styletext-align: center; h1SSE 接收服务端事件消息数据/h1 div idmessage等待连接.../div /div script let source null; let userId 7777 function 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 MapString, 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); }接下来向userId7777的用户发送消息127.0.0.1:9033/sendSseMsg/7777?msg欢迎关注--程序员小富该消息可以在页面上实时展示。而且SSE有一点比较好客户端与服务端一旦建立连接即便服务端发生重启也可以做到自动重连。StreamingResponseBodyStreamingResponseBody与其他响应处理方式略有不同主要用于处理大数据量或持续数据流的传输支持将数据直接写入OutputStream。例如当我们需要下载一个超大文件时使用 StreamingResponseBody 可以避免将文件数据一次性加载到内存中而是持续不断的把文件流发送给客户端从而解决下载大文件时常见的内存溢出问题。接口实现直接返回 StreamingResponseBody 对象将数据写入输出流并刷新调用一次flush就会向客户端写入一次数据。GetMapping(/streamingResponse) public ResponseEntityStreamingResponseBody 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这里输出的是简单的文本流如果是下载文件那么转换成文件流效果是一样的。