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