Android RxJava操作符的學習---過濾操作符
阿新 • • 發佈：2018-11-09
3.5 過濾操作符
-
作用
過濾 / 篩選 被觀察者（Observable）傳送的事件 & 觀察者 （Observer）接收的事件 -
應用場景
- 根據 指定條件 過濾事件
- 根據 指定事件數量 過濾事件
- 根據 指定時間 過濾事件
- 根據 指定事件位置 過濾事件
-
型別
根據上述應用場景，常見的過濾操作符 主要有：
3.5.1. 應用場景 & 對應操作符詳解
- 過濾操作符的應用場景包括：
- 根據 指定條件 過濾事件
- 根據 指定事件數量 過濾事件
- 根據 指定時間 過濾事件
- 根據 指定事件位置 過濾事件
- 下面，我將根據上述應用場景，講解對應的操作符使用
注：在使用RxJava 2操作符前，記得在專案的Gradle中新增依賴：
dependencies {
compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
compile 'io.reactivex.rxjava2:rxjava:2.0.7'
// 注：RxJava2 與 RxJava1 不能共存，即依賴不能同時存在
}
3.5.1.1 根據 指定條件 過濾事件
-
需求場景
通過設定指定的過濾條件，當且僅當該事件滿足條件，就將該事件過濾（不傳送） -
對應操作符型別
- 對應操作符使用
Filter（）
-
作用
過濾 特定條件的事件 -
原理
- 具體使用
Observable.create(new ObservableOnSubscribe<Integer>() { @Override public void subscribe(ObservableEmitter<Integer> emitter) throws Exception { // 1. 傳送5個事件 emitter.onNext(1); emitter.onNext(2); emitter.onNext(3); emitter.onNext(4); emitter.onNext(5); } // 2. 採用filter（）變換操作符 }).filter(new Predicate<Integer>() { // 根據test()的返回值 對被觀察者傳送的事件進行過濾 & 篩選 // a. 返回true，則繼續傳送 // b. 返回false，則不傳送（即過濾） @Override public boolean test(Integer integer) throws Exception { return integer > 3; // 本例子 = 過濾了整數≤3的事件 } }).subscribe(new Observer<Integer>() { @Override public void onSubscribe(Disposable d) { Log.d(TAG, "開始採用subscribe連線"); } @Override public void onNext(Integer value) { Log.d(TAG, "過濾後得到的事件是："+ value ); } @Override public void onError(Throwable e) { Log.d(TAG, "對Error事件作出響應"); } @Override public void onComplete() { Log.d(TAG, "對Complete事件作出響應"); } });
- 測試結果
ofType（）
-
作用
過濾 特定資料型別的資料 -
具體使用
Observable.just(1, "Carson", 3, "Ho", 5)
.ofType(Integer.class) // 篩選出 整型資料
.subscribe(new Consumer<Integer>() {
@Override
public void accept( Integer integer) throws Exception {
Log.d(TAG,"獲取到的整型事件元素是： "+ integer);
}
});
- 測試結果
skip（） / skipLast（）
-
作用
跳過某個事件 -
具體使用
// 使用1：根據順序跳過資料項
Observable.just(1, 2, 3, 4, 5)
.skip(1) // 跳過正序的前1項
.skipLast(2) // 跳過正序的後2項
.subscribe(new Consumer<Integer>() {
@Override
public void accept( Integer integer) throws Exception {
Log.d(TAG,"獲取到的整型事件元素是： "+ integer);
}
});
// 使用2：根據時間跳過資料項
// 傳送事件特點：傳送資料0-5，每隔1s傳送一次，每次遞增1；第1次傳送延遲0s
Observable.intervalRange(0, 5, 0, 1, TimeUnit.SECONDS)
.skip(1, TimeUnit.SECONDS) // 跳過第1s傳送的資料
.skipLast(1, TimeUnit.SECONDS) // 跳過最後1s傳送的資料
.subscribe(new Consumer<Long>() {
@Override
public void accept( Long along ) throws Exception {
Log.d(TAG,"獲取到的整型事件元素是： "+ along);
}
});
- 測試結果
distinct（） / distinctUntilChanged（）
-
作用
過濾事件序列中重複的事件 / 連續重複的事件 -
具體使用
// 使用1：過濾事件序列中重複的事件
Observable.just(1, 2, 3, 1 , 2 )
.distinct()
.subscribe(new Consumer<Integer>() {
@Override
public void accept( Integer integer) throws Exception {
Log.d(TAG,"不重複的整型事件元素是： "+ integer);
}
});
// 使用2：過濾事件序列中 連續重複的事件
// 下面序列中，連續重複的事件 = 3、4
Observable.just(1,2,3,1,2,3,3,4,4 )
.distinctUntilChanged()
.subscribe(new Consumer<Integer>() {
@Override
public void accept( Integer integer) throws Exception {
Log.d(TAG,"不連續重複的整型事件元素是： "+ integer);
}
});
- 測試結果
3.5.1.2 根據 指定事件數量 過濾事件
-
需求場景
通過設定指定的事件數量，僅傳送特定數量的事件 -
對應操作符型別
take（）&takeLast（） -
對應操作符使用
take（）
-
作用
指定觀察者最多能接收到的事件數量 -
原理
- 具體使用
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
// 1. 傳送5個事件
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
emitter.onNext(4);
emitter.onNext(5);
}
// 採用take（）變換操作符
// 指定了觀察者只能接收2個事件
}).take(2)
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "開始採用subscribe連線");
}
@Override
public void onNext(Integer value) {
Log.d(TAG, "過濾後得到的事件是："+ value );
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "對Error事件作出響應");
}
@Override
public void onComplete() {
Log.d(TAG, "對Complete事件作出響應");
}
});
// 實際上，可理解為：被觀察者還是傳送了5個事件，只是因為操作符的存在攔截了3個事件，最終觀察者接收到的是2個事件
- 測試結果
takeLast（）
-
作用
指定觀察者只能接收到被觀察者傳送的最後幾個事件 -
具體使用
Observable.just(1, 2, 3, 4, 5)
.takeLast(3) //指定觀察者只能接受被觀察者傳送的3個事件
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "開始採用subscribe連線");
}
@Override
public void onNext(Integer value) {
Log.d(TAG, "過濾後得到的事件是："+ value );
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "對Error事件作出響應");
}
@Override
public void onComplete() {
Log.d(TAG, "對Complete事件作出響應");
}
});
- 測試結果
3.5.1.3 根據 指定時間 過濾事件
-
需求場景
通過設定指定的時間，僅傳送在該時間內的事件 -
對應操作符型別
- 對應操作符使用
throttleFirst（）/ throttleLast（）
- 作用
在某段時間內，只發送該段時間內第1次事件 / 最後1次事件
如，1段時間內連續點選按鈕，但只執行第1次的點選操作
- 原理示意圖
- 具體使用
<<- 在某段時間內，只發送該段時間內第1次事件 ->>
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
// 隔段事件傳送時間
e.onNext(1);
Thread.sleep(500);
e.onNext(2);
Thread.sleep(400);
e.onNext(3);
Thread.sleep(300);
e.onNext(4);
Thread.sleep(300);
e.onNext(5);
Thread.sleep(300);
e.onNext(6);
Thread.sleep(400);
e.onNext(7);
Thread.sleep(300);
e.onNext(8);
Thread.sleep(300);
e.onNext(9);
Thread.sleep(300);
e.onComplete();
}
}).throttleFirst(1, TimeUnit.SECONDS)//每1秒中採用資料
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "開始採用subscribe連線");
}
@Override
public void onNext(Integer value) {
Log.d(TAG, "接收到了事件"+ value );
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "對Error事件作出響應");
}
@Override
public void onComplete() {
Log.d(TAG, "對Complete事件作出響應");
}
});
<<- 在某段時間內，只發送該段時間內最後1次事件 ->>
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
// 隔段事件傳送時間
e.onNext(1);
Thread.sleep(500);
e.onNext(2);
Thread.sleep(400);
e.onNext(3);
Thread.sleep(300);
e.onNext(4);
Thread.sleep(300);
e.onNext(5);
Thread.sleep(300);
e.onNext(6);
Thread.sleep(400);
e.onNext(7);
Thread.sleep(300);
e.onNext(8);
Thread.sleep(300);
e.onNext(9);
Thread.sleep(300);
e.onComplete();
}
}).throttleLast(1, TimeUnit.SECONDS)//每1秒中採用資料
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "開始採用subscribe連線");
}
@Override
public void onNext(Integer value) {
Log.d(TAG, "接收到了事件"+ value );
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "對Error事件作出響應");
}
@Override
public void onComplete() {
Log.d(TAG, "對Complete事件作出響應");
}
});
- 測試結果
Sample（）
- 作用
在某段時間內，只發送該段時間內最新（最後）1次事件
與
throttleLast（）操作符類似
- 具體使用
僅需要把上文的 throttleLast（） 改成Sample（）操作符即可，此處不作過多描述
throttleWithTimeout （） / debounce（）
-
作用
傳送資料事件時，若2次傳送事件的間隔＜指定時間，就會丟棄前一次的資料，直到指定時間內都沒有新資料發射時才會傳送後一次的資料 -
具體使用
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
// 隔段事件傳送時間
e.onNext(1);
Thread.sleep(500);
e.onNext(2); // 1和2之間的間隔小於指定時間1s，所以前1次資料（1）會被拋棄，2會被保留
Thread.sleep(1500); // 因為2和3之間的間隔大於指定時間1s，所以之前被保留的2事件將發出
e.onNext(3);
Thread.sleep(1500); // 因為3和4之間的間隔大於指定時間1s，所以3事件將發出
e.onNext(4);
Thread.sleep(500); // 因為4和5之間的間隔小於指定時間1s，所以前1次資料（4）會被拋棄，5會被保留
e.onNext(5);
Thread.sleep(500); // 因為5和6之間的間隔小於指定時間1s，所以前1次資料（5）會被拋棄，6會被保留
e.onNext(6);
Thread.sleep(1500); // 因為6和Complete實踐之間的間隔大於指定時間1s，所以之前被保留的6事件將發出
e.onComplete();
}
}).throttleWithTimeout(1, TimeUnit.SECONDS)//每1秒中採用資料
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Integer value) {
Log.d(TAG, "接收到了事件"+ value );
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "對Error事件作出響應");
}
@Override
public void onComplete() {
Log.d(TAG, "對Complete事件作出響應");
}
});
- 測試結果
3.5.1.4 根據 指定事件位置 過濾事件
-
需求場景
通過設定指定的位置，過濾在該位置的事件 -
對應操作符型別
- 對應操作符使用
firstElement（） / lastElement（）
-
作用
僅選取第1個元素 / 最後一個元素 -
具體使用
// 獲取第1個元素
Observable.just(1, 2, 3, 4, 5)
.firstElement()
.subscribe(new Consumer<Integer>() {
@Override
public void accept( Integer integer) throws Exception {
Log.d(TAG,"獲取到的第一個事件是： "+ integer);
}
});
// 獲取最後1個元素
Observable.just(1, 2, 3, 4, 5)
.lastElement()
.subscribe(new Consumer<Integer>() {
@Override
public void accept( Integer integer) throws Exception {
Log.d(TAG,"獲取到的最後1個事件是： "+ integer);
}
});
- 測試結果
elementAt（）
- 作用
指定接收某個元素（通過 索引值 確定）
注：允許越界，即獲取的位置索引 ＞ 傳送事件序列長度
- 具體使用
// 使用1：獲取位置索引 = 2的 元素
// 位置索引從0開始
Observable.just(1, 2, 3, 4, 5)
.elementAt(2)
.subscribe(new Consumer<Integer>() {
@Override
public void accept( Integer integer) throws Exception {
Log.d(TAG,"獲取到的事件元素是： "+ integer);
}
});
// 使用2：獲取的位置索引 ＞ 傳送事件序列長度時，設定預設引數
Observable.just(1, 2, 3, 4, 5)
.elementAt(6,10)
.subscribe(new Consumer<Integer>() {
@Override
public void accept( Integer integer) throws Exception {
Log.d(TAG,"獲取到的事件元素是： "+ integer);
}
});
- 測試結果
elementAtOrError（）
-
作用
在elementAt（）的基礎上，當出現越界情況（即獲取的位置索引 ＞ 傳送事件序列長度）時，即丟擲異常 -
具體使用
Observable.just(1, 2, 3, 4, 5)
.elementAtOrError(6)
.subscribe(new Consumer<Integer>() {
@Override
public void accept( Integer integer) throws Exception {
Log.d(TAG,"獲取到的事件元素是： "+ integer);
}
});
- 測試結果
總結
