RxJava操作符系列三（上）
RxJava操作符系列傳送門
- RxJava操作符源碼
https://github.com/xiehui999/fuseProgram
前言
在之前的文章，我們介紹了一些Observable的創(chuàng)建以及數(shù)據(jù)轉(zhuǎn)換的操作符，其中的一些數(shù)據(jù)轉(zhuǎn)換的操作符理解還是有一定的難度的，但是相信如果敲一遍代碼并且修改各種參數(shù)的值，去觀察執(zhí)行的日志，相信還是很容易的理解的。在官網(wǎng)，每個操作符都給出了圖例，如果你對文字的理解不夠清楚明白，也可以去參考圖示幫助自己理解。在這篇文章中，我們將介紹一些常見的過濾操作符，在RxJava中過濾操作符也是比較好理解的，好了，讓我們一起繼續(xù)開啟學習之旅吧。
Filter
該操作符接收一個Func1參數(shù)，我們可以在其中通過運用你自己的判斷條件去判斷我們要過濾的數(shù)據(jù)，當數(shù)據(jù)通過判斷條件后返回true表示發(fā)射該項數(shù)據(jù)，否則就不發(fā)射，這樣就過濾出了我們想要的數(shù)據(jù)。如下，我們過濾出不能被2整除的數(shù)
- Integer[] ints = {1, 2, 3, 4, 5, 6, 7, 8, 9};
- Observable observable = Observable.from(ints).filter(new Func1<Integer, Boolean>() {
- @Override
- public Boolean call(Integer integer) {
- return integer % 2 != 0;//返回true，就不會過濾掉，會發(fā)射數(shù)據(jù)，過濾掉返回false的值
- }
- });
- Action1 action1 = new Action1<Integer>() {
- @Override
- public void call(Integer i) {
- Log.e(TAG, "call: "+i );
- }
- };
- observable.subscribe(action1);
輸出日志信息
- call: 1
- call: 3
- call: 5
- call: 7
- call: 9
ofType
該操作符是filter操作符的一個特殊形式。它過濾一個Observable只返回指定類型的數(shù)據(jù)，例如當數(shù)據(jù)源有字符串和int型數(shù)據(jù)時，我們想要過濾出字符串就可以使用這個操作符，如下示例代碼
- Observable.just(0, "one", 6, 4, "two", 8, "three", 1, "four", 0)
- .ofType(String.class)
- .subscribe(new Subscriber<String>() {
- @Override
- public void onCompleted() {
- Log.e(TAG, "onCompleted:ofType ");
- }
- @Override
- public void onError(Throwable e) {
- Log.e(TAG, "onError:ofType ");
- }
- @Override
- public void onNext(String string) {
- Log.e(TAG, "onNext:ofType " + string);
- }
- });
輸出日志信息
- onNext:ofType one
- onNext:ofType two
- onNext:ofType three
- onNext:ofType four
- onCompleted:ofType
當然除了過濾基本類型的數(shù)據(jù)，也可以過濾自定義類型數(shù)據(jù)。
First
如果我們只對Observable發(fā)射的***項數(shù)據(jù)，或者滿足某個條件的***項數(shù)據(jù)感興趣，則可以使用First操作符。
- Observable.just(10, 11, 12, 13).first().subscribe(new Action1() {
- @Override
- public void call(Integer integer) {
- Log.e(TAG, integer+"");
- }
- });
上面日志只打印一個值10，當然我們也可以給first傳一個參數(shù)Fun1，指定一個條件如下
- Observable.just(10, 11, 12, 13).first(new Func1<Integer, Boolean>() {
- @Override
- public Boolean call(Integer integer) {
- return integer > 12;
- }
- }).subscribe(new Action1<Integer>() {
- @Override
- public void call(Integer integer) {
- Log.e(TAG, integer+"");
- }
- });
此時輸出的信息就是滿足integer > 12的***項數(shù)據(jù)13。
firstOrDefault
該操作符是first操作符的變形。主要是在沒有發(fā)射任何數(shù)據(jù)時發(fā)射一個你在參數(shù)中指定的默認值。如下，它有有兩個重載方法。
- Observable.just(11,12,13).firstOrDefault(10).subscribe(new Action1<Object>() {
- @Override
- public void call(Object o) {
- Log.e(TAG, o.toString());
- }
- });
如果寫成上面的代碼，這個執(zhí)行會和first效果一樣。因為沒有發(fā)射數(shù)據(jù)的時候才用到默認值，那么我們將上面代碼更改如下，使用empty創(chuàng)建一個不發(fā)射任何數(shù)據(jù)但是正常終止的Observable。
- Observable.empty().firstOrDefault(10).subscribe(new Action1<Object>() {
- @Override
- public void call(Object o) {
- Log.e(TAG, o.toString());
- }
- });
發(fā)現(xiàn)此時輸出了數(shù)據(jù)10.該操作符還提供了兩個參數(shù)的重載方法firstOrDefault(T defaultValue, Func1 super T, Boolean> predicate)。我們可以增加一個條件。如下示例
- Observable.just(10，13,16).firstOrDefault(15, new Func1<Integer, Boolean>() {
- @Override
- public Boolean call(Integer integer) {
- return integer>20;
- }
- }).subscribe(new Action1<Integer>() {
- @Override
- public void call(Integer integer) {
- Log.e(TAG, ""+integer);
- }
- });
此時數(shù)據(jù)源10,13,16都不滿足大于20,則此時將輸出默認值15，如果我們將數(shù)據(jù)源數(shù)據(jù)增加一個值22.那么此時將不再輸出默認值，而是輸出22。
takeFirst
該操作符與first操作符的區(qū)別就是如果原始Observable沒有發(fā)射任何滿足條件的數(shù)據(jù)，first會拋出一個NoSuchElementException直接執(zhí)行onError()，而takeFist會返回一個空的Observable(不調(diào)用onNext()但是會調(diào)用onCompleted)
如下面下面示例代碼
- Observable.just(10,11).filter(new Func1<Integer, Boolean>() {
- @Override
- public Boolean call(Integer integer) {
- return integer>20;
- }
- }).first().subscribe(new Subscriber<Object>() {
- @Override
- public void onCompleted() {
- Log.e(TAG, "onCompleted: ");
- }
- @Override
- public void onError(Throwable e) {
- Log.e(TAG, "onError: "+e.toString());
- }
- @Override
- public void onNext(Object o) {
- Log.e(TAG, "onNext: "+o.toString());
- }
- });
執(zhí)行后輸出的信息如下
- onError: java.util.NoSuchElementException: Sequence contains no elements
若此時用takeFirst
- Observable.just(10,11).takeFirst(new Func1<Integer, Boolean>() {
- @Override
- public Boolean call(Integer integer) {
- Log.e(TAG, "call: takeFirst" );
- return integer>30;
- }
- }).subscribe(new Subscriber<Object>() {
- @Override
- public void onCompleted() {
- Log.e(TAG, "onCompleted: ");
- }
- @Override
- public void onError(Throwable e) {
- Log.e(TAG, "onError: "+e.toString());
- }
- @Override
- public void onNext(Object o) {
- Log.e(TAG, "onNext: "+o.toString());
- }
- });
發(fā)現(xiàn)此時不會出現(xiàn)異常，而是執(zhí)行了onCompleted()。
single
如果原始Observable在完成之前不是正好發(fā)射一次數(shù)據(jù)，它會拋出一個NoSuchElementException，白話可以理解為發(fā)送數(shù)據(jù)是一項的話輸出此項的值，若是多個數(shù)據(jù)則拋出異常執(zhí)行onError()方法。
如下代碼
- Observable.just(10, 11, 12, 13).single().subscribe(new Subscriber<Integer>() {
- @Override
- public void onCompleted() {
- Log.e(TAG, "onCompleted");
- }
- @Override
- public void onError(Throwable e) {
- Log.e(TAG, "onError"+e.toString());
- }
- @Override
- public void onNext(Integer integer) {
- Log.e(TAG, integer);
- }
- });
輸出信息
- onError: java.util.NoSuchElementException: Sequence contains no elements
如果將上述代碼做下簡單更改
- Observable.just(10, 11, 12, 13).filter(new Func1<Integer, Boolean>() {
- @Override
- public Boolean call(Integer integer) {
- return integer > 12;
- }
- }).subscribe(new Subscriber<Integer>() {
- @Override
- public void onCompleted() {
- Log.e(TAG, "onCompleted");
- }
- @Override
- public void onError(Throwable e) {
- Log.e(TAG, "onError"+e.toString());
- }
- @Override
- public void onNext(Integer integer) {
- Log.e(TAG, integer);
- }
- });
此時會輸出數(shù)據(jù)13,因為此時通過filter后就只有一條數(shù)據(jù)。single也有singleOrDefault(T)和singleOrDefault(T,Func1)兩個變體，具體可以自己代碼測試區(qū)別。
Last
該操作符與first意義相反，若我們只對Observable發(fā)射的***一項數(shù)據(jù)，或者滿足某個條件的***一項數(shù)據(jù)感興趣時使用該操作符。
示例代碼
- Observable.just(10, 11, 12, 13).last().subscribe(new Action1<Integer>() {
- @Override
- public void call(Integer integer) {
- Log.e(TAG, "call: "+integer);
- }
- });
執(zhí)行后輸出13.它有一個重載方法可以指定條件，獲取滿足條件的***一項數(shù)據(jù)的。將上面代碼修改如下
- Observable.just(10, 11, 12, 13).last(new Func1<Integer, Boolean>() {
- @Override
- public Boolean call(Integer integer) {
- return integer < 12;
- }
- }).subscribe(new Action1<Integer>() {
- @Override
- public void call(Integer integer) {
- Log.e(TAG, "call: "+integer);
- }
- });
此時最終輸出數(shù)據(jù)就是11.該操作符和first一樣也有幾種變體，如lastOrDefault，TakeLast，具體效果可自己測試。
Skip
該操作符是跳過之前的前幾項數(shù)據(jù)，然后再發(fā)射數(shù)據(jù)。
- Observable.range(1, 10).skip(6).subscribe(new Action1<Integer>() {
- @Override
- public void call(Integer integer) {
- Log.e(TAG, "call: "+integer );
- }
- });
輸出日志信息
- call: 7
- call: 8
- call: 9
- call: 10
skip還有兩個重載方法.skip(long time, TimeUnit unit)默認是在computation調(diào)度器上執(zhí)行，如果要有更新UI操作需要通過observeOn方法指定為AndroidSchedulers.mainThread()，當然還有一個重載方法skip(long time, TimeUnit unit, Scheduler scheduler)可以指定調(diào)度器。注意的一點是這兩個重載方法的***個參數(shù)不是跳過的數(shù)據(jù)數(shù)量，指的是時間。
- Observable.interval(500, TimeUnit.MILLISECONDS)
- .skip(2, TimeUnit.SECONDS)
- .observeOn(AndroidSchedulers.mainThread())
- .subscribe(new Subscriber<Long>() {
- @Override
- public void onCompleted() {
- }
- @Override
- public void onError(Throwable e) {
- }
- @Override
- public void onNext(Long aLong) {
- tv.append("\n" + aLong);
- if (aLong > 10) {
- this.unsubscribe();
- }
- }
- });
如上代碼，通過interval每隔500毫秒產(chǎn)生一個數(shù)據(jù)，通過skip設(shè)定跳過時間為2秒。并且當數(shù)據(jù)大于10時解除訂閱。
skipLast
正好和skip 相反，忽略***產(chǎn)生的n個數(shù)據(jù)項
- Observable.range(1, 10).skipLast(6).subscribe(new Action1<Integer>() {
- @Override
- public void call(Integer integer) {
- Log.e(TAG, "call: "+integer );
- }
- });
輸出日志信息
- call: 1
- call: 2
- call: 3
- call: 4
接下文
