RxJava 實(shí)戰(zhàn)問題求解用 Observable 操作符鏈解決 Euler 求和與斐波那契數(shù)列生成【免費(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本篇文章圍繞 RxJava 倉庫中 Problem-Solving-Examples-in-RxJava.md 提供的兩道經(jīng)典編程謎題展開用 RxJava 操作符求解 Project Euler 第 1 題3 與 5 的倍數(shù)求和以及用純操作符組合生成斐波那契數(shù)列。通過逐步拆解range、filter、map、takeWhile、merge、distinct、reduce、create、repeat、scan、take等操作符的調(diào)用鏈讀者將掌握用聲明式數(shù)據(jù)流思維替代命令式循環(huán)的 RxJava 解題范式并理解這些操作符在當(dāng)前倉庫源碼中的真實(shí)實(shí)現(xiàn)行為。從兩道謎題理解 RxJava 的思考方式RxJava 的核心思想是把異步與事件驅(qū)動(dòng)的程序組織成可組合的可觀察序列observable sequences。傳統(tǒng)的命令式寫法用循環(huán)和中間變量描述怎么做而 RxJava 用操作符鏈描述要什么——數(shù)據(jù)源從上游流出經(jīng)過一個(gè)個(gè)操作符的變換、過濾、合并、聚合最終被訂閱者消費(fèi)。下面的兩個(gè)例子恰好覆蓋了 RxJava 操作符的幾大典型類別創(chuàng)建型range、create、fromArray、repeat變換型map、scan過濾型filter、takeWhile、take、distinct合并型merge聚合型reduceProject Euler 第 1 題3 或 5 的倍數(shù)求和原題是這樣的熱身題如果我們列出 10 以下所有是 3 或 5 倍數(shù)的自然數(shù)會(huì)得到 3、5、6 和 9。這些數(shù)的和是 23。請(qǐng)找出 1000 以下所有是 3 或 5 的倍數(shù)的自然數(shù)之和。答案是 233168。下面用兩種 RxJava 思路分別求解。思路一rangefilter從全集里篩最直接的做法是枚舉 1000 以下的所有自然數(shù)然后用filter保留那些是 3 或 5 的倍數(shù)的項(xiàng)ObservableInteger threesAndFives Observable.range(1, 999).filter(e - e % 3 0 || e % 5 0);在 Groovy 中寫法幾乎一樣條件取反來兼容 Groovy 的 truthinessdef threesAndFives Observable.range(1,999).filter({ !((it % 3) (it % 5)) });其中range(start, count)是 RxJava 的靜態(tài)創(chuàng)建操作符range(1, 999)會(huì)按順序發(fā)射 1 到 999 共 999 個(gè)整數(shù)。從當(dāng)前倉庫源碼看Observable.range 會(huì)對(duì)參數(shù)做嚴(yán)格校驗(yàn)count 0直接拋出IllegalArgumentException(count 0 required but it was ...)count 0返回empty()count 1退化為just(start)并且檢測(cè)start count - 1是否會(huì)溢出Integer.MAX_VALUE最終包裝為內(nèi)部的ObservableRange并經(jīng)過RxJavaPlugins.onAssembly裝配后返回。filter則接收一個(gè)Predicate? super TObservable.filter 對(duì)每個(gè)通過謂詞判斷的元素向下游發(fā)射其余元素被丟棄。思路二maptakeWhilemergedistinct從倍數(shù)集合合并另一種做法是構(gòu)造兩個(gè)序列一個(gè)專門生成 3 的倍數(shù)一個(gè)專門生成 5 的倍數(shù)。方法是先用map把每個(gè)自然數(shù)映射成它的倍數(shù)再用takeWhile在倍數(shù)小于 1000 時(shí)截?cái)嘧詈笥胢erge合并兩條流ObservableInteger threes Observable.range(1, 999).map(e - e * 3).takeWhile(e - e 1000); ObservableInteger fives Observable.range(1, 999).map(e - e * 5).takeWhile(e - e 1000); ObservableInteger threesAndFives Observable.merge(threes, fives).distinct();def threes Observable.range(1,999).map({it*3}).takeWhile({it1000}); def fives Observable.range(1,999).map({it*5}).takeWhile({it1000}); def threesAndFives Observable.merge(threes, fives).distinct();兩個(gè)細(xì)節(jié)值得注意takeWhile在這里是邊界守衛(wèi)map會(huì)把 1..999 映射成 3..2997但我們只關(guān)心 1000 以下的倍數(shù)takeWhile(e - e 1000)一旦遇到第一個(gè)不滿足條件的值就停止發(fā)射并完成。這也是 Observable.takeWhile 的標(biāo)準(zhǔn)語義——逐項(xiàng)檢查謂詞遇到首個(gè)為false的項(xiàng)即終止序列。不能漏掉distinct15、30 這類數(shù)同時(shí)是 3 和 5 的倍數(shù)會(huì)同時(shí)出現(xiàn)在兩條流里merge只是把兩條流的時(shí)間線交錯(cuò)合并并不會(huì)去重。distinct()見 Observable.distinct內(nèi)部維護(hù)一個(gè)HashSet保證每個(gè)值只向下游發(fā)射一次。merge的實(shí)現(xiàn)也值得看一下靜態(tài)重載 Observable.merge(ObservableSource? extends ObservableSource? extends T) 在底層復(fù)用了ObservableFlatMap以Functions.identity()作為映射函數(shù)、Integer.MAX_VALUE作為最大并發(fā)訂閱數(shù)——也就是說merge本質(zhì)上是不做變換的無限并發(fā) flatMap上游多條流交錯(cuò)地向下游推數(shù)據(jù)。求和reduce把序列折疊成一個(gè)數(shù)現(xiàn)在得到了去重后的倍數(shù)序列接下來要把它求和。如果你安裝了可選的rxjava-math模塊可以直接用sumInteger/sumLong之類的數(shù)學(xué)操作符見 Mathematical-and-Aggregate-Operators。但如果不想依賴這個(gè)模塊標(biāo)準(zhǔn) RxJava 里最合適的聚合操作符就是reduceSingleInteger summer threesAndFives.reduce(0, (a, b) - a b);def summer threesAndFives.reduce(0, { a, b - ab });注意這里的返回類型當(dāng)前倉庫中 Observable.reduce(R seed, BiFunction) 返回的是SingleR不是Observable——這正是 RxJava 3/4 時(shí)代的類型設(shè)計(jì)一個(gè)確定會(huì)恰好發(fā)射一個(gè)結(jié)果并完成的流用Single表達(dá)語義更準(zhǔn)確。其底層被包裝為ObservableReduceSeedSingle。reduce的折疊過程如下以0作為種子值seed每當(dāng)上游threesAndFives發(fā)射一個(gè)元素b就調(diào)用累加函數(shù)(a, b) - a b把當(dāng)前種子a和發(fā)射值相加返回的新值覆蓋種子當(dāng)上游完成時(shí)把最終的種子值作為唯一一次發(fā)射推給下游。下表演示了折疊過程的前幾步與最后一步整個(gè)序列共有 466 個(gè)去重后的倍數(shù)iterationseedemissionreduce1033235838614…………466232169999233168訂閱輸出到目前為止我們只是搭建了一條管道還沒有人消費(fèi)它。RxJava 是惰性的只有調(diào)用subscribe才會(huì)真正觸發(fā)數(shù)據(jù)流動(dòng)summer.subscribe(System.out::print);summer.subscribe({println(it);});Single.subscribe(Consumer)在成功時(shí)把唯一的結(jié)果233168交給回調(diào)因此上面這行會(huì)直接打印出答案。這正體現(xiàn)了構(gòu)建數(shù)據(jù)流與訂閱數(shù)據(jù)流兩個(gè)階段的分工。生成斐波那契數(shù)列第二個(gè)謎題是如何創(chuàng)建一個(gè)發(fā)射斐波那契數(shù)列的Observable方法一create 循環(huán)從頭造一個(gè) Observable最直接的方式是用create操作符從零開始制造一個(gè)Observable在傳給它的回調(diào)里用傳統(tǒng)循環(huán)生成序列ObservableInteger fibonacci Observable.create(emitter - { int f1 0, f2 1, f 1; while (!emitter.isDisposed()) { emitter.onNext(f); f f1 f2; f1 f2; f2 f; } });def fibonacci Observable.create({ emitter - def f10, f21, f1; while(!emitter.isDisposed()) { emitter.onNext(f); f f1f2; f1 f2; f2 f; }; });這種寫法能工作但它太像普通線性編程了——本質(zhì)上還是命令式的循環(huán)只是把輸出換成了emitter.onNext(...)。而且注意這個(gè)循環(huán)只有在emitter.isDisposed()即訂閱者取消訂閱時(shí)才會(huì)停下所以在實(shí)際使用中一定要配合take(n)限制發(fā)射數(shù)量否則它會(huì)無限發(fā)射下去。從源碼看Observable.create 接收一個(gè)ObservableOnSubscribeT這里用 lambda 實(shí)現(xiàn)通過Objects.requireNonNull拒絕空值并包裝為內(nèi)部類ObservableCreate后經(jīng)插件裝配返回。emitter.isDisposed()對(duì)應(yīng)ObservableEmitter的取消訂閱狀態(tài)檢查——這正是響應(yīng)式取消在自定義序列里的典型用法。方法二純操作符組合用repeatscanmap驅(qū)動(dòng)狀態(tài)機(jī)有沒有更Rx 風(fēng)格的做法文檔給出的方案是用一個(gè)只發(fā)射 0 的無限序列作為磨盤驅(qū)動(dòng)scan這個(gè)狀態(tài)機(jī)不斷推進(jìn)ObservableInteger fibonacci Observable.fromArray(0) .repeat() .scan(new int[]{0, 1}, (a, b) - new int[]{a[1], a[0] a[1]}) .map(a - a[1]);def fibonacci Observable.from(0).repeat().scan([0,1], { a,b - [a[1], a[0]a[1]] }).map({it[1]});這個(gè)方法略顯取巧原文稱之為 janky但拆開看每一步都很有教學(xué)價(jià)值fromArray(0).repeat()先發(fā)射一個(gè)0然后repeat()反復(fù)重放這個(gè)序列Observable.repeat 在源完成后自動(dòng)重新訂閱于是得到一條無限長(zhǎng)的全 0 流。這些 0 本身毫無意義純粹是給scan提供事件節(jié)拍。scan(new int[]{0, 1}, (a, b) - new int[]{a[1], a[0] a[1]})scan與reduce類似都帶種子值并逐項(xiàng)累積但區(qū)別在于scan把每一步的中間結(jié)果都發(fā)射出去而reduce只在最后發(fā)射一次。這里的累加器完全忽略上游傳來的b反正都是 0只把寄存器從[a, b]變換為[b, ab]。帶初值的重載對(duì)應(yīng)源碼 Observable.scan(R initialValue, BiFunction)。map(a - a[1])寄存器按如下順序演變[0,1] → [1,1] → [1,2] → [2,3] → [3,5] → [5,8] → …數(shù)組中第二個(gè)元素恰好就是斐波那契數(shù)列 1, 1, 2, 3, 5, 8, …用map把它單獨(dú)取出來即可。也就是說scan在這里扮演了一個(gè)隨每次輸入變換內(nèi)部狀態(tài)并對(duì)外發(fā)布狀態(tài)快照的有狀態(tài)變換器這正是用操作符組合替代顯式循環(huán)變量的關(guān)鍵技巧。打印前 15 項(xiàng)無論用哪種方法生成輸出前 15 項(xiàng)都一樣簡(jiǎn)單fibonacci.take(15).subscribe(System.out::println);fibonnaci.take(15).subscribe({println(it)})];take(15)在發(fā)射了 15 個(gè)元素后立即讓上游完成并斷開訂閱見 Observable.take這對(duì)于上述無限序列至關(guān)重要——它既是只取一部分的需求表達(dá)也是防止無限循環(huán)泄漏資源的機(jī)制。有沒有更優(yōu)雅的方案create版本太命令式repeatscan版本又有點(diǎn)繞它專門造了一條毫無信息量的全 0 流只是為了讓scan的曲柄轉(zhuǎn)起來。原文檔提到一個(gè)名為generate的操作符本可以更優(yōu)雅地解決這個(gè)問題——它允許用一個(gè)狀態(tài) 生成函數(shù)直接描述迭代式序列無需借助無意義的占位流但這個(gè)操作符在當(dāng)時(shí)尚未并入 RxJava。作為練習(xí)你可以嘗試思考更簡(jiǎn)潔的組合方式例如用concatMap配合狀態(tài)對(duì)象或在create回調(diào)里維護(hù)狀態(tài)但保持聲明式風(fēng)格。這道開放題本身就是理解 RxJava 操作符組合力的最好訓(xùn)練。從兩道題看 RxJava 的通用解題套路把上面的兩個(gè)例子放在一起可以提煉出 RxJava 處理數(shù)學(xué)計(jì)算 事件流問題的通用模式選擇數(shù)據(jù)源有限范圍用range/fromArray無限序列用create或repeat 占位流。變換與篩選map做一對(duì)一映射filter按謂詞剔除takeWhile/take控制序列邊界distinct去重。合并多源merge交錯(cuò)多條流底層是flatMap identity 映射默認(rèn)無限并發(fā)。聚合收口要每一步都看中間結(jié)果用scan只要最終一個(gè)結(jié)果用reduce注意當(dāng)前倉庫中帶種子值的reduce返回Single。訂閱觸發(fā)所有鏈?zhǔn)綐?gòu)建都是惰性的subscribe才是數(shù)據(jù)流動(dòng)的起點(diǎn)對(duì)無限序列務(wù)必配合take之類的終止操作符。這些操作符的語義細(xì)節(jié)參數(shù)校驗(yàn)、返回值類型、惰性機(jī)制都可以在 src/main/java/io/reactivex/rxjava4/core/Observable.java 中找到對(duì)應(yīng)實(shí)現(xiàn)配套的測(cè)試用例如 flowable/FlowableReduceTests.java、flowable/FlowableConcatTests.java則從行為層面驗(yàn)證了這些語義。如果你想繼續(xù)深入倉庫的 Observable.md 匯總了訂閱協(xié)議Creating-Observables、Transforming-Observables、Filtering-Observables、Combining-Observables 與 Mathematical-and-Aggregate-Operators 則分別覆蓋了本文用到的各類操作符。【免費(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），僅供參考