rxjava 循環(huán)發(fā)送事件

Spring Framework 4.2 GA即將發(fā)布，讓我們看一下它提供的一些新功能。 引起我注意的一個事件是一個簡單的新類SseEmitter ，它是對Spring MVC控制器中容易使用的發(fā)送事件的抽象。 SSE是一項技術(shù)，可讓您在一個HTTP連接內(nèi)沿一個方向?qū)?shù)據(jù)從服務(wù)器流式傳輸?shù)綖g覽器。 聽起來像是websocket可以做什么的子集。 但是，由于它是一個簡單得多的協(xié)議，因此可以在不需要全雙工的情況下使用，例如實時推動股價變化或顯示長時間運行的進程。 這將是我們的例子。

假設(shè)我們有一個具有以下API的虛擬硬幣礦工：

public interface CoinMiner {BigDecimal mine() {//...} }

每次調(diào)用mine()我們都必須等待幾秒鐘，才能獲得大約1個硬幣的回報（平均）。 如果要挖掘多個硬幣，則必須多次調(diào)用此方法：

@RestController public class MiningController {//...@RequestMapping("/mine/{count}")void mine(@PathVariable int count) {IntStream.range(0, count).forEach(x -> coinMiner.mine());}}

這項工作，我們可以請求/mine/10和mine()方法將執(zhí)行10次。 到目前為止，一切都很好。 但是挖掘是一項占用大量CPU的任務(wù)，將計算分散到多個內(nèi)核將是有益的。 此外，即使使用并行化，我們的API端點也相當(dāng)慢，我們必須耐心等待直到所有工作完成而沒有任何進度通知。 讓我們首先修復(fù)并行性–但是，由于并行流無法控制底層線程池，因此我們來使用顯式的ExecutorService ：

@Component class CoinMiner {CompletableFuture<BigDecimal> mineAsync(ExecutorService executorService) {return CompletableFuture.supplyAsync(this::mine, executorService);}//...}

客戶端代碼必須顯式提供ExecutorService （只是設(shè)計選擇）：

@RequestMapping("/mine/{count}") void mine(@PathVariable int count) {final List<CompletableFuture<BigDecimal>> futures = IntStream.range(0, count).mapToObj(x -> coinMiner.mineAsync(executorService)).collect(toList());futures.forEach(CompletableFuture::join); }

首先多次調(diào)用mineAsync ，然后（作為第二階段）等待所有期貨完成并join ，這mineAsync重要。 很容易寫：

IntStream.range(0, count).mapToObj(x -> coinMiner.mineAsync(executorService)).forEach(CompletableFuture::join);

但是，由于Java 8中流的惰性，該任務(wù)將按順序執(zhí)行！ 如果您還不習(xí)慣流的懶惰，請始終從下至上閱讀它們：我們要求join一些將來的內(nèi)容，以便流上升并只調(diào)用一次mineAsync() （惰性！），并將其傳遞給join() 。 當(dāng)join()完成時，它再次上升并要求另一個Future 。 通過使用collect()我們強制所有mineAsync()執(zhí)行，開始所有異步計算。 稍后，我們等待每一個。

介紹

現(xiàn)在該變得更具React性了（我說過了）。 控制器可以返回SseEmitter的實例。 從處理程序方法return后，容器線程將被釋放并可以處理更多即將到來的請求。 但是連接沒有關(guān)閉，客戶端一直在等待！ 我們應(yīng)該做的是保留對SseEmitter實例的引用，并在以后從另一個線程調(diào)用其send()和complete方法。 例如，我們可以啟動一個長時間運行的進程，并保持send()從任意線程進行進度。 完成該過程后，我們complete() SseEmitter ，最后關(guān)閉HTTP連接（至少從邏輯SseEmitter ，請記住Keep-alive ）。 在下面的示例中，我們有一堆CompletableFuture ，當(dāng)每個CompletableFuture完成時，我們只需將1發(fā)送給客戶端（ notifyProgress() ）。 當(dāng)所有期貨都完成后，我們完成流（ thenRun(sseEmitter::complete) ），關(guān)閉連接：

@RequestMapping("/mine/{count}") SseEmitter mine(@PathVariable int count) {final SseEmitter sseEmitter = new SseEmitter();final List<CompletableFuture<BigDecimal>> futures = mineAsync(count);futures.forEach(future ->future.thenRun(() -> notifyProgress(sseEmitter)));final CompletableFuture[] futuresArr = futures.toArray(new CompletableFuture[futures.size()]);CompletableFuture.allOf(futuresArr).thenRun(sseEmitter::complete);return sseEmitter; }private void notifyProgress(SseEmitter sseEmitter) {try {sseEmitter.send(1);} catch (IOException e) {throw new RuntimeException(e);} }private List<CompletableFuture<BigDecimal>> mineAsync(@PathVariable int count) {return IntStream.range(0, count).mapToObj(x -> coinMiner.mineAsync(executorService)).collect(toList()); }

調(diào)用此方法將產(chǎn)生以下響應(yīng)（注意Content-Type ）：

< HTTP/1.1 200 OK < Content-Type: text/event-stream;charset=UTF-8 < Transfer-Encoding: chunked < data:1data:1data:1data:1* Connection #0 to host localhost left intact

稍后我們將學(xué)習(xí)如何在客戶端解釋這種響應(yīng)。 現(xiàn)在暫時讓我們整理一下設(shè)計。

與引進RxJava

上面的代碼有效，但是看起來很混亂。 我們實際上有一系列事件，每個事件都代表計算的進度。 計算最終完成，因此流也應(yīng)發(fā)出信號結(jié)束。 聽起來就像是Observable ！ 我們從重構(gòu)CoinMiner開始，以返回Observable<BigDecimal ：

Observable<BigDecimal> mineMany(int count, ExecutorService executorService) {final ReplaySubject<BigDecimal> subject = ReplaySubject.create();final List<CompletableFuture<BigDecimal>> futures = IntStream.range(0, count).mapToObj(x -> mineAsync(executorService)).collect(toList());futures.forEach(future ->future.thenRun(() -> subject.onNext(BigDecimal.ONE)));final CompletableFuture[] futuresArr = futures.toArray(new CompletableFuture[futures.size()]);CompletableFuture.allOf(futuresArr).thenRun(subject::onCompleted);return subject; }

每當(dāng)mineMany()返回的事件出現(xiàn)在Observable ，我們就mineMany()那么多硬幣。 當(dāng)所有期貨都完成后，我們也完成了交易。 在實現(xiàn)方面，這看起來還沒有改善，但是從控制器的角度來看，它有多干凈：

@RequestMapping("/mine/{count}") SseEmitter mine(@PathVariable int count) {final SseEmitter sseEmitter = new SseEmitter();coinMiner.mineMany(count, executorService).subscribe(value -> notifyProgress(sseEmitter),sseEmitter::completeWithError,sseEmitter::complete);return sseEmitter; }

調(diào)用coinMiner.mineMany()我們只需訂閱事件。 事實證明Observable和SseEmitter方法匹配1：1。 這里發(fā)生的事情是不言自明的：啟動異步計算，每當(dāng)后臺計算發(fā)出任何進度信號時，將其轉(zhuǎn)發(fā)給客戶端。 好的，讓我們回到實現(xiàn)上。 由于我們將CompletableFuture和Observable混合使用，因此看起來很混亂。 我已經(jīng)描述了如何僅使用一個元素將CompletableFuture轉(zhuǎn)換為Observable 。 這是一個概述，包括rx.Single從RxJava 1.0.13開始發(fā)現(xiàn)的rx.Single抽象（此處未使用）：

public class Futures {public static <T> Observable<T> toObservable(CompletableFuture<T> future) {return Observable.create(subscriber ->future.whenComplete((result, error) -> {if (error != null) {subscriber.onError(error);} else {subscriber.onNext(result);subscriber.onCompleted();}}));}public static <T> Single<T> toSingle(CompletableFuture<T> future) {return Single.create(subscriber ->future.whenComplete((result, error) -> {if (error != null) {subscriber.onError(error);} else {subscriber.onSuccess(result);}}));}}

將這些實用程序運算符放在某個地方，我們可以改善實現(xiàn)并避免混合使用兩個API：

Observable<BigDecimal> mineMany(int count, ExecutorService executorService) {final List<Observable<BigDecimal>> observables = IntStream.range(0, count).mapToObj(x -> mineAsync(executorService)).collect(toList());return Observable.merge(observables); }Observable<BigDecimal> mineAsync(ExecutorService executorService) {final CompletableFuture<BigDecimal> future = CompletableFuture.supplyAsync(this::mine, executorService);return Futures.toObservable(future); }

RxJava有一個內(nèi)置的運算符，用于將多個Observable合并為一個，我們的每個基礎(chǔ)Observable發(fā)出一個事件。

深入研究RxJava運算符

讓我們使用RxJava的功能來稍微改善流式傳輸。

scan（）

當(dāng)前，每次我們開采一枚硬幣時，我們都會send(1)客戶端send(1)事件。 這意味著每個客戶都必須跟蹤其已經(jīng)收到的硬幣數(shù)量，以便計算總的計算數(shù)量。 如果服務(wù)器總是發(fā)送總金額而不是增量，那就太好了。 但是，我們不想更改實現(xiàn)。 事實證明，使用Observable.scan()運算符非常簡單：

@RequestMapping("/mine/{count}") SseEmitter mine(@PathVariable int count) {final SseEmitter sseEmitter = new SseEmitter();coinMiner.mineMany(count, executorService).scan(BigDecimal::add).subscribe(value -> notifyProgress(sseEmitter, value),sseEmitter::completeWithError,sseEmitter::complete);return sseEmitter; }private void notifyProgress(SseEmitter sseEmitter, BigDecimal value) {try {sseEmitter.send(value);} catch (IOException e) {e.printStackTrace();} }

scan()運算符接收上一個事件和當(dāng)前事件，并將它們組合在一起。 通過應(yīng)用BigDecimal::add我們只需將所有數(shù)字相加即可。 例如1、1 + 1，（1 + 1）+ 1，依此類推。 scan()類似于flatMap() ，但保留中間值。

用sample()采樣

可能是因為我們的后端服務(wù)產(chǎn)生了太多的進度更新，我們無法使用。 我們不想給客戶端增加不相關(guān)的更新并飽和帶寬。 每秒最多發(fā)送兩次更新聽起來很合理。 幸運的是，RxJava也有一個內(nèi)置的運算符：

Observable<BigDecimal> obs = coinMiner.mineMany(count, executorService); obs.scan(BigDecimal::add).sample(500, TimeUnit.MILLISECONDS).subscribe(//...);

sample()將定期查看底層流，并僅發(fā)出最新的項，并丟棄中間項。 幸運的是，我們使用scan()即時聚合了項目，因此我們不會丟失任何更新。

window() –恒定的發(fā)射間隔

不過有一個陷阱。 如果在選定的500毫秒內(nèi)沒有新內(nèi)容出現(xiàn)， sample()將不會兩次發(fā)出相同的項目。 很好，但是請記住我們正在通過TCP / IP連接推送這些更新。 最好定期將更新發(fā)送給客戶端，即使在此期間什么也沒發(fā)生–只是為了保持連接的正常運行，就像ping 。 可能有多種方法可以實現(xiàn)此要求，例如，涉及timeout()運算符。 我選擇使用window()運算符每500毫秒對所有事件進行分組：

Observable<BigDecimal> obs = coinMiner.mineMany(count, executorService); obs.window(500, TimeUnit.MILLISECONDS).flatMap(window -> window.reduce(BigDecimal.ZERO, BigDecimal::add)).scan(BigDecimal::add).subscribe(//...);

這是一個棘手的問題。 首先，我們將所有進度更新分組在500毫秒的窗口中。 然后，我們使用reduce來計算在此時間段內(nèi)開采的硬幣的總數(shù)（類似于scan() ）。 如果在此期間未開采任何硬幣，我們只需返回ZERO 。 最后，我們使用scan()匯總每個窗口的小計。 我們不再需要sample()因為window()確保每500毫秒發(fā)出一個事件。

客戶端

JavaScript中有很多SSE用法示例，因此為您提供一種調(diào)用我們的控制器的快速解決方案：

var source = new EventSource("/mine/10"); source.onmessage = function (event) {console.info(event); };

我相信SseEmitter是Spring MVC的一項重大改進，它將使我們能夠編寫更健壯和更快的Web應(yīng)用程序，需要即時的單向更新。

翻譯自: https://www.javacodegeeks.com/2015/08/server-sent-events-with-rxjava-and-sseemitter.html

rxjava 循環(huán)發(fā)送事件

總結(jié)

以上是生活随笔為你收集整理的rxjava 循环发送事件_使用RxJava和SseEmitter进行服务器发送的事件的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

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