rx.observable_在Spring MVC流中使用rx-java Observable

rx.observable

Spring MVC现在已经支持异步请求处理流程了一段时间，该支持内部利用了Tomcat / Jetty等容器的Servlet 3异步支持。

Spring Web Async支持

考虑一下需要花一点时间处理的服务呼叫，该服务呼叫具有延迟：

public CompletableFuture<Message> getAMessageFuture() {return CompletableFuture.supplyAsync(() -> {logger.info("Start: Executing slow task in Service 1");Util.delay(1000);logger.info("End: Executing slow task in Service 1");return new Message("data 1");}, futureExecutor);
}

如果要在用户请求流中调用此服务，则传统的阻塞控制器流将如下所示：

@RequestMapping("/getAMessageFutureBlocking")
public Message getAMessageFutureBlocking() throws Exception {return service1.getAMessageFuture().get();
}

更好的方法是使用Spring异步支持，以便在CompletableFuture可用时将结果返回给用户，这样就不会占用容器线程了：

@RequestMapping("/getAMessageFutureAsync")
public DeferredResult<Message> getAMessageFutureAsync() {DeferredResult<Message> deffered = new DeferredResult<>(90000);CompletableFuture<Message> f = this.service1.getAMessageFuture();f.whenComplete((res, ex) -> {if (ex != null) {deffered.setErrorResult(ex);} else {deffered.setResult(res);}});return deffered;
}

在异步流中使用Observable

现在到本文的主题，最近我一直在使用Rx-java的出色的Observable类型作为我的服务返回类型，并想确保Web层在处理从服务调用返回的Observable类型时也保持异步。

考虑现在修改的上述服务以返回一个Observable：

public Observable<Message> getAMessageObs() {return Observable.<Message>create(s -> {logger.info("Start: Executing slow task in Service 1");Util.delay(1000);s.onNext(new Message("data 1"));logger.info("End: Executing slow task in Service 1");s.onCompleted();}).subscribeOn(Schedulers.from(customObservableExecutor));
}

我可以通过在Web层进行阻塞调用来结束返回Observable的所有好处，其中一个朴素的调用将是以下内容：

@RequestMapping("/getAMessageObsBlocking")
public Message getAMessageObsBlocking() {return service1.getAMessageObs().toBlocking().first();
}

为了使该流在Web层中异步，以下是处理此调用的更好方法，基本上是将Observable转换为Spring的DeferredResult类型：

@RequestMapping("/getAMessageObsAsync")
public DeferredResult<Message> getAMessageAsync() {Observable<Message> o = this.service1.getAMessageObs();DeferredResult<Message> deffered = new DeferredResult<>(90000);o.subscribe(m -> deffered.setResult(m), e -> deffered.setErrorResult(e));return deffered;
}

这将确保处理用户流的线程将在服务调用完成后立即返回，并且一旦可观察到的开始发出值，就将对用户响应进行React性处理。

如果您有兴趣进一步探索，这里是一个带有工作示例的github存储库：https://github.com/bijukunjummen/spring-web-observable。

参考文献：

Spring关于Web层中异步流的参考指南：http://docs.spring.io/spring/docs/current/spring-framework-reference/html/mvc.html#mvc-ann-async

NoBlogDefFound博客上无与伦比的Tomasz Nurkiewicz提供的有关Spring DeferredResult的更多详细信息 -http://www.nurkiewicz.com/2013/03/deferredresult-asynchronous-processing.html