RxJava 核心結構與六大實戰場景速覽

前言

RxJava 以「觀察者模式 + 操作符鏈」重塑異步編程體驗。本文先拆解 Observable.create/subscribe 底層流程，再列舉 Android/後端開發中六大高頻場景的 RxJava 解決方案。

一、核心結構最小化拆解

1.1 三大核心角色

• Observable：事件源，持有 OnSubscribe 回調。
• Subscriber：消費者，實現 onNext/onError/onCompleted。
• Subscription：訂閱關係持有者，支持 unsubscribe() 取消。

1.2 創建與訂閱流程

// 1. 創建 Observable，保存 onSubscribe 回調
Observable.create(onSubscriber1)  // 返回 Observable 實例 observable1

// 2. 訂閱
observable1.subscribe(subscriber1) 
// 內部調用：hook.onSubscribeStart(observable, observable.onSubscribe).call(subscriber)
// 即：onSubscriber1.call(subscriber1) → 觸發 onNext/onCompleted

本質：Observable 只是「配方」，subscribe 才真正「開火」，實現惰性求值。

二、六大實戰場景

場景 1：多級緩存降級 —— 內存 → 磁盤 → 網絡

需求：任一級命中即終止後續查詢。
方案：Observable.concat(memory, disk, network).first()。
concat 保證順序串行，first() 取首個發射項後自動取消後續。

Observable.concat(
    Observable.create(s -> { if (memCache!=null) s.onNext(memCache); else s.onCompleted(); }),
    Observable.create(s -> { String disk = ...; if (!empty) s.onNext(disk); else s.onCompleted(); }),
    Observable.just("network")
).first()
.subscribeOn(Schedulers.io())
.subscribe(s -> { /* 使用數據 */ });

場景 2：多接口並發聚合 —— 界面等待多個請求完成

方案：Observable.merge(api1, api2, ...) 合併流，不保證順序，誰先完成誰先發射。配合 toList() 或 zip 聚合結果。

Observable.merge(
    api1.subscribeOn(Schedulers.io()),
    api2.subscribeOn(Schedulers.io())
).subscribeOn(Schedulers.io())
.subscribe(result -> { /* 更新 UI */ });

場景 3：鏈式依賴調用 —— 登錄獲取 Token 後請求數據

痛點：傳統回調嵌套導致「Callback Hell」。
方案：flatMap 將上游發射項轉換為新 Observable 並扁平化合併。

NetworkService.getToken("user", "pass")
    .flatMap(token -> NetworkService.getMessage(token))
    .subscribe(msg -> { /* 顯示消息 */ });

場景 4：按鈕防抖 —— 避免連續點擊重複提交

方案：RxView.clicks(btn).throttleFirst(1, TimeUnit.SECONDS)。
throttleFirst：每個時間窗口只取首次事件，其餘丟棄；對比 debounce（靜默期後取最後一次）。

場景 5：響應式 UI 綁定 —— CheckBox 狀態同步到 SharedPreferences

方案：RxCompoundButton.checkedChanges(cb).subscribe(rxPref.getBoolean("key").asAction())。
RxPreferences 將 Preference 封裝為 Action，實現雙向綁定。
同理適用： RxTextView.textChanges、RxAdapterView.itemClicks 等。

場景 6：複雜數據流變換 —— 集合過濾、去重、聚合

Observable.just("1","2","2","3","4","5")
    .map(Integer::parseInt)
    .filter(i -> i > 1)
    .distinct()
    .take(3)
    .reduce((a,b) -> a + b)  // 2+3+4=9
    .subscribe(System.out::println);

操作符鏈將命令式迴圈轉為聲明式流水線，可讀性與可組合性大增。

三、核心操作符分類速查

四、避坑指南

1. 線程切換：subscribeOn 決定上游（創建端）線程，observeOn 決定下游（消費端）線程，可多次切換。
2. 內存洩漏：Activity/Fragment 銷毀時調用 compositeDisposable.clear() 取消訂閱，防止持有 Context。
3. 背壓：RxJava 2+ 引入 Flowable 支持背壓，高速產生/慢速消費場景必用。
4. 異常處理：onErrorResumeNext/retry/retryWhen 實現降級重試，避免流中斷。

RxJava 學習曲線陡峭，但掌握核心「流+操作符」思維後，能將複雜異步邏輯化繁為簡。