RxJava 是一个基于观察者模式的响应式编程库，在 Android 开发中被广泛使用。其中，过滤和条件操作是 RxJava 中非常重要的一部分，它们允许我们对数据流进行精细控制。本文将详细介绍 RxJava 中常用的过滤与条件操作符及其使用场景。

一、过滤操作符

1. filter()

filter() 是最基本的过滤操作符，它根据给定的条件过滤发射的数据项。

java

Observable.just(1, 2, 3, 4, 5).filter(number -> number % 2 == 0).subscribe(evenNumber -> Log.d("TAG", "Even number: " + evenNumber));
// 输出: Even number: 2, Even number: 4

2. take()

take() 只发射前 N 项数据，然后取消订阅。

java

Observable.just(1, 2, 3, 4, 5).take(3).subscribe(number -> Log.d("TAG", "Number: " + number));
// 输出: Number: 1, Number: 2, Number: 3

3. takeLast()

takeLast() 只发射最后 N 项数据。

java

Observable.just(1, 2, 3, 4, 5).takeLast(2).subscribe(number -> Log.d("TAG", "Number: " + number));
// 输出: Number: 4, Number: 5

4. skip()

skip() 跳过前 N 项数据，发射剩余的数据。

java

Observable.just(1, 2, 3, 4, 5).skip(2).subscribe(number -> Log.d("TAG", "Number: " + number));
// 输出: Number: 3, Number: 4, Number: 5

5. skipLast()

skipLast() 跳过后 N 项数据，发射前面的数据。

java

Observable.just(1, 2, 3, 4, 5).skipLast(2).subscribe(number -> Log.d("TAG", "Number: " + number));
// 输出: Number: 1, Number: 2, Number: 3

6. distinct()

distinct() 过滤掉重复的数据项。

java

Observable.just(1, 2, 2, 3, 4, 4, 5).distinct().subscribe(number -> Log.d("TAG", "Number: " + number));
// 输出: Number: 1, Number: 2, Number: 3, Number: 4, Number: 5

7. distinctUntilChanged()

distinctUntilChanged() 过滤掉连续重复的数据项。

java

Observable.just(1, 2, 2, 3, 2, 4, 4, 5).distinctUntilChanged().subscribe(number -> Log.d("TAG", "Number: " + number));
// 输出: Number: 1, Number: 2, Number: 3, Number: 2, Number: 4, Number: 5

8. elementAt()

elementAt() 只发射指定索引位置的数据项。

java

Observable.just(1, 2, 3, 4, 5).elementAt(2).subscribe(number -> Log.d("TAG", "Number: " + number));
// 输出: Number: 3

9. ignoreElements()

ignoreElements() 忽略所有数据项，只发射完成或错误通知。

java

Observable.just(1, 2, 3, 4, 5).ignoreElements().subscribe(() -> Log.d("TAG", "Completed"),throwable -> Log.e("TAG", "Error", throwable));
// 输出: Completed

10. debounce()

debounce() 仅在特定时间段内没有发射其他数据时才发射数据，常用于搜索输入框。

java

RxTextView.textChanges(searchEditText).debounce(300, TimeUnit.MILLISECONDS).subscribe(charSequence -> performSearch(charSequence.toString()));

11. throttleFirst()

throttleFirst() 在指定时间窗口内只发射第一个数据项，常用于防止按钮重复点击。

java

RxView.clicks(button).throttleFirst(1, TimeUnit.SECONDS).subscribe(aVoid -> performAction());

12. throttleLast() / sample()

throttleLast() 或 sample() 在指定时间窗口内只发射最后一个数据项。

java

Observable.interval(100, TimeUnit.MILLISECONDS).throttleLast(1, TimeUnit.SECONDS).subscribe(number -> Log.d("TAG", "Number: " + number));
// 每秒发射一个数字

二、条件操作符

1. all()

all() 判断是否所有数据项都满足某个条件。

java

Observable.just(2, 4, 6, 8).all(number -> number % 2 == 0).subscribe(isAllEven -> Log.d("TAG", "All even? " + isAllEven));
// 输出: All even? true

2. contains()

contains() 判断数据流是否包含某个特定项。

java

Observable.just(1, 2, 3, 4, 5).contains(3).subscribe(contains -> Log.d("TAG", "Contains 3? " + contains));
// 输出: Contains 3? true

3. isEmpty()

isEmpty() 判断数据流是否为空。

java

Observable.empty().isEmpty().subscribe(isEmpty -> Log.d("TAG", "Is empty? " + isEmpty));
// 输出: Is empty? true

4. defaultIfEmpty()

defaultIfEmpty() 如果数据流为空，则发射一个默认值。

java

Observable.empty().defaultIfEmpty("Default Value").subscribe(value -> Log.d("TAG", "Value: " + value));
// 输出: Value: Default Value

5. sequenceEqual()

sequenceEqual() 比较两个 Observable 发射的数据序列是否相同。

java

Observable.sequenceEqual(Observable.just(1, 2, 3),Observable.just(1, 2, 3)
).subscribe(isEqual -> Log.d("TAG", "Sequences equal? " + isEqual));
// 输出: Sequences equal? true

6. amb()

amb() 从多个 Observable 中选择第一个发射数据的 Observable。

java

Observable.amb(Observable.just(1, 2, 3).delay(1, TimeUnit.SECONDS),Observable.just(4, 5, 6)
).subscribe(number -> Log.d("TAG", "Number: " + number));
// 输出: Number: 4, Number: 5, Number: 6

三、实际应用场景

1. 搜索功能优化

java

RxTextView.textChanges(searchEditText).debounce(300, TimeUnit.MILLISECONDS)  // 防抖.filter(text -> text.length() > 2)     // 至少3个字符才搜索.distinctUntilChanged()               // 内容变化才搜索.switchMap(query -> searchApi(query))  // 取消之前的搜索.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(results -> updateUI(results));

2. 防止按钮重复点击

java

RxView.clicks(button).throttleFirst(1, TimeUnit.SECONDS)  // 1秒内只响应第一次点击.subscribe(aVoid -> performAction());

3. 数据校验

java

Observable.just(userInput).filter(input -> !TextUtils.isEmpty(input)).map(input -> Integer.parseInt(input)).filter(number -> number > 0 && number < 100).subscribe(validNumber -> processInput(validNumber));