RxJava 操作符大全：過濾類操作符詳解與實戰範例

建立 Observable 的基礎操作符

just(T...)：發射固定序列

Observable.just("1", "2", "3")
    .subscribe(s -> System.out.println(s));
// 輸出: 1 2 3

repeat(n)：重複發射序列

Observable.just("1", "2", "3").repeat(3)
    .subscribe(s -> System.out.print(s));
// 輸出: 123123123

defer(() -> Observable)：延遲建立（訂閱時才建立）

Observable.defer(() -> {
    System.out.println("create");
    return Observable.just(1);
}).subscribe();
// 僅在 subscribe 時印出 "create"

range(start, count)：發射連續整數

Observable.range(3, 2).subscribe(i -> System.out.print(i));
// 輸出: 34

核心過濾操作符

filter(predicate)：條件篩選

Observable.range(0, 50)
    .filter(i -> i % 3 == 0)
    .subscribe(i -> System.out.print(i + " "));
// 輸出: 0 3 6 9 12 15 18 21 24 27 30 33 36 39 42 45 48

take(n) / takeLast(n)：取前/後 N 個

Observable.range(0, 50).take(3).subscribe(...);       // 0 1 2
Observable.range(0, 50).takeLast(3).subscribe(...);   // 47 48 49

distinct()：全局去重（需記住所有已發射值，大數據慎用）

Observable.just(1,2,3).repeat(3).distinct().subscribe(...); // 1 2 3

distinctUntilChanged()：相鄰去重（僅比較前一值）

Observable.just(1,1,2,2,1).distinctUntilChanged().subscribe(...); // 1 2 1

first() / last()：取首/尾（可帶條件）

Observable.range(0, 50)
    .last(i -> i % 3 == 0 && i != 0)
    .subscribe(i -> System.out.println(i)); // 48

skip(n) / skipLast(n)：跳過前/後 N 個

Observable.range(0, 50).skip(3).skipLast(3).subscribe(...); // 3..46

elementAt(index) / elementAtOrDefault(index, default)：取指定位置

Observable.range(0, 50).take(3).elementAtOrDefault(5, 999).subscribe(...); // 999 (越界回傳預設值)

時間相關操作符

interval(delay, unit)：定期輪詢（預設 computation 執行緒）

Observable.interval(3, TimeUnit.SECONDS)
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(l -> Log.i("info", l.toString()));
// 每 3 秒發射 0,1,2...

進階：遍歷列表並定期發射

List list = Arrays.asList("aa","bb","cc","dd");
Observable.interval(3, TimeUnit.SECONDS)
    .flatMap(i -> Observable.just(list.get(i.intValue())))
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new Subscriber() {
        public void onNext(String s) { Log.i("info", "onNext--" + s); }
        public void onError(Throwable e) { Log.i("info", "error"); }
        public void onCompleted() { Log.i("info", "completed"); }
    });
// 輸出: 每 3 秒印一個，最後 onError (索引越界)

timer(delay, unit)：延遲一次性發射

Observable.timer(3, TimeUnit.SECONDS)
    .flatMap(l -> Observable.just(list.get(l.intValue())))
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(...);
// 3 秒後發射 "aa" 並 completed

sample(period, unit)：定期採樣最新值

兩種情況：

• 源發射間隔 > sample 間隔 → 每次採樣皆為新值
• 源發射間隔 < sample 間隔 → 取最近一次值

// 源 3 秒一個，sample 2 秒採樣
Observable.interval(3, TimeUnit.SECONDS)
    .flatMap(i -> Observable.just(list.get(i.intValue())))
    .sample(2, TimeUnit.SECONDS)
    .subscribe(...);
// 發射間隔: 4s, 2s, 4s... (見圖解)

timeout(timeout, unit)：超時檢測（間隔超時則報錯終止）

debounce(timeout, unit)：防抖（頻繁發射僅保留靜默期後最後一值）

選型速查表