前言：響應式編程範式
在異步事件流、實時數據處理日益普及的今天，傳統命令式編程面臨回調地獄、線程管理複雜、錯誤傳播困難等挑戰。RxJava 作為 ReactiveX 在 JVM 平台的實現，引入了觀察者模式、迭代器模式與函數式編程的融合體系，提供了一套聲明式、可組合、線程安全的異步編程框架。
核心概念
三大核心角色
- Observable（可觀察者/被觀察者）：事件源，發射數據流（0~n 個
onNext）、終止信號（onCompleted 或 onError）。
- Observer / Subscriber（觀察者/訂閱者）：消費者，實現
onNext(T)、onCompleted()、onError(Throwable) 三個回調。
- Subscription（訂閱關係）：連接雙方的句柄，支持
unsubscribe() 取消訂閱，防止內存洩漏。
Observer vs Subscriber
Subscriber 是實現 Observer 接口的抽象類，額外提供：
onStart()：訂閱開始前調用，用於初始化重置，但執行在 subscribe() 所在線程，不適合 UI 操作。unsubscribe() / isUnsubscribed()：主動切斷訂閱，釋放引用，**必須在 Activity/Fragment 銷毀時調用**，否則 Observable 持有引用導致內存洩漏。
需指定線程做準備工作時，使用 doOnSubscribe() 操作符。
操作符全景分類
操作符賦予數據流函數式轉換能力，鏈式調用構建處理管道。
1. 創建型操作符
| 操作符 | 用途 | 示例 |
|---|
| create() | 底層構建，手動調用 onNext/onCompleted/onError | 自定義事件源 |
| just() | 將對象/集合轉為發射單個序列的 Observable | Observable.just(1,2,3) |
| from() | 數組/Iterable/Future 轉 Observable | from(array) |
| interval() | 定時發射整數序列 | interval(1, TimeUnit.SECONDS) |
| timer() | 延遲發射單值 | timer(5, TimeUnit.SECONDS) |
| range() | 發射指定範圍整數 | range(1, 10) |
| repeat() | 重複發射序列 | repeat(3) |
| defer() | 訂閱時才創建，每次新實例 | 延遲初始化 |
2. 變換型操作符
| 操作符 | 語義 | 類比 |
|---|
| map() | 一對一元素變換 | Stream.map() |
| flatMap() | 一對多展平：元素→Observable→合併單流 | 異步請求鏈、嵌套集合攤平 |
| concatMap() | flatMap 有序版，保持發射順序 | 順序敏感場景 |
| buffer() | 定期收集元素打包成 List 發射 | 批量處理、緩衝 |
| window() | 切分為子 Observable 窗口 | 時間/數量分窗 |
| groupBy() | 按 Key 分組為多個 Observable | 分組聚合 |
| scan() | 累積狀態（如 reduce 但發射中間值） | 運行總和、狀態機 |
flatMap 核心：將每個元素映射為 Observable，再將所有內部 Observable 合併（merge）成單一流。適用於「異步請求依賴上游結果」場景。
3. 過濾型操作符
| 操作符 | 語義 |
|---|
| filter() | 謂詞判斷，保留通過項 |
| take(n) | 取前 n 項 |
| skip(n) | 跳過前 n 項 |
| takeLast(n) | 取後 n 項 |
| skipLast(n) | 跳過後 n 項 |
| distinct() | 去重（基於 equals/hashCode） |
| elementAt(n) | 取指定索引項 |
| first() / last() | 首/尾項（可帶謂詞） |
| debounce() | 防抖：靜默期後發射最新值 |
| throttleFirst() / sample() | 節流：固定間隔取樣 |
| ignoreElements() | 僅保留終止通知 |
4. 組合型操作符
| 操作符 | 語義 |
|---|
| merge() | 多流交織合併（無序） |
| concat() | 順序串聯（前序完成才訂閱後序） |
| zip() | 按索引打包多流元素（最短流決定長度） |
| combineLatest() | 任一流發射則組合各流最新值 |
| startWith() | 前置發射指定序列 |
| switchOnNext() | 切換到最新內部 Observable |
| join() / and-then-when() | 基於時間窗口關聯 |
5. 錯誤處理操作符
| 操作符 | 策略 |
|---|
| onErrorReturn() | 發射默認值並正常完成 |
| onErrorResumeNext() | 切換到備用 Observable |
| retry(n) | 重訂閱上游 n 次 |
| retryWhen() | 自定義重試邏輯（延遲、條件） |
6. 輔助/工具操作符
| 操作符 | 用途 |
|---|
| subscribeOn(Scheduler) | 指定上游（發射端）執行線程 |
| observeOn(Scheduler) | 指定下游（消費端）執行線程 |
| doOnNext/doOnSubscribe/doOnError... | 副作用回調（日誌、進度條、埋點） |
| delay() | 整體延遲發射 |
| timeout() | 超時發射錯誤 |
| timestamp() / timeInterval() | 添加時間戳/間隔 |
| serialize() | 強制順序發射 |
| using() | 綁定資源生命週期（自動釋放） |
7. 條件/布爾/算術聚合
- 條件：
all()、contains()、amb()（取最先發射流）、defaultIfEmpty()、sequenceEqual()、skipUntil/takeUntil()、skipWhile/takeWhile()。
- 聚合：
count()、sum()、average()、reduce()、max()、min()。
8. 連接型操作符（ConnectableObservable）
| 操作符 | 語義 |
|---|
| publish() | 轉為可連接熱 Observable |
| connect() | 手動觸發發射 |
| refCount() | 訂閱數>0 時自動 connect，歸零自動斷開 |
| replay() | 緩存重放，晚訂閱者收到歷史數據 |
9. 轉換型操作符
toList()、toMap()、toSortedList()：收集為集合。blockingFirst()、blockingIterable()、blockingSubscribe()：阻塞式轉同步（僅測試/主線程慎用）。
線程調度最佳實踐
Observable.just("input")
.subscribeOn(Schedulers.io()) // IO 線程執行上游耗時操作
.map(this::heavyComputation)
.observeOn(AndroidSchedulers.mainThread()) // 主線程更新 UI
.subscribe(result -> updateUI(result),
error -> handleError(error));
subscribeOn：最近調用者生效，多次調用**只有第一個生效**，決定源頭線程。observeOn：每次調用切換下游線程，可多次切換。- 常用調度器：
Schedulers.io()（IO）、computation()（CPU 密集）、newThread()、single()、trampoline()（當前線程排隊）。
內存洩漏防範清單
- 持有
Subscription 引用，在 onDestroy()/onStop() 調用 unsubscribe()。
- 使用
CompositeSubscription 批量管理多個訂閱。
- RxJava 2.x 改用
Disposable，配合 CompositeDisposable。
- 避免在長生命週期 Observable 中引用短生命週期對象（Activity/View）。
總結
RxJava 以「一切皆流」為核心，通過豐富操作符實現異步邏輯的聲明式組合。掌握創建、變換、過濾、組合、錯誤處理、線程調度六大類操作符，理解冷/熱 Observable、背壓機制（RxJava 2.x Flowable）、訂閱生命週期管理，是編寫健壯、可維護響應式應用的關鍵。實戰中建議從簡單鏈式調用入手，逐步引入複雜操作符，並始終關注線程切換與資源釋放。
