您现在的位置：程式碼之家 >> 編程語言 > >> Java 程式設計 >> Java 相關知識

RxJava 實戰場景總結：延遲執行、線程切換、優先級資料源與鏈式調用優化

編輯：Java 相關知識

場景一：延遲執行動作

需求：5ms 後執行某邏輯並發射結果。

Observable.timer(5, TimeUnit.MILLISECONDS)
    .map(v -> doSomething())
    .subscribe(System.out::println);

場景二：立即執行、延遲發射結果

需求：doSomething() 立即執行，但結果 5ms 後才下發給下游。

Observable.zip(
    Observable.timer(5, TimeUnit.MILLISECONDS),
    Observable.just(doSomething()), // 立即執行
    (t, result) -> result           // 取第二個源的值
).subscribe(System.out::println);

原理：zip 等待所有源在相同位置發射數據後才組合發射。第一個源 timer 延遲發射，第二個源 just 立即就緒，最終結果被延遲推送。

場景三：指定線程執行特定動作（defer 實現）

問題：Observable.just(doSomething()) 中的 doSomething() 會在組裝鏈時（即調用 just 時）在當前線程立即執行，而非訂閱線程。

錯誤示範

Observable.just(doSomething()) // 主線程立即執行！
    .subscribeOn(Schedulers.io())
    .observeOn(Schedulers.computation())
    .subscribe(v -> log(v));

正確寫法：create / defer

// 方案 A：create 手動控制
Observable.create(emitter -> {
    emitter.onNext(doSomething()); // 在 subscribeOn 指定的線程執行
    emitter.onComplete();
})
.subscribeOn(Schedulers.io())
.observeOn(Schedulers.computation())
.subscribe(v -> log(v));

// 方案 B：defer 延遲創建 Observable（推薦，語義更清晰）
Observable.defer(() -> Observable.just(doSomething()))
    .subscribeOn(Schedulers.io())
    .observeOn(Schedulers.computation())
    .subscribe(v -> log(v));

defer 確保 Observable.just(...) 在訂閱時刻才創建，從而將 doSomething() 執行推遲到 subscribeOn 指定的線程。

場景四：compose 復用線程調度邏輯，保持鏈式風格

痛點：到處重複 .subscribeOn(Schedulers.io()).observeOn(Schedulers.computation()) 破壞鏈式結構。

Transformer 封裝

// 定義復用轉換器
private static  ObservableTransformer applySchedulers() {
    return upstream -> upstream
        .subscribeOn(Schedulers.io())
        .observeOn(Schedulers.computation());
}

// 使用：compose 內聯，鏈式不中斷
Observable.just(doSomething())
    .compose(applySchedulers())
    .subscribe(v -> log(v));

Lambda 進一步簡化：return upstream -> upstream.subscribeOn(...).observeOn(...);

場景五：優先級資料源（緩存優先、網絡兜底）

需求：先讀緩存，有數據直接返回；無數據再請網絡。且緩存命中時網絡請求不發起。

concat + first 實現

Observable.concat(
    getDataFromCache(),     // 先發射緩存源
    getDataFromNetwork()    // 緩存無數據(onComplete)才訂閱網絡源
)
.first()                    // 取第一個發射的數據
.subscribe(v -> System.out.println("result: " + v));

// 緩存源
Observable getDataFromCache() {
    return Observable.create(emitter -> {
        String cached = loadFromCache();
        if (cached != null) emitter.onNext(cached);
        emitter.onComplete(); // 必須 onComplete，否則 concat 會卡住等待
    });
}

// 網絡源
Observable getDataFromNetwork() {
    return Observable.create(emitter -> {
        String net = requestNetwork();
        emitter.onNext(net);
        emitter.onComplete();
    });
}

關鍵細節

concat 按順序串聯，前一個 onComplete 後才訂閱下一個。
first() 取首個元素並完成，後續源不再被訂閱（網絡請求不發起）。
若兩源均無數據，first() 拋 NoSuchElementException，可用 firstOrDefault(defaultVal) 避免。
緩存源若用 onError 表示無數據，會導致 concat 終止，需改用 concatDelayError 將錯誤延遲到最後處理。