RxJava 數(shù)學(xué)與聚合算子詳解count、reduce、collect、toList、toMap 全解析【免費(fèi)下載鏈接】RxJavaRxJava – Reactive Extensions for the JVM – a library for composing asynchronous and event-based programs using observable sequences for the Java VM.項(xiàng)目地址: https://gitcode.com/gh_mirrors/rx/RxJava本文以 docs/Mathematical-and-Aggregate-Operators.md 為骨架系統(tǒng)講解 RxJava 中作用于“整條序列”的數(shù)學(xué)與聚合算子包括來自rxjava2-extensions擴(kuò)展模塊的 8 個(gè)數(shù)學(xué)算子averageDouble、averageFloat、max、min、sumDouble、sumFloat、sumInt、sumLong以及 10 個(gè)標(biāo)準(zhǔn)聚合算子count、reduce、reduceWith、collect、collectInto、toList、toSortedList、toMap、toMultimap。讀完后你將理解這些算子為何必須緩沖整個(gè)序列、返回Single/Maybe與返回Observable/Flowable的設(shè)計(jì)差異并能結(jié)合 Observable.java 的源碼與內(nèi)部實(shí)現(xiàn)類正確選型和使用這些算子。聚合算子的共同前提必須等上游完整結(jié)束這類算子對(duì)Observable或Flowable發(fā)出的整個(gè)物品序列執(zhí)行數(shù)學(xué)或其他操作。由于它們必須等待源序列發(fā)出所有物品甚至完成信號(hào)之后才能構(gòu)造自己的發(fā)射值而且通常必須緩沖這些物品因此在可能擁有非常長(zhǎng)甚至是無限序列的流上使用這類算子是有危險(xiǎn)的。這一限制在所有聚合算子上都成立使用時(shí)請(qǐng)注意對(duì)無限流如Observable.never()、interval、持續(xù)推送的 WebSocket 數(shù)據(jù)調(diào)用toList()或reduce()緩沖區(qū)會(huì)持續(xù)增長(zhǎng)直至內(nèi)存耗盡對(duì)超長(zhǎng)流應(yīng)優(yōu)先考慮分段聚合配合buffer、window先切塊再聚合或流式近似方案。另一個(gè)貫穿全文的關(guān)鍵設(shè)計(jì)差異標(biāo)準(zhǔn)聚合算子count、reduce等返回Single或Maybe因?yàn)檩敵鑫锲返臄?shù)量恒定為至多一個(gè)數(shù)學(xué)算子MathObservable/MathFlowable提供的一組則返回Observable和Flowable這是兩者簽名上最根本的不同。數(shù)學(xué)算子來自 rxjava2-extensions 模塊本節(jié)的算子屬于RxJava2Extensions項(xiàng)目并非 RxJava 核心庫自帶。需要將rxjava2-extensions模塊作為依賴加入項(xiàng)目groupId 為com.github.akarnokd。以下示例假設(shè)已引入MathObservable和MathFlowable類import hu.akarnokd.rxjava2.math.MathObservable; import hu.akarnokd.rxjava2.math.MathFlowable;這組算子均可用于Flowable與Observable但不適用于Maybe、Single、Completable。averageDouble計(jì)算Observable發(fā)出的Number的平均值并以Double形式發(fā)出該平均值。ObservableInteger numbers Observable.just(1, 2, 3); MathObservable.averageDouble(numbers).subscribe((Double avg) - System.out.println(avg)); // prints 2.0averageFloat計(jì)算Observable發(fā)出的Number的平均值并以Float形式發(fā)出該平均值。ObservableInteger numbers Observable.just(1, 2, 3); MathObservable.averageFloat(numbers).subscribe((Float avg) - System.out.println(avg)); // prints 2.0max發(fā)出源Observable發(fā)出的最大值。可以指定一個(gè)Comparator用于比較源發(fā)出的元素。ObservableInteger numbers Observable.just(4, 9, 5); MathObservable.max(numbers).subscribe(System.out::println); // prints 9指定Comparator可以在源中查找最長(zhǎng)的Stringfinal ObservableString names Observable.just(Kirk, Spock, Chekov, Sulu); MathObservable.max(names, Comparator.comparingInt(String::length)) .subscribe(System.out::println); // prints Chekovmin發(fā)出源Observable發(fā)出的最小值。同樣可以指定Comparator自定義比較邏輯。ObservableInteger numbers Observable.just(4, 9, 5); MathObservable.min(numbers).subscribe(System.out::println); // prints 4sumDouble將Observable發(fā)出的Double相加并發(fā)出這個(gè)和。ObservableDouble numbers Observable.just(1.0, 2.0, 3.0); MathObservable.sumDouble(numbers).subscribe((Double sum) - System.out.println(sum)); // prints 6.0sumFloat將Observable發(fā)出的Float相加并發(fā)出這個(gè)和。ObservableFloat numbers Observable.just(1.0F, 2.0F, 3.0F); MathObservable.sumFloat(numbers).subscribe((Float sum) - System.out.println(sum)); // prints 6.0sumInt將Observable發(fā)出的Integer相加并發(fā)出這個(gè)和。ObservableInteger numbers Observable.range(1, 100); MathObservable.sumInt(numbers).subscribe((Integer sum) - System.out.println(sum)); // prints 5050sumLong將Observable發(fā)出的Long相加并發(fā)出這個(gè)和。ObservableLong numbers Observable.rangeLong(1L, 100L); MathObservable.sumLong(numbers).subscribe((Long sum) - System.out.println(sum)); // prints 5050注意從源碼結(jié)構(gòu)看這些sum*算子在MathObservable/MathFlowable中是按數(shù)值類型分別提供的本倉庫中搜索不到MathObservable、averageDouble等相關(guān)定義證實(shí)它們來自外部的rxjava2-extensions模塊選擇時(shí)應(yīng)與源元素類型匹配以避免不必要的裝箱/拆箱與類型轉(zhuǎn)換。標(biāo)準(zhǔn)聚合算子注意這些標(biāo)準(zhǔn)聚合算子返回Single或Maybe因?yàn)檩敵鑫锲返臄?shù)量恒定為至多一個(gè)。count計(jì)算Observable發(fā)出的物品數(shù)量并以Long形式發(fā)出該計(jì)數(shù)。Maybe同樣提供該算子Single與Completable不提供。Observable.just(1, 2, 3).count().subscribe(System.out::println); // prints 3在 Observable.java 中count()是一個(gè)final方法直接構(gòu)造內(nèi)部實(shí)現(xiàn)類。從源碼結(jié)構(gòu)看核心庫為計(jì)數(shù)場(chǎng)景準(zhǔn)備了兩個(gè)實(shí)現(xiàn)ObservableCount.java返回Observable的形態(tài)與 ObservableCountSingle.java返回SingleLong并實(shí)現(xiàn)FuseToObservable支持融合優(yōu)化。計(jì)數(shù)本身無需緩沖任何物品只維護(hù)一個(gè)Long計(jì)數(shù)器因此它是聚合算子中內(nèi)存開銷最小的一類。reduce按順序?qū)⒑瘮?shù)應(yīng)用于每個(gè)發(fā)出的物品只發(fā)出最終累積值。reduce(seed, reducer)以一個(gè)初始種子值啟動(dòng)累積過程。Observable.range(1, 5) .reduce((product, x) - product * x) .subscribe(System.out::println); // prints 120在 Observable.java 中簽名形如public final NonNull R SingleR reduce(R seed, NonNull BiFunctionR, ? super T, R reducer)其內(nèi)部只需要在內(nèi)存中保存一個(gè)“當(dāng)前累積值”并不緩沖整個(gè)序列——這是reduce相對(duì)toList的重要優(yōu)勢(shì)對(duì)超長(zhǎng)序列做純累積運(yùn)算求和、求最大、拼接字符串等時(shí)reduce的內(nèi)存占用是 O(1) 的而先toList再處理則是 O(n) 的。reduceWith與reduce相同——按順序應(yīng)用函數(shù)、只發(fā)出最終累積值——區(qū)別在于初始值由一個(gè)Supplier提供Observable.java#L8985public final NonNull R SingleR reduceWith(NonNull SupplierR seedSupplier, NonNull BiFunctionR, ? super T, R reducer)這使得每個(gè)訂閱者都可以擁有獨(dú)立創(chuàng)建的累積容器。Observable.just(1, 2, 2, 3, 4, 4, 4, 5) .reduceWith(TreeSet::new, (set, x) - { set.add(x); return set; }) .subscribe(System.out::println); // prints [1, 2, 3, 4, 5]collect將源Observable發(fā)出的物品收集進(jìn)一個(gè)由你創(chuàng)建的可變數(shù)據(jù)結(jié)構(gòu)返回發(fā)出該結(jié)構(gòu)的SingleObservable.java#L5455。Observable.just(Kirk, Spock, Chekov, Sulu) .collect(() - new StringJoiner( \uD83D\uDD96 ), StringJoiner::add) .map(StringJoiner::toString) .subscribe(System.out::println); // prints Kirk Spock Chekov Sulu簽名中第一個(gè)參數(shù)是Supplier? extends U initialItemSupplier第二個(gè)參數(shù)是BiConsumer? super U, ? super T collector。與reduceWith不同collect的回調(diào)不需要返回值——容器就地修改即可代碼更簡(jiǎn)潔。Java 8 還可用Collector風(fēng)格的collectObservable.java#L14495其實(shí)現(xiàn)位于 ObservableCollectWithCollector.java 與 ObservableCollectWithCollectorSingle.java可以復(fù)用 JDK 標(biāo)準(zhǔn)庫中的Collector實(shí)例。collectInto與collect類似但初始容器由調(diào)用方直接傳入而非由Supplier創(chuàng)建Observable.java#L5490。注意將用于收集物品的可變值如下例的StringBuilder會(huì)在多個(gè)訂閱者之間共享。Observable.just(R, x, J, a, v, a) .collectInto(new StringBuilder(), StringBuilder::append) .map(StringBuilder::toString) .subscribe(System.out::println); // prints RxJava這一“共享可變狀態(tài)”的特性意味著collectInto更適合一次性消費(fèi)的冷場(chǎng)景若被多次訂閱多個(gè)訂閱者會(huì)并發(fā)寫入同一容器。從源碼結(jié)構(gòu)看collect的返回類型是SingleU每個(gè)訂閱觸發(fā)新建容器而collectInto同樣返回Single但對(duì)熱源或重復(fù)訂閱需要自行處理容器的可見性與線程安全。toList收集Observable的所有物品作為單個(gè)List發(fā)出Observable.java#L12758。Observable.just(2, 1, 3) .toList() .subscribe(System.out::println); // prints [2, 1, 3]除無參版本外還有兩個(gè)實(shí)用重載toList(int capacityHint)Observable.java#L12791提示內(nèi)部ArrayList的初始容量在已知數(shù)據(jù)規(guī)模時(shí)可減少擴(kuò)容開銷toList(SupplierU collectionSupplier)Observable.java#L12826用自定義集合實(shí)現(xiàn)如LinkedList、TreeSet等替代默認(rèn)ArrayList。toSortedList收集所有物品并作為已排序的List發(fā)出。默認(rèn)使用自然排序也可指定Comparator與capacityHinttoSortedList()Observable.java#L13168內(nèi)部使用Functions.naturalComparator()toSortedList(Comparator)Observable.java#L13196toSortedList(Comparator, int capacityHint)Observable.java#L13229toSortedList(int capacityHint)Observable.java#L13264。Observable.just(2, 1, 3) .toSortedList(Comparator.reverseOrder()) .subscribe(System.out::println); // prints [3, 2, 1]toMap將源發(fā)出的序列轉(zhuǎn)換為以指定 key 函數(shù)為鍵的Map。提供三個(gè)重載toMap(keySelector)Observable.java#L12858物品本身作為 valuetoMap(keySelector, valueSelector)Observable.java#L12893分別指定 key 與 value 選擇器toMap(keySelector, valueSelector, mergeFunction)Observable.java#L12930遇到重復(fù) key 時(shí)用 merge 函數(shù)合并而不是拋異常。Observable.just(1, 2, 3, 4) .toMap((x) - { // defines the key in the Map return x; }, (x) - { // defines the value that is mapped to the key return (x % 2 0) ? even : odd; }) .subscribe(System.out::println); // prints {1odd, 2even, 3odd, 4even}使用提示如果源中可能出現(xiàn)重復(fù) key 且你不關(guān)心合并規(guī)則應(yīng)選擇帶mergeFunction的重載避免下游收到IllegalStateException。toMultimap將源發(fā)出的序列轉(zhuǎn)換為一個(gè)既是Collection又是Map的結(jié)構(gòu)——相同 key 對(duì)應(yīng)多個(gè) value天然規(guī)避了toMap的重復(fù) key 問題。同樣提供單/雙選擇器及帶容器工廠的重載Observable.java#L12964 至 Observable.java#L13080。Observable.just(1, 2, 3, 4) .toMultimap((x) - { // defines the key in the Map return (x % 2 0) ? even : odd; }, (x) - { // defines the value that is mapped to the key return x; }) .subscribe(System.out::println); // prints {even[2, 4], odd[1, 3]}源碼級(jí)實(shí)現(xiàn)與選型小結(jié)從源碼結(jié)構(gòu)看上述標(biāo)準(zhǔn)聚合算子在核心庫Observable與Flowable中是對(duì)稱提供的Flowable對(duì)應(yīng)實(shí)現(xiàn)位于src/main/java/io/reactivex/rxjava4/internal/operators/flowable/下同名類且聚合路徑普遍實(shí)現(xiàn)了FuseToObservable如 ObservableCountSingle.java、ObservableCollectSingle.java即在與相鄰算子組合時(shí)可以跳過中間隊(duì)列、直接同步傳遞結(jié)果降低訂閱延遲。選型時(shí)可以按以下決策鏈快速判斷需求推薦算子內(nèi)存特征返回類型只統(tǒng)計(jì)條數(shù)count()O(1)不緩沖物品SingleLong純累積運(yùn)算和、積、最大、拼接reduce/reduceWithO(1) 累積值SingleR收集到自定義可變結(jié)構(gòu)StringJoiner等collect/collectIntoO(n)SingleU收集為標(biāo)準(zhǔn)List可指定容量/集合類型toListO(n)SingleListT需要排序結(jié)果toSortedListO(n)SingleListT一對(duì)一按鍵索引toMap重 key 用 merge 重載O(n)SingleMapK, V一對(duì)多按鍵分組toMultimapO(n)SingleMapK, CollectionV數(shù)值型平均/求和核心庫未覆蓋的類型MathObservable/MathFlowable的average*、sum*O(1)~O(n)Observable/Flowable最后重申使用邊界所有聚合算子都要等上游終止才能產(chǎn)出結(jié)果并通常緩沖整個(gè)序列count、reduce這類僅需累積值的除外因此對(duì)長(zhǎng)流或無限流應(yīng)謹(jǐn)慎使用需要外部依賴的數(shù)學(xué)算子rxjava2-extensions請(qǐng)確認(rèn)項(xiàng)目中已引入該模塊而count、reduce、toList、toMap、toMultimap、toSortedList、collect、collectInto均為 RxJava 核心庫自帶Maybe額外提供count。【免費(fèi)下載鏈接】RxJavaRxJava – Reactive Extensions for the JVM – a library for composing asynchronous and event-based programs using observable sequences for the Java VM.項(xiàng)目地址: https://gitcode.com/gh_mirrors/rx/RxJava創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考