這可能是最好的 RxJava 2.x 入門(mén)教程系列專(zhuān)欄
文章鏈接：
這可能是最好的 RxJava 2.x 入門(mén)教程(完結(jié)版)[推薦直接看這個(gè)]
這可能是最好的RxJava 2.x 入門(mén)教程（一）
這可能是最好的RxJava 2.x 入門(mén)教程（二）
這可能是最好的RxJava 2.x 入門(mén)教程（三）
這可能是最好的RxJava 2.x 入門(mén)教程（四）
這可能是最好的RxJava 2.x 入門(mén)教程（五）
GitHub 代碼同步更新：https://github.com/nanchen2251/RxJava2Examples
為了滿足大家的饑渴難耐性含，GitHub 將同步更新代碼辞槐，主要包含基本的代碼封裝瞧栗，RxJava 2.x 所有操作符應(yīng)用場(chǎng)景介紹和實(shí)際應(yīng)用場(chǎng)景山孔，后期除了 RxJava 可能還會(huì)增添其他東西，總之祠墅，GitHub 上的 Demo 專(zhuān)為大家傾心打造腐缤。傳送門(mén)：https://github.com/nanchen2251/RxJava2Examples
前言
很快我們就迎來(lái)了第二期慢蜓，上一期我們主要講解了 RxJava 1.x 到 2.x 的變化概覽，相信各位熟練掌握RxJava 1.x的老司機(jī)們隨便看一下變化概覽就可以上手RxJava 2.x了鼠锈，但為了滿足更廣大的年輕一代司機(jī)（未來(lái)也是老司機(jī)）闪檬，在本節(jié)中，我們將學(xué)習(xí)RxJava 2.x 強(qiáng)大的操作符章節(jié)购笆。
【注】以下所有操作符標(biāo)題都可直接點(diǎn)擊進(jìn)入官方doc查看粗悯。
正題
Create
create 操作符應(yīng)該是最常見(jiàn)的操作符了，主要用于產(chǎn)生一個(gè) Obserable 被觀察者對(duì)象同欠，為了方便大家的認(rèn)知样傍，以后的教程中統(tǒng)一把被觀察者 Observable 稱(chēng)為發(fā)射器（上游事件），觀察者 Observer 稱(chēng)為接收器（下游事件）铺遂。
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
mRxOperatorsText.append("Observable emit 1" + "\n");
Log.e(TAG, "Observable emit 1" + "\n");
e.onNext(1);
mRxOperatorsText.append("Observable emit 2" + "\n");
Log.e(TAG, "Observable emit 2" + "\n");
e.onNext(2);
mRxOperatorsText.append("Observable emit 3" + "\n");
Log.e(TAG, "Observable emit 3" + "\n");
e.onNext(3);
e.onComplete();
mRxOperatorsText.append("Observable emit 4" + "\n");
Log.e(TAG, "Observable emit 4" + "\n" );
e.onNext(4);
}
}).subscribe(new Observer<Integer>() {
private int i;
private Disposable mDisposable;
@Override
public void onSubscribe(@NonNull Disposable d) {
mRxOperatorsText.append("onSubscribe : " + d.isDisposed() + "\n");
Log.e(TAG, "onSubscribe : " + d.isDisposed() + "\n" );
mDisposable = d;
}
@Override
public void onNext(@NonNull Integer integer) {
mRxOperatorsText.append("onNext : value : " + integer + "\n");
Log.e(TAG, "onNext : value : " + integer + "\n" );
i++;
if (i == 2) {
// 在RxJava 2.x 中衫哥，新增的Disposable可以做到切斷的操作，讓Observer觀察者不再接收上游事件
mDisposable.dispose();
mRxOperatorsText.append("onNext : isDisposable : " + mDisposable.isDisposed() + "\n");
Log.e(TAG, "onNext : isDisposable : " + mDisposable.isDisposed() + "\n");
}
}
@Override
public void onError(@NonNull Throwable e) {
mRxOperatorsText.append("onError : value : " + e.getMessage() + "\n");
Log.e(TAG, "onError : value : " + e.getMessage() + "\n" );
}
@Override
public void onComplete() {
mRxOperatorsText.append("onComplete" + "\n");
Log.e(TAG, "onComplete" + "\n" );
}
});
輸出：
需要注意的幾點(diǎn)是：
在發(fā)射事件中襟锐，我們?cè)诎l(fā)射了數(shù)值 3 之后撤逢，直接調(diào)用了
e.onComlete()，雖然無(wú)法接收事件粮坞，但發(fā)送事件還是繼續(xù)的蚊荣。另外一個(gè)值得注意的點(diǎn)是，在 RxJava 2.x 中莫杈，可以看到發(fā)射事件方法相比 1.x 多了一個(gè) throws Excetion互例，意味著我們做一些特定操作再也不用 try-catch 了。
并且 2.x 中有一個(gè)
Disposable概念姓迅，這個(gè)東西可以直接調(diào)用切斷敲霍，可以看到俊马，當(dāng)它的isDisposed()返回為 false 的時(shí)候丁存，接收器能正常接收事件，但當(dāng)其為 true 的時(shí)候柴我，接收器停止了接收解寝。所以可以通過(guò)此參數(shù)動(dòng)態(tài)控制接收事件了。
Map
Map 基本算是 RxJava 中一個(gè)最簡(jiǎn)單的操作符了艘儒，熟悉 RxJava 1.x 的知道聋伦，它的作用是對(duì)發(fā)射時(shí)間發(fā)送的每一個(gè)事件應(yīng)用一個(gè)函數(shù)夫偶，是的每一個(gè)事件都按照指定的函數(shù)去變化，而在 2.x 中它的作用幾乎一致觉增。
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
e.onNext(1);
e.onNext(2);
e.onNext(3);
}
}).map(new Function<Integer, String>() {
@Override
public String apply(@NonNull Integer integer) throws Exception {
return "This is result " + integer;
}
}).subscribe(new Consumer<String>() {
@Override
public void accept(@NonNull String s) throws Exception {
mRxOperatorsText.append("accept : " + s +"\n");
Log.e(TAG, "accept : " + s +"\n" );
}
});
輸出：
是的兵拢，map 基本作用就是將一個(gè) Observable 通過(guò)某種函數(shù)關(guān)系，轉(zhuǎn)換為另一種 Observable逾礁，上面例子中就是把我們的 Integer 數(shù)據(jù)變成了 String 類(lèi)型说铃。從Log日志顯而易見(jiàn)。
Zip
zip 專(zhuān)用于合并事件嘹履，該合并不是連接（連接操作符后面會(huì)說(shuō)）腻扇，而是兩兩配對(duì)，也就意味著砾嫉，最終配對(duì)出的 Observable 發(fā)射事件數(shù)目只和少的那個(gè)相同幼苛。
Observable.zip(getStringObservable(), getIntegerObservable(), new BiFunction<String, Integer, String>() {
@Override
public String apply(@NonNull String s, @NonNull Integer integer) throws Exception {
return s + integer;
}
}).subscribe(new Consumer<String>() {
@Override
public void accept(@NonNull String s) throws Exception {
mRxOperatorsText.append("zip : accept : " + s + "\n");
Log.e(TAG, "zip : accept : " + s + "\n");
}
});
private Observable<String> getStringObservable() {
return Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
if (!e.isDisposed()) {
e.onNext("A");
mRxOperatorsText.append("String emit : A \n");
Log.e(TAG, "String emit : A \n");
e.onNext("B");
mRxOperatorsText.append("String emit : B \n");
Log.e(TAG, "String emit : B \n");
e.onNext("C");
mRxOperatorsText.append("String emit : C \n");
Log.e(TAG, "String emit : C \n");
}
}
});
}
private Observable<Integer> getIntegerObservable() {
return Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
if (!e.isDisposed()) {
e.onNext(1);
mRxOperatorsText.append("Integer emit : 1 \n");
Log.e(TAG, "Integer emit : 1 \n");
e.onNext(2);
mRxOperatorsText.append("Integer emit : 2 \n");
Log.e(TAG, "Integer emit : 2 \n");
e.onNext(3);
mRxOperatorsText.append("Integer emit : 3 \n");
Log.e(TAG, "Integer emit : 3 \n");
e.onNext(4);
mRxOperatorsText.append("Integer emit : 4 \n");
Log.e(TAG, "Integer emit : 4 \n");
e.onNext(5);
mRxOperatorsText.append("Integer emit : 5 \n");
Log.e(TAG, "Integer emit : 5 \n");
}
}
});
}
輸出：
需要注意的是：
zip組合事件的過(guò)程就是分別從發(fā)射器 A 和發(fā)射器 B 各取出一個(gè)事件來(lái)組合，并且一個(gè)事件只能被使用一次焕刮，組合的順序是嚴(yán)格按照事件發(fā)送的順序來(lái)進(jìn)行的舶沿，所以上面截圖中，可以看到济锄，1 永遠(yuǎn)是和 A 結(jié)合的暑椰，2 永遠(yuǎn)是和 B 結(jié)合的。最終接收器收到的事件數(shù)量是和發(fā)送器發(fā)送事件最少的那個(gè)發(fā)送器的發(fā)送事件數(shù)目相同荐绝，所以如截圖中一汽，5 很孤單，沒(méi)有人愿意和它交往低滩，孤獨(dú)終老的單身狗召夹。
Concat
對(duì)于單一的把兩個(gè)發(fā)射器連接成一個(gè)發(fā)射器，雖然 zip 不能完成恕沫，但我們還是可以自力更生监憎，官方提供的 concat 讓我們的問(wèn)題得到了完美解決。
Observable.concat(Observable.just(1,2,3), Observable.just(4,5,6))
.subscribe(new Consumer<Integer>() {
@Override
public void accept(@NonNull Integer integer) throws Exception {
mRxOperatorsText.append("concat : "+ integer + "\n");
Log.e(TAG, "concat : "+ integer + "\n" );
}
});
輸出：
如圖婶溯，可以看到鲸阔。發(fā)射器 B 把自己的三個(gè)孩子送給了發(fā)射器 A，讓他們組合成了一個(gè)新的發(fā)射器迄委，非常懂事的孩子褐筛，有條不紊的排序接收。
FlatMap
FlatMap 是一個(gè)很有趣的東西叙身，我堅(jiān)信你在實(shí)際開(kāi)發(fā)中會(huì)經(jīng)常用到渔扎。它可以把一個(gè)發(fā)射器 Observable 通過(guò)某種方法轉(zhuǎn)換為多個(gè) Observables，然后再把這些分散的 Observables裝進(jìn)一個(gè)單一的發(fā)射器 Observable信轿。但有個(gè)需要注意的是晃痴，flatMap 并不能保證事件的順序残吩，如果需要保證，需要用到我們下面要講的 ConcatMap倘核。
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
e.onNext(1);
e.onNext(2);
e.onNext(3);
}
}).flatMap(new Function<Integer, ObservableSource<String>>() {
@Override
public ObservableSource<String> apply(@NonNull Integer integer) throws Exception {
List<String> list = new ArrayList<>();
for (int i = 0; i < 3; i++) {
list.add("I am value " + integer);
}
int delayTime = (int) (1 + Math.random() * 10);
return Observable.fromIterable(list).delay(delayTime, TimeUnit.MILLISECONDS);
}
}).subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<String>() {
@Override
public void accept(@NonNull String s) throws Exception {
Log.e(TAG, "flatMap : accept : " + s + "\n");
mRxOperatorsText.append("flatMap : accept : " + s + "\n");
}
});
輸出：
一切都如我們預(yù)期中的有意思泣侮，為了區(qū)分 concatMap（下一個(gè)會(huì)講），我在代碼中特意動(dòng)了一點(diǎn)小手腳紧唱，我采用一個(gè)隨機(jī)數(shù)旁瘫，生成一個(gè)時(shí)間，然后通過(guò) delay（后面會(huì)講）操作符琼蚯，做一個(gè)小延時(shí)操作酬凳，而查看 Log 日志也確認(rèn)驗(yàn)證了我們上面的說(shuō)法，它是無(wú)序的遭庶。
concatMap
上面其實(shí)就說(shuō)了宁仔，concatMap 與 FlatMap 的唯一區(qū)別就是 concatMap 保證了順序，所以峦睡，我們就直接把 flatMap 替換為 concatMap 驗(yàn)證吧翎苫。
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
e.onNext(1);
e.onNext(2);
e.onNext(3);
}
}).concatMap(new Function<Integer, ObservableSource<String>>() {
@Override
public ObservableSource<String> apply(@NonNull Integer integer) throws Exception {
List<String> list = new ArrayList<>();
for (int i = 0; i < 3; i++) {
list.add("I am value " + integer);
}
int delayTime = (int) (1 + Math.random() * 10);
return Observable.fromIterable(list).delay(delayTime, TimeUnit.MILLISECONDS);
}
}).subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<String>() {
@Override
public void accept(@NonNull String s) throws Exception {
Log.e(TAG, "flatMap : accept : " + s + "\n");
mRxOperatorsText.append("flatMap : accept : " + s + "\n");
}
});
輸出：
結(jié)果的確和我們預(yù)想的一樣。
寫(xiě)在最后
好了榨了，這一節(jié)就先介紹到這里煎谍，下一節(jié)我們將學(xué)習(xí)其它的一些操作符，在操作符講完后再帶大家進(jìn)入實(shí)際情景龙屉，希望持續(xù)關(guān)注呐粘，代碼傳送門(mén)
