在第1部分：線程池中，我們?cè)O(shè)計(jì)并實(shí)現(xiàn)了相對(duì)簡(jiǎn)單的系統(tǒng)，用于實(shí)時(shí)處理事件。 確保您已閱讀上一部分，因?yàn)樗恍┪覀儗⒅赜玫念?lèi)。 以防萬(wàn)一這是要求：

一個(gè)系統(tǒng)每秒傳送約一千個(gè)事件。 每個(gè)Event至少具有兩個(gè)屬性：

• clientId –我們期望一個(gè)客戶(hù)端每秒最多可以處理幾個(gè)事件
• UUID –全球唯一

消耗一個(gè)事件大約需要10毫秒。 設(shè)計(jì)此類(lèi)流的使用者：

允許實(shí)時(shí)處理事件

與一個(gè)客戶(hù)端有關(guān)的事件應(yīng)按順序進(jìn)行處理，即，您不能并行處理同一clientId事件

如果10秒鐘內(nèi)出現(xiàn)重復(fù)的UUID ，請(qǐng)將其刪除。 假設(shè)10秒鐘后不會(huì)出現(xiàn)重復(fù)

到目前為止，我們提出的是線程池和共享緩存的組合。 這次我們將使用RxJava實(shí)現(xiàn)解決方案。 首先，我沒(méi)有透露EventStream的實(shí)現(xiàn)方式，僅提供了API：

interface EventStream {void consume(EventConsumer consumer);}

實(shí)際上，對(duì)于手動(dòng)測(cè)試，我構(gòu)建了一個(gè)簡(jiǎn)單的RxJava流，其行為與需求類(lèi)似，類(lèi)似于系統(tǒng)：

@Slf4j class EventStream {void consume(EventConsumer consumer) {observe().subscribe(consumer::consume,e -> log.error("Error emitting event", e));}Observable<Event> observe() {return Observable.interval(1, TimeUnit.MILLISECONDS).delay(x -> Observable.timer(RandomUtils.nextInt(0, 1_000), TimeUnit.MICROSECONDS)).map(x -> new Event(RandomUtils.nextInt(1_000, 1_100), UUID.randomUUID())).flatMap(this::occasionallyDuplicate, 100).observeOn(Schedulers.io());}private Observable<Event> occasionallyDuplicate(Event x) {final Observable<Event> event = Observable.just(x);if (Math.random() >= 0.01) {return event;}final Observable<Event> duplicated =event.delay(RandomUtils.nextInt(10, 5_000), TimeUnit.MILLISECONDS);return event.concatWith(duplicated);}}

了解此模擬器的工作原理不是必不可少的，但很有趣。 首先，我們產(chǎn)生的源源不斷的Long值（ 0 ， 1 ， 2 ...）每毫秒使用（每秒千個(gè)事件） interval()操作。 然后，我們使用delay()運(yùn)算符將每個(gè)事件延遲0到1_000微秒之間的隨機(jī)時(shí)間。 這樣，事件將在難以預(yù)測(cè)的時(shí)刻出現(xiàn)，而情況會(huì)更加現(xiàn)實(shí)。 最后，我們將每個(gè)Long值映射（使用ekhem， map()運(yùn)算符） map()到一個(gè)隨機(jī)Event ，該Event的clientId在1_000和1_100之間（包括端1_100在內(nèi)）。

最后一點(diǎn)很有趣。 我們想模擬偶爾的重復(fù)。 為此，我們將每個(gè)事件（使用flatMap() ）映射到自身（在99％的情況下）。 但是，在1％的情況下，我們兩次返回此事件，第二次發(fā)生在10毫秒至5秒后。 在實(shí)踐中，該事件的重復(fù)實(shí)例將在其他數(shù)百個(gè)事件之后出現(xiàn)，這使流的行為逼真。

與EventStream進(jìn)行交互的方式有兩種：通過(guò)consume()回調(diào)和通過(guò)observe()流。 我們可以利用Observable<Event>來(lái)快速建立功能與第1部分非常相似但更簡(jiǎn)單的處理管道。

缺少背壓

利用RxJava的第一個(gè)幼稚方法很快就失敗了：

EventStream es = new EventStream(); EventConsumer clientProjection = new ClientProjection(new ProjectionMetrics(new MetricRegistry()));es.observe().subscribe(clientProjection::consume,e -> log.error("Fatal error", e));

（ ClientProjection ， ProjectionMetrics等人來(lái)自第1部分 ）。 我們幾乎立即獲得MissingBackpressureException ，這是預(yù)期的。 還記得我們的第一個(gè)解決方案是如何通過(guò)處理越來(lái)越多的延遲來(lái)滯后嗎？ RxJava嘗試避免這種情況，并避免隊(duì)列溢出。 由于使用者（ ClientProjection ）無(wú)法實(shí)時(shí)處理事件，因此拋出MissingBackpressureException 。 這是快速失敗的行為。 最快的解決方案是像以前一樣使用RxJava的功能將消耗轉(zhuǎn)移到一個(gè)單獨(dú)的線程池中：

EventStream es = new EventStream(); EventConsumer clientProjection = new FailOnConcurrentModification(new ClientProjection(new ProjectionMetrics(new MetricRegistry())));es.observe().flatMap(e -> clientProjection.consume(e, Schedulers.io())).window(1, TimeUnit.SECONDS).flatMap(Observable::count).subscribe(c -> log.info("Processed {} events/s", c),e -> log.error("Fatal error", e));

EventConsumer接口具有一個(gè)輔助方法，該方法可以在提供的Scheduler上異步使用事件：

@FunctionalInterface interface EventConsumer {Event consume(Event event);default Observable<Event> consume(Event event, Scheduler scheduler) {return Observable.fromCallable(() -> this.consume(event)).subscribeOn(scheduler);}}

通過(guò)在單獨(dú)的Scheduler.io()使用flatMap()使用事件，可以異步調(diào)用每個(gè)使用。 這次事件幾乎是實(shí)時(shí)處理的，但是存在更大的問(wèn)題。 由于某種原因，我用FailOnConcurrentModification裝飾了ClientProjection 。 事件彼此獨(dú)立使用，因此可能會(huì)同時(shí)處理同一clientId兩個(gè)事件。 不好。 幸運(yùn)的是，在RxJava中解決此問(wèn)題比使用普通線程要容易得多：

es.observe().groupBy(Event::getClientId).flatMap(byClient -> byClient.observeOn(Schedulers.io()).map(clientProjection::consume)).window(1, TimeUnit.SECONDS).flatMap(Observable::count).subscribe(c -> log.info("Processed {} events/s", c),e -> log.error("Fatal error", e));

有點(diǎn)改變了。 首先，我們將事件按clientId分組。 這將單個(gè)Observable流拆分為流 。 每個(gè)名為byClient子流代表與同一clientId相關(guān)的所有事件。 現(xiàn)在，如果我們映射到此子流，我們可以確保與同一個(gè)clientId相關(guān)的事件不會(huì)同時(shí)發(fā)生。 外部流很懶，因此我們必須訂閱它。 與其單獨(dú)訂閱每個(gè)事件，我們不每秒收集事件并進(jìn)行計(jì)數(shù)。 這樣，我們每秒就會(huì)收到一個(gè)Integer類(lèi)型的單個(gè)事件，該事件表示每秒消耗的事件數(shù)。

使用全局狀態(tài)的不純，非慣常，容易出錯(cuò)，不安全的重復(fù)數(shù)據(jù)刪除解決方案

現(xiàn)在我們必須刪除重復(fù)的UUID 。 丟棄重復(fù)項(xiàng)的最簡(jiǎn)單但非常愚蠢的方法是利用全局狀態(tài)。 我們可以通過(guò)在filter()運(yùn)算符之外可用的緩存中查找重復(fù)項(xiàng)來(lái)簡(jiǎn)單地過(guò)濾掉重復(fù)項(xiàng)：

final Cache<UUID, UUID> seenUuids = CacheBuilder.newBuilder().expireAfterWrite(10, TimeUnit.SECONDS).build();es.observe().filter(e -> seenUuids.getIfPresent(e.getUuid()) == null).doOnNext(e -> seenUuids.put(e.getUuid(), e.getUuid())).subscribe(clientProjection::consume,e -> log.error("Fatal error", e));

如果要監(jiān)視此機(jī)制的使用，只需添加指標(biāo)：

Meter duplicates = metricRegistry.meter("duplicates");es.observe().filter(e -> {if (seenUuids.getIfPresent(e.getUuid()) != null) {duplicates.mark();return false;} else {return true;}})

從操作員內(nèi)部訪問(wèn)全局狀態(tài)，尤其是可變狀態(tài)非常危險(xiǎn)，并且破壞了RxJava的唯一目的-簡(jiǎn)化并發(fā)。 顯然，我們使用了Guava的線程安全Cache ，但是在許多情況下，很容易錯(cuò)過(guò)從多個(gè)線程訪問(wèn)共享全局可變狀態(tài)的地方。 如果您發(fā)現(xiàn)自己在運(yùn)算符鏈之外修改了一些變量，請(qǐng)非常小心。

RxJava 1.x中的自定義

RxJava 1.x有一個(gè)distinct()運(yùn)算符，大概可以完成此工作：

es.observe().distinct(Event::getUuid).groupBy(Event::getClientId)

不幸的是， distinct()在內(nèi)部將所有密鑰（ UUID distinct()存儲(chǔ)在不斷增長(zhǎng)的HashSet 。 但是我們只關(guān)心最近10秒鐘內(nèi)的重復(fù)！ 通過(guò)復(fù)制粘貼DistinctOperator的實(shí)現(xiàn)，我創(chuàng)建了DistinctEvent運(yùn)算符，該運(yùn)算符利用Guava的緩存僅存儲(chǔ)了最后10秒鐘的UUID值。 我故意在此運(yùn)算符中對(duì)Event進(jìn)行了硬編碼，而不是使其變得更通用以使代碼更易于理解：

class DistinctEvent implements Observable.Operator<Event, Event> {private final Duration duration;DistinctEvent(Duration duration) {this.duration = duration;}@Overridepublic Subscriber<? super Event> call(Subscriber<? super Event> child) {return new Subscriber<Event>(child) {final Map<UUID, Boolean> keyMemory = CacheBuilder.newBuilder().expireAfterWrite(duration.toMillis(), TimeUnit.MILLISECONDS).<UUID, Boolean>build().asMap();@Overridepublic void onNext(Event event) {if (keyMemory.put(event.getUuid(), true) == null) {child.onNext(event);} else {request(1);}}@Overridepublic void onError(Throwable e) {child.onError(e);}@Overridepublic void onCompleted() {child.onCompleted();}};} }

用法非常簡(jiǎn)單，整個(gè)實(shí)現(xiàn)（加上自定義運(yùn)算符）如下：

es.observe().lift(new DistinctEvent(Duration.ofSeconds(10))).groupBy(Event::getClientId).flatMap(byClient -> byClient.observeOn(Schedulers.io()).map(clientProjection::consume)).window(1, TimeUnit.SECONDS).flatMap(Observable::count).subscribe(c -> log.info("Processed {} events/s", c),e -> log.error("Fatal error", e));

實(shí)際上，如果您跳過(guò)每秒的日志記錄，它甚至可以更短：

es.observe().lift(new DistinctEvent(Duration.ofSeconds(10))).groupBy(Event::getClientId).flatMap(byClient -> byClient.observeOn(Schedulers.io()).map(clientProjection::consume)).subscribe(e -> {},e -> log.error("Fatal error", e));

該解決方案比以前基于線程池和裝飾器的解決方案要短得多。 唯一尷尬的部分是自定義運(yùn)算符，該自定義運(yùn)算符可在存儲(chǔ)太多歷史UUID時(shí)避免內(nèi)存泄漏。 幸運(yùn)的是RxJava 2得以解救！

RxJava 2.x和更強(qiáng)大的內(nèi)置

實(shí)際上，我是從提交公關(guān)RxJava具有更強(qiáng)大的執(zhí)行這種緊密distinct()操作。 但是在我檢查2.x分支之前，它是： distinct()允許提供自定義Collection而不是硬編碼的HashSet 。 信不信由你，依賴(lài)倒置不僅涉及Spring框架或Java EE。 當(dāng)庫(kù)允許您提供其內(nèi)部數(shù)據(jù)結(jié)構(gòu)的自定義實(shí)現(xiàn)時(shí)，這也是DI。 首先，我創(chuàng)建一個(gè)輔助方法，該方法可以構(gòu)建由Map<UUID, Boolean>支持，由Cache<UUID, Boolean>支持的Set<UUID> Cache<UUID, Boolean> 。 我們一定喜歡代表團(tuán)！

private Set<UUID> recentUuids() {return Collections.newSetFromMap(CacheBuilder.newBuilder().expireAfterWrite(10, TimeUnit.SECONDS).<UUID, Boolean>build().asMap()); }

有了這種方法，我們可以使用以下表達(dá)式實(shí)現(xiàn)整個(gè)任務(wù)：

es.observe().distinct(Event::getUuid, this::recentUuids).groupBy(Event::getClientId).flatMap(byClient -> byClient.observeOn(Schedulers.io()).map(clientProjection::consume)).subscribe(e -> {},e -> log.error("Fatal error", e));

優(yōu)雅，簡(jiǎn)潔，清晰！ 它讀起來(lái)幾乎像一個(gè)問(wèn)題：

• 觀察事件流
• 僅考慮不同的UUID
• 客戶(hù)分組活動(dòng)
• 為每個(gè)客戶(hù)消耗（依次）

希望您喜歡所有這些解決方案，并發(fā)現(xiàn)它們對(duì)您的日常工作很有用。

也可以看看：

• 小規(guī)模流處理kata。 第1部分：線程池
• 小規(guī)模流處理kata。 第2部分：RxJava 1.x / 2.x

翻譯自: https://www.javacodegeeks.com/2016/10/small-scale-stream-processing-kata-part-2-rxjava-1-x2-x.html

總結(jié)

以上是生活随笔為你收集整理的小规模流处理kata。 第2部分：RxJava 1.x / 2.x的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

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