Spring MVC的异步模式（ResponseBodyEmitter、SseEmitter、StreamingResponseBody） 高级使用篇

DeferredResult高级使用

上篇博文介绍的它的基本使用，那么本文主要结合一些特殊的使用场景，来介绍下它的高级使用，让能更深入的理解DeferredResult的强大之处。

它的优势也是很是明显的，可以实现两个彻底不相干的线程间的通讯。处理的时候请注意图中标记的线程安全问题~~~

 

 

实现长轮询服务端推送消息（long polling）

简单科普双向通讯的方式

在WebSocket协议以前(它是2011年发布的)，有三种实现双向通讯的方式：轮询（polling）、长轮询（long-polling）和iframe流（streaming）。

• 轮询（polling）：这个不解释了。优势是实现简单粗暴，后台处理简单。缺点也是大大的，耗流量、耗CPU。。。
• 长轮询（long-polling）：长轮询是对轮询的改进版。客户端发送HTTP给服务器以后，看有没有新消息，若是没有新消息，就一直等待（而不是一直去请求了）。当有新消息的时候，才会返回给客户端。 优势是对轮询作了优化，时效性也较好。缺点是：保持链接会消耗资源; 服务器没有返回有效数据，程序超时~~~
• iframe流（streaming）：是在页面中插入一个隐藏的iframe，利用其src属性在服务器和客户端之间建立一条长链接，服务器向iframe传输数据（一般是HTML，内有负责插入信息的javascript），来实时更新页面。(我的以为还不如长轮询呢。。。)
• WebSocket：WebSocket协议是基于TCP的一种新的网络协议。它实现了浏览器与服务器全双工(full-duplex)通讯——容许服务器主动发送信息给客户端。它将TCP的Socket（套接字）应用在了webpage上。 它的有点一大把：支持双向通讯，实时性更强；可发送二进制文件；很是节省流量。 但也是有缺点的：浏览器支持程度不一致，不支持断开重连 （实际上是最推荐的~~~）

---

以前看apollo配置中心的实现原理，apollo的发布配置推送变动消息就是用DeferredResult实现的。它的大概实现步骤以下：

1. apollo客户端会像服务端发送长轮询http请求，超时时间60秒
2. 当超时后返回客户端一个304 httpstatus,代表配置没有变动，客户端继续这个步骤重复发起请求
3. 当有发布配置的时候，服务端会调用DeferredResult.setResult返回200状态码。客户端收到响应结果后，会发起请求获取变动后的配置信息（注意这里是另一个请求哦~）。

为了演示，简单的按照此方式，写一个Demo：

@Configuration
@EnableWebMvc
public class AppConfig implements WebMvcConfigurer { @Override public void configureAsyncSupport(AsyncSupportConfigurer configurer) { // 超时时间设置为60s configurer.setDefaultTimeout(TimeUnit.SECONDS.toMillis(60)); } }

服务端简单代码模拟以下：

@Slf4j
@RestController
public class ApolloController { // 值为List，由于监视同一个名称空间的长轮询可能有N个（毕竟可能有多个客户端用同一份配置嘛） private Map<String, List<DeferredResult<String>>> watchRequests = new ConcurrentHashMap<>(); @GetMapping(value = "/all/watchrequests") public Object getWatchRequests() { return watchRequests; } // 模拟长轮询：apollo客户端来监听配置文件的变动~ 能够指定namespace 监视指定的NameSpace @GetMapping(value = "/watch/{namespace}") public DeferredResult<String> watch(@PathVariable("namespace") String namespace) { log.info("Request received,namespace is" + namespace + ",当前时间：" + System.currentTimeMillis()); DeferredResult<String> deferredResult = new DeferredResult<>(); //当deferredResult完成时（不管是超时仍是异常仍是正常完成），都应该移除watchRequests中相应的watch key deferredResult.onCompletion(() -> { log.info("onCompletion，移除对namespace：" + namespace + "的监视~"); List<DeferredResult<String>> list = watchRequests.get(namespace); list.remove(deferredResult); if (list.isEmpty()) { watchRequests.remove(namespace); } }); List<DeferredResult<String>> list = watchRequests.computeIfAbsent(namespace, (k) -> new ArrayList<>()); list.add(deferredResult); return deferredResult; } //模拟发布namespace配置：修改配置 @GetMapping(value = "/publish/{namespace}") public void publishConfig(@PathVariable("namespace") String namespace) { //do Something for update config if (watchRequests.containsKey(namespace)) { List<DeferredResult<String>> deferredResults = watchRequests.get(namespace); //通知全部watch这个namespace变动的长轮训配置变动结果 for (DeferredResult<String> deferredResult : deferredResults) { deferredResult.setResult(namespace + " changed，时间为" + System.currentTimeMillis()); } } } }

apollo处理超时时候会抛出一个异常AsyncRequestTimeoutException，所以咱们全局处理一下就成：

@Slf4j
@ControllerAdvice
class GlobalControllerExceptionHandler { @ResponseStatus(HttpStatus.NOT_MODIFIED)//返回304状态码 效果同HttpServletResponse#sendError(int) 但这样更优雅 @ResponseBody @ExceptionHandler(AsyncRequestTimeoutException.class) //捕获特定异常 public void handleAsyncRequestTimeoutException(AsyncRequestTimeoutException e) { System.out.println("handleAsyncRequestTimeoutException"); } }

用Ajax模拟Client端的伪代码以下：

 		//长轮询：一直去监听指定namespace的配置文件
        function watchConfig(){ $.ajax({ url:"http://localhost:8080/demo_war/watch/classroomconfig", method:"get", success:function(response,status){ if(status == 304){ watchConfig(); //超时，没有更改，那就继续去监听 }else if(status == 200){ getNewConfig(); //监听到更改后，立马去获取最新的配置文件内容回来作事 ... watchConfig(); // 昨晚过后又去监听着 } } }); } // 调用去监听获取配置文件的函数 watchConfig();

这样子咱们就基本模拟了一个长轮询的案例~

长轮询的应用场景也是不少的，好比咱们如今要实现这样一个功能：浏览器要实时展现服务端计算出来的数据。（这个用普通轮询就会有延迟且浪费资源，可是用这种相似长链接的方案就很合适）

ResponseBodyEmitter和SseEmitter

Callback和DeferredResult用于设置单个结果，若是有多个结果须要set返回给客户端时，可使用SseEmitter以及ResponseBodyEmitter，each object is written with a compatible HttpMessageConverter。返回值能够直接写他们自己，也能够放在ResponseEntity里面

它俩都是Spring4.2以后提供的类。由ResponseBodyEmitterReturnValueHandler负责处理。 这个和Spring5提供的webFlux技术已经很像了，后续讲到的时候还会提到他们~~~~ Emitter：发射器

它们的使用方式几乎同：DeferredResult，这里我只把官方的例子拿出来你就懂了

 

 

SseEmitter是ResponseBodyEmitter的子类,它提供Server-Sent Events（Sse）.服务器事件发送是”HTTP Streaming”的另外一个变种技术.只是从服务器发送的事件按照W3C Server-Sent Events规范来的（推荐使用） 它的使用方式上，彻底同上

Server-Sent Events这个规范可以来用于它们的预期使用目的：就是从server发送events到clients（服务器推）.在Spring MVC中能够很容易的实现.仅仅须要返回一个SseEmitter类型的值.

向这种场景在在线游戏、在线协做、金融领域等等都有很好的应用。固然，若是你对稳定性什么的要求都很是高，官方也推荐最好是使用WebSocket来实现~

ResponseBodyEmitter容许经过HttpMessageConverter把发送的events写到对象到response中.这多是最多见的状况。例如写JSON数据 但是有时候它被用来绕开message转换直接写入到response的OutputStream。例如文件下载.这样能够经过返回StreamingResponseBody类型的值作到.

StreamingResponseBody （很方便的文件下载）

它用于直接将结果写出到Response的OutputStream中； 如文件下载等

 

 

接口源码很是简单：

@FunctionalInterface
public interface StreamingResponseBody { void writeTo(OutputStream outputStream) throws IOException; }

异步优化

Spring内部默认不使用线程池处理的（经过源码分析后面咱们是能看到的）,为了提升处理的效率，咱们能够本身优化，建议本身在配置里注入一个线程池供给使用，参考以下：

	// 提供一个mvc里专用的线程池。。。  这是全局的方式~~~~
    @Bean
    public ThreadPoolTaskExecutor mvcTaskExecutor() { ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); executor.setCorePoolSize(10); executor.setQueueCapacity(100); executor.setMaxPoolSize(25); return executor; } // 最优解决方案不是像上面同样配置通用的，而是配置一个单独的专用的，以下~~~~ @Configuration @EnableWebMvc public class WebMvcConfig extends WebMvcConfigurerAdapter { // 配置异步支持~~~~ @Override public void configureAsyncSupport(AsyncSupportConfigurer configurer) { // 设置一个用于异步执行的执行器~~~AsyncTaskExecutor configurer.setTaskExecutor(mvcTaskExecutor()); configurer.setDefaultTimeout(60000L); } }

总结

总的来讲，Spring MVC提供的便捷的异步支持，可以大大的提升Tomcat容器等的性能。同时也给咱们的应用提供了更多的便利。这也为Spring5之后的Reactive编程模型提供了有利的支持和保障。 Spring是一个易学难精的技术，想要把各类技术融汇贯通，还有后续更扎实的深挖~