rx.observable

Spring MVC現(xiàn)在已經(jīng)支持異步請(qǐng)求處理流程了一段時(shí)間，該支持內(nèi)部利用了Tomcat / Jetty等容器的Servlet 3異步支持。

Spring Web Async支持

考慮一下需要花一點(diǎn)時(shí)間處理的服務(wù)呼叫，該服務(wù)呼叫具有延遲：

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); }

如果要在用戶(hù)請(qǐng)求流中調(diào)用此服務(wù)，則傳統(tǒng)的阻塞控制器流將如下所示：

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

更好的方法是使用Spring異步支持，以便在CompletableFuture可用時(shí)將結(jié)果返回給用戶(hù)，這樣就不會(huì)占用容器線(xiàn)程了：

@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

現(xiàn)在到本文的主題，最近我一直在使用Rx-java的出色的Observable類(lèi)型作為我的服務(wù)返回類(lèi)型，并想確保Web層在處理從服務(wù)調(diào)用返回的Observable類(lèi)型時(shí)也保持異步。

考慮現(xiàn)在修改的上述服務(wù)以返回一個(gè)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)); }

我可以通過(guò)在Web層進(jìn)行阻塞調(diào)用來(lái)結(jié)束返回Observable的所有好處，其中一個(gè)樸素的調(diào)用將是以下內(nèi)容：

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

為了使該流在Web層中異步，以下是處理此調(diào)用的更好方法，基本上是將Observable轉(zhuǎn)換為Spring的DeferredResult類(lèi)型：

@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; }

這將確保處理用戶(hù)流的線(xiàn)程將在服務(wù)調(diào)用完成后立即返回，并且一旦可觀(guān)察到的開(kāi)始發(fā)出值，就將對(duì)用戶(hù)響應(yīng)進(jìn)行React性處理。

如果您有興趣進(jìn)一步探索， 這里是一個(gè)帶有工作示例的github存儲(chǔ)庫(kù)：https://github.com/bijukunjummen/spring-web-observable。

參考文獻(xiàn)：

Spring關(guān)于Web層中異步流的參考指南：http://docs.spring.io/spring/docs/current/spring-framework-reference/html/mvc.html#mvc-ann-async

NoBlogDefFound博客上無(wú)與倫比的Tomasz Nurkiewicz提供的有關(guān)Spring DeferredResult的更多詳細(xì)信息 -http://www.nurkiewicz.com/2013/03/deferredresult-asynchronous-processing.html

翻譯自: https://www.javacodegeeks.com/2015/03/using-rx-java-observable-in-a-spring-mvc-flow.html

rx.observable

總結(jié)

以上是生活随笔為你收集整理的rx.observable_在Spring MVC流中使用rx-java Observable的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。