RxJava操作符系列傳送門
RxJava操作符源碼
RxJava操作符系列一
RxJava操作符系列二
前言
在之前的文章茎芭，我們介紹了一些Observable的創(chuàng)建以及數(shù)據(jù)轉(zhuǎn)換的操作符刹枉，其中的一些數(shù)據(jù)轉(zhuǎn)換的操作符理解還是有一定的難度的湃望，但是相信如果敲一遍代碼并且修改各種參數(shù)的值榔昔，去觀察執(zhí)行的日志槽袄，相信還是很容易的理解的掌呜。在官網(wǎng)，每個(gè)操作符都給出了圖例名党，如果你對(duì)文字的理解不夠清楚明白叹阔，也可以去參考圖示幫助自己理解。在這篇文章中传睹，我們將介紹一些常見(jiàn)的過(guò)濾操作符耳幢，在RxJava中過(guò)濾操作符也是比較好理解的，好了蒋歌，讓我們一起繼續(xù)開(kāi)啟學(xué)習(xí)之旅吧帅掘。
Filter
該操作符接收一個(gè)Func1參數(shù)，我們可以在其中通過(guò)運(yùn)用你自己的判斷條件去判斷我們要過(guò)濾的數(shù)據(jù)堂油，當(dāng)數(shù)據(jù)通過(guò)判斷條件后返回true表示發(fā)射該項(xiàng)數(shù)據(jù)修档，否則就不發(fā)射，這樣就過(guò)濾出了我們想要的數(shù)據(jù)府框。如下吱窝，我們過(guò)濾出不能被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，就不會(huì)過(guò)濾掉迫靖，會(huì)發(fā)射數(shù)據(jù)院峡，過(guò)濾掉返回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操作符的一個(gè)特殊形式。它過(guò)濾一個(gè)Observable只返回指定類型的數(shù)據(jù)系宜，例如當(dāng)數(shù)據(jù)源有字符串和int型數(shù)據(jù)時(shí)照激，我們想要過(guò)濾出字符串就可以使用這個(gè)操作符，如下示例代碼
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
當(dāng)然除了過(guò)濾基本類型的數(shù)據(jù)盹牧，也可以過(guò)濾自定義類型數(shù)據(jù)俩垃。
First
如果我們只對(duì)Observable發(fā)射的第一項(xiàng)數(shù)據(jù)，或者滿足某個(gè)條件的第一項(xiàng)數(shù)據(jù)感興趣汰寓，則可以使用First操作符口柳。
Observable.just(10, 11, 12, 13).first().subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.e(TAG, integer+"");
}
});
上面日志只打印一個(gè)值10，當(dāng)然我們也可以給first傳一個(gè)參數(shù)Fun1有滑，指定一個(gè)條件如下
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+"");
}
});
此時(shí)輸出的信息就是滿足integer > 12的第一項(xiàng)數(shù)據(jù)13跃闹。
firstOrDefault
該操作符是first操作符的變形。主要是在沒(méi)有發(fā)射任何數(shù)據(jù)時(shí)發(fā)射一個(gè)你在參數(shù)中指定的默認(rèn)值毛好。如下望艺，它有有兩個(gè)重載方法。
Observable.just(11,12,13).firstOrDefault(10).subscribe(new Action1<Object>() {
@Override
public void call(Object o) {
Log.e(TAG, o.toString());
}
});
如果寫(xiě)成上面的代碼肌访，這個(gè)執(zhí)行會(huì)和first效果一樣荣茫。因?yàn)闆](méi)有發(fā)射數(shù)據(jù)的時(shí)候才用到默認(rèn)值，那么我們將上面代碼更改如下场靴，使用empty創(chuàng)建一個(gè)不發(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í)輸出了數(shù)據(jù)10.該操作符還提供了兩個(gè)參數(shù)的重載方法firstOrDefault(T defaultValue, Func1<? super T, Boolean> predicate)。我們可以增加一個(gè)條件旨剥。如下示例
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í)數(shù)據(jù)源10,13,16都不滿足大于20,則此時(shí)將輸出默認(rèn)值15，如果我們將數(shù)據(jù)源數(shù)據(jù)增加一個(gè)值22.那么此時(shí)將不再輸出默認(rèn)值轨帜，而是輸出22魄咕。
takeFirst
該操作符與first操作符的區(qū)別就是如果原始Observable沒(méi)有發(fā)射任何滿足條件的數(shù)據(jù)，first會(huì)拋出一個(gè)NoSuchElementException直接執(zhí)行onError()蚌父，而takeFist會(huì)返回一個(gè)空的Observable（不調(diào)用onNext()但是會(huì)調(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
若此時(shí)用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)此時(shí)不會(huì)出現(xiàn)異常哮兰，而是執(zhí)行了onCompleted（）。
single
如果原始Observable在完成之前不是正好發(fā)射一次數(shù)據(jù)苟弛，它會(huì)拋出一個(gè)NoSuchElementException喝滞，白話可以理解為發(fā)送數(shù)據(jù)是一項(xiàng)的話輸出此項(xiàng)的值，若是多個(gè)數(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
如果將上述代碼做下簡(jiǎn)單更改
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í)會(huì)輸出數(shù)據(jù)13,因?yàn)榇藭r(shí)通過(guò)filter后就只有一條數(shù)據(jù)右遭。single也有singleOrDefault(T)和singleOrDefault(T,Func1)兩個(gè)變體，具體可以自己代碼測(cè)試區(qū)別缤削。
Last
該操作符與first意義相反窘哈，若我們只對(duì)Observable發(fā)射的最后一項(xiàng)數(shù)據(jù)，或者滿足某個(gè)條件的最后一項(xiàng)數(shù)據(jù)感興趣時(shí)使用該操作符亭敢。
示例代碼
Observable.just(10, 11, 12, 13).last().subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
Log.e(TAG, "call: "+integer);
}
});
執(zhí)行后輸出13.它有一個(gè)重載方法可以指定條件滚婉，獲取滿足條件的最后一項(xiàng)數(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í)最終輸出數(shù)據(jù)就是11.該操作符和first一樣也有幾種變體帅刀，如lastOrDefault让腹，TakeLast，具體效果可自己測(cè)試劝篷。
Skip
該操作符是跳過(guò)之前的前幾項(xiàng)數(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還有兩個(gè)重載方法.skip(long time, TimeUnit unit)默認(rèn)是在computation調(diào)度器上執(zhí)行娇妓，如果要有更新UI操作需要通過(guò)observeOn方法指定為AndroidSchedulers.mainThread()像鸡，當(dāng)然還有一個(gè)重載方法skip(long time, TimeUnit unit, Scheduler scheduler)可以指定調(diào)度器。注意的一點(diǎn)是這兩個(gè)重載方法的第一個(gè)參數(shù)不是跳過(guò)的數(shù)據(jù)數(shù)量哈恰，指的是時(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();
}
}
});
如上代碼，通過(guò)interval每隔500毫秒產(chǎn)生一個(gè)數(shù)據(jù)着绷，通過(guò)skip設(shè)定跳過(guò)時(shí)間為2秒蛔钙。并且當(dāng)數(shù)據(jù)大于10時(shí)解除訂閱。
skipLast
正好和skip 相反荠医，忽略最后產(chǎn)生的n個(gè)數(shù)據(jù)項(xiàng)
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
Take
Take操作符可以修改Observable的行為吁脱，只返回前面的N項(xiàng)數(shù)據(jù)桑涎，然后發(fā)射完成通知，忽略剩余的數(shù)據(jù)兼贡。
Observable.range(1,8)
.take(4)
.subscribe(new Subscriber<Integer>() {
@Override
public void onNext(Integer item) {
Log.e(TAG, "Next: " + item);
}
@Override
public void onError(Throwable error) {
Log.e(TAG, "Error: " + error.getMessage());
}
@Override
public void onCompleted() {
Log.e(TAG, "complete.");
}
});
輸出日志信息
Next: 1
Next: 2
Next: 3
Next: 4
complete
take和skip一樣也有其它兩個(gè)重載方法take(long time, TimeUnit unit)攻冷，take(long time, TimeUnit unit, Scheduler scheduler)，默認(rèn)在computation調(diào)度器上執(zhí)行遍希。
take還有變體操作符TakeLast等曼，takeLastBuffer具體執(zhí)行效果可自行代碼。
Debounce
該操作符指的是過(guò)了一段指定的時(shí)間還沒(méi)發(fā)射數(shù)據(jù)時(shí)才發(fā)射一個(gè)數(shù)據(jù)凿蒜，聽(tīng)著可能有點(diǎn)繞禁谦。你可以理解對(duì)源Observable間隔期產(chǎn)生的結(jié)果進(jìn)行過(guò)濾，如果在這個(gè)規(guī)定的間隔期內(nèi)沒(méi)有別的結(jié)果產(chǎn)生废封，則將這個(gè)結(jié)果提交給訂閱者州泊，否則忽略該結(jié)果，原理有點(diǎn)像光學(xué)防抖
上代碼
Integer[] ints = {1, 2, 3, 4, 5, 6, 7, 8, 9};
Observable<String> observable = Observable.from(ints).flatMap(new Func1<Integer, Observable<String>>() {
@Override
public Observable<String> call(Integer integer) {
try {
Thread.currentThread().sleep(200 * integer);
} catch (InterruptedException e) {
e.printStackTrace();
}
return Observable.just(integer + "");
}
});
observable.subscribeOn(Schedulers.newThread())
.debounce(1, TimeUnit.SECONDS)
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Subscriber<String>() {
@Override
public void onCompleted() {
Log.e(TAG, "onCompleted: ");
}
@Override
public void onError(Throwable e) {
Log.e(TAG, "onError: ");
}
@Override
public void onNext(String s) {
Log.e(TAG, "onNext: " + s);
}
});
輸出信息
onNext: 4
onNext: 5
onNext: 6
onNext: 7
onNext: 8
onNext: 9
onCompleted:
這個(gè)輸出數(shù)據(jù)不一定一樣虱饿，有可能從5開(kāi)始拥诡。
Distinct
這個(gè)比較好理解，它就是過(guò)濾掉重復(fù)的數(shù)據(jù)氮发，只允許還沒(méi)有發(fā)射過(guò)的數(shù)據(jù)項(xiàng)通過(guò)渴肉。
示例代碼
Observable.just(0, 0, 6, 4, 2, 8, 2, 1, 9, 0)
.distinct()
.subscribe(new Subscriber<Integer>() {
@Override
public void onCompleted() {
Log.e(TAG, "onCompleted:Distinct ");
}
@Override
public void onError(Throwable e) {
Log.e(TAG, "onError:Distinct ");
}
@Override
public void onNext(Integer integer) {
Log.e(TAG, "onNext:Distinct " + integer);
}
});
輸出日志信息
onNext:Distinct 0
onNext:Distinct 6
onNext:Distinct 4
onNext:Distinct 2
onNext:Distinct 8
onNext:Distinct 1
onNext:Distinct 9
onCompleted:Distinct
ElementAt
該操作符獲取原始Observable發(fā)射的數(shù)據(jù)序列指定索引位置的數(shù)據(jù)項(xiàng)，然后當(dāng)做自己的唯一數(shù)據(jù)發(fā)射爽冕。給它傳遞一個(gè)基于0的索引值仇祭，它會(huì)發(fā)射原始Observable數(shù)據(jù)序列對(duì)應(yīng)索引位置的值，如果你傳遞給elementAt的值為4颈畸，那么它會(huì)發(fā)射第5項(xiàng)的數(shù)據(jù)乌奇。如下示例代碼
Observable.just(0, 0, 6, 4, 2, 8, 2, 1, 9, 0)
.elementAt(4)
.subscribe(new Subscriber<Integer>() {
@Override
public void onCompleted() {
Log.e(TAG, "onCompleted:ElementAt ");
}
@Override
public void onError(Throwable e) {
Log.e(TAG, "onError:ElementAt ");
}
@Override
public void onNext(Integer integer) {
Log.e(TAG, "onNext:ElementAt " + integer);
}
});
輸出日志信息
onNext:ElementAt 2
onCompleted:ElementAt
IgnoreElements
操作符抑制原始Observable發(fā)射的所有數(shù)據(jù)，只允許它的終止通知（onError或onCompleted）通過(guò)眯娱，使用該操作符onNext()方法不會(huì)執(zhí)行礁苗。
Observable.just(1, 2, 3).ignoreElements().subscribe(new Subscriber<Integer>() {
@Override
public void onCompleted() {
Log.e(TAG, "onCompleted");
}
@Override
public void onError(Throwable e) {
Log.e(TAG, "onError");
}
@Override
public void onNext(Integer integer) {
Log.e(TAG, "onNext");
}
});
執(zhí)行后只會(huì)輸出onCompleted。這個(gè)操作符效果就如同empty（）方法創(chuàng)建一個(gè)空的Observable,只會(huì)執(zhí)行onCompleted()方法徙缴，不同的是ignoreElements是對(duì)數(shù)據(jù)源的處理试伙，而empty（）是創(chuàng)建Observable。
今天的這篇文章就到此結(jié)束于样，歡迎大家閱讀疏叨，若發(fā)現(xiàn)文中有錯(cuò)誤的地方歡迎留言提出，感謝穿剖。
