從16年11月份推出RxJava 2.0 ，到現(xiàn)在差不多大半年的時(shí)間里轧钓，RxJava已經(jīng)來(lái)到了2.x時(shí)代序厉，RxJava 1.x 可能也慢慢地被2.x 代替。RxJava 2.x在Reactive-Streams規(guī)范的基礎(chǔ)上從頭開(kāi)始完全重寫(xiě)毕箍，雖然操作符基本沒(méi)有發(fā)生變化弛房，但是因?yàn)镽eactive-Streams具有不同的架構(gòu)，因此對(duì)一些眾所周知的RxJava類型進(jìn)行了更改霉晕。
RxJava 2.x相對(duì)與1.x還是有很多的不同庭再，RxJava的類型更改了捞奕，很多類的命名和方法的命名發(fā)生了變化(可能功能與1.x相同),要想從RxJava 1.x 順利地過(guò)渡到2.x, 就得了解這些變化牺堰。因此本教程就帶你了解這些變化，從而能更快地上手RxJava2.x颅围。
本篇文章就來(lái)先了解一下RxJava 2.x的5種響應(yīng)類型伟葫。
一、RxJava 2.x中5種類型介紹
1 . Observable and Flowable
關(guān)于在RxJava 0.x版本引入背壓的一個(gè)小的遺憾是院促，沒(méi)有設(shè)計(jì)一個(gè)單獨(dú)的基礎(chǔ)反應(yīng)類筏养，而是對(duì)Observable本身進(jìn)行了改裝。背壓的主要問(wèn)題在于熱源（如：UI事件）常拓，不能合理地反壓并導(dǎo)致不可預(yù)料的異常MissingBackpressureException渐溶，這是初學(xué)者不期望看到的。
在RxJava 2.x版本中修復(fù)了這種情況弄抬，將o.reactivex.Observable作為非背壓茎辐，引入新的io.reactivex.Flowable作為啟用背壓基礎(chǔ)反應(yīng)類。
好消息是掂恕，在2.x版本中拖陆，主要的操作符保持不變（同1.x版本），壞消息是懊亡，在導(dǎo)入的時(shí)候應(yīng)當(dāng)小心依啰，它可能會(huì)無(wú)意的選擇非背壓的o.reactivex.Observable.
我們應(yīng)該選哪種？
當(dāng)構(gòu)建數(shù)據(jù)流（作為RxJava的最終消費(fèi)者）或考慮您的2.x兼容庫(kù)應(yīng)該采取和返回什么類型時(shí)店枣，您可以考慮幾個(gè)因素速警，以幫助您避免諸如MissingBackpressureException或OutOfMemoryError之類的問(wèn)題。
Observable使用場(chǎng)景:
- 數(shù)據(jù)流最長(zhǎng)不超過(guò)1000個(gè)元素鸯两，即隨著時(shí)間的流逝闷旧，應(yīng)用沒(méi)有機(jī)會(huì)報(bào)OOME(OutOfMemoryError)錯(cuò)誤。
- 處理諸如鼠標(biāo)移動(dòng)或觸摸事件之類的GUI事件
Flowable使用場(chǎng)景:
- 處理超過(guò)10K+ 的元素流
- 從磁盤(pán)讀人ψ俊（解析文件）
- 從數(shù)據(jù)庫(kù)讀取數(shù)據(jù)
- 從網(wǎng)絡(luò)獲取數(shù)據(jù)流
2 . Single 使用介紹
Single是2.x版本中的一種基礎(chǔ)響應(yīng)類型鸠匀，Single是從頭開(kāi)始重新設(shè)計(jì)的，能單獨(dú)發(fā)射一個(gè)onSuccess或者onError事件逾柿，它現(xiàn)在的架構(gòu)來(lái)自于the Reactive-Streams設(shè)計(jì)缀棍。它的消費(fèi)者類型已經(jīng)從接受rx.Subscription的rx.Single.SingleSubscriber<T>變?yōu)榱?code>io.reactivex.SingleObserver<T>,有3個(gè)方法：
interface SingleObserver<T> {
void onSubscribe(Disposable d);
void onSuccess(T value);
void onError(Throwable error);
}
3 . Completable使用介紹
Completeble類型基本保持不變宅此，1.x的版本已經(jīng)沿著Reactive-Streams風(fēng)格設(shè)計(jì)，所以沒(méi)有用戶級(jí)別的更改爬范。
相似地命名改變父腕，rx.Completable.CompletableSubscriber變?yōu)閹в?code>onSubscribe(Disposable)方法的io.reactivex.CompletableObserver:
interface CompletableObserver<T> {
void onSubscribe(Disposable d);
void onComplete();
void onError(Throwable error);
}
4 . Maybe 使用介紹
RxJava 2.0.0-RC2 介紹了一個(gè)新的基礎(chǔ)響應(yīng)類型，它叫做Maybe青瀑。從概念上來(lái)將璧亮，它像是 Single和Completable的結(jié)合，它可能發(fā)射0個(gè)或者1個(gè)項(xiàng)目斥难，或者一個(gè)error信號(hào)枝嘶。
Maybe類通過(guò)依賴MaybeSource作為它的基礎(chǔ)接口類型MaybeObserver<T>作為信號(hào)響應(yīng)接口并且遵循協(xié)議onSubscribe (onSuccess | onError | onComplete)?因?yàn)樽疃嗫赡馨l(fā)射1個(gè)元素，所以Maybe類型沒(méi)有背壓的概念（因?yàn)樗鼪](méi)有像Flowable和Observable一樣有未知長(zhǎng)度的可膨脹緩沖區(qū)）
這意味著onSubscribe(Disposable)的調(diào)用潛在地跟隨著其他onXXX方法之一的調(diào)用哑诊，不同于Flowable,如果這兒只有一個(gè)信號(hào)值發(fā)射信號(hào)群扶，那么只有onSuccess被調(diào)用，而不會(huì)調(diào)用complete镀裤。
二竞阐、RxJava 2.x中5種類型使用示例
1 . Observable示例
在寫(xiě)示例之前，我們先來(lái)回顧一下 1.x 版本是如何創(chuàng)建Observable和如何訂閱的：
RxJava 1.x :
//創(chuàng)建 observable
Observable observable = Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("hello world");
subscriber.onCompleted();
}
}).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread());
//訂閱方式一
observable.subscribe(new Subscriber() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Object o) {
}
});
// 訂閱方式二
observable.subscribe(new Action1() {
@Override
public void call(Object o) {
// onNext
}
});
通過(guò)create方法創(chuàng)建Observable,接收一個(gè)OnSubscribe接口暑劝，其中有一個(gè)回調(diào)方法call骆莹，參數(shù)為Subscriber,我們用Subscriber來(lái)發(fā)射數(shù)據(jù)。通過(guò)subscribe方法來(lái)訂閱担猛，可以接收一個(gè)Subscriber 實(shí)現(xiàn)全部方法幕垦，也可以接收一個(gè)Action1只實(shí)現(xiàn)onNext方法。
RxJava 2.x :
//創(chuàng)建Observable
Observable observable = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
Log.e(TAG,"start emitter data");
e.onNext("Hello");
e.onNext("world");
e.onComplete();
}
})
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread());
// 訂閱方式一：下游消費(fèi)者 Observer
observable.subscribe(new Observer<String>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
// onSubscribe 是2.x新添加的方法毁习，在發(fā)射數(shù)據(jù)前被調(diào)用智嚷，相當(dāng)于1.x的onStart方法
Log.e(TAG,"onSubscribe");
}
@Override
public void onNext(@NonNull String s) {
Log.e(TAG,"onNext");
Log.e(TAG,s);
}
@Override
public void onError(@NonNull Throwable e) {
Log.e(TAG,"onError");
}
@Override
public void onComplete() {
Log.e(TAG,"onComplete");
}
});
// 訂閱方式二：Consumer
observable.subscribe(new Consumer<String>() {
@Override
public void accept(@NonNull String o) throws Exception {
Log.e(TAG,"consumer:"+o);
}
});
打印結(jié)果如下：
06-25 14:31:35.435 21505-21505/com.zhouwei.demoforrxjava2 E/MainActivity: onSubscribe
06-25 14:31:35.437 21505-21853/com.zhouwei.demoforrxjava2 E/MainActivity: start emitter data
06-25 14:31:35.438 21505-21505/com.zhouwei.demoforrxjava2 E/MainActivity: onNext:Hello
06-25 14:31:35.438 21505-21505/com.zhouwei.demoforrxjava2 E/MainActivity: onNext:world
06-25 14:31:35.438 21505-21505/com.zhouwei.demoforrxjava2 E/MainActivity: onComplete
其實(shí)我們可以對(duì)比一下，1.x 和 2.x 方法都試一樣的纺且，只是它們所接收的響應(yīng)接口改變了,對(duì)應(yīng)變化如下：
RxJava 1.x －> RxJava 2.x
---------------------------------------
OnSubscribe<T> －> ObservableOnSubscribe<T>
Subscriber<T> －> Observer<T>
Subscriber<T> －> ObservableEmitter<T>
Action1<T> －> Consumer<T>
RxJava 2.x 中對(duì)這些接口進(jìn)行了重新設(shè)計(jì)盏道，讓一個(gè)接口的職責(zé)更加單一，類的命名和方法命名與它的功能更佳符合（見(jiàn)名知意）载碌。如在1.x 中猜嘱，Subscriber 既能發(fā)射數(shù)據(jù)，又能消費(fèi)數(shù)據(jù)嫁艇，充當(dāng)觀察者和被觀察者朗伶。在2.x 中 把它拆解成了2個(gè)接口。
ObservableEmitter<T>專門(mén)用來(lái)發(fā)射數(shù)據(jù)步咪，Consumer專門(mén)用來(lái)消費(fèi)數(shù)據(jù)论皆。 除此之外，在RxJava 2.x 中，多了一個(gè)void onSubscribe(@NonNull Disposable d)回調(diào)方法点晴，參數(shù)為Disposable,Disposable是用來(lái)解除訂閱關(guān)系的感凤，這讓我們的解除訂閱變得更佳容易（比起1.x 通過(guò)subscribe返回 Subscription）。
上面對(duì)比了在RxJava 1.x 和2.x 版本創(chuàng)建Observable的方式粒督，其實(shí)在RxJava 2.x中陪竿，這5種類型的用法是非常相似的，它們的接口命名規(guī)則相同屠橄，只要你知道其中一種族跛，就知道其他幾種類型該如何在上游發(fā)射數(shù)據(jù)和在下游消費(fèi)數(shù)據(jù)。create接收的類型都為xxxOnSubscrible(xxx為5種類型對(duì)應(yīng)的名字),發(fā)射器為xxxEmitter锐墙，具體如下表：
| RxJava 2.x 類型 | create參數(shù)（響應(yīng)接口） | 發(fā)射器 | Observer |
|---|---|---|---|
| Observable | ObservableOnSubscribe | ObservableEmitter | Observer |
| Flowable | FlowableOnSubscribe | FlowableEmitter | FlowableSubscriber |
| Single | SingleOnSubscribe | SingleEmitter | SingleObserver |
| Completable | CompletableOnSubscribe | CompletableEmitter | CompletableObserver |
| Maybe | MaybeOnSubscribe | MaybeEmitter | MaybeObserver |
2 . Flowable示例
上面對(duì)比了RxJava 1.x 和 2.x 創(chuàng)建使用Observable的方式礁哄，并且總結(jié)了2.x 相關(guān)類的改變，如上面表贮匕。那么使用Flowable的方式和Observable是很相似的姐仅，看一下代碼：
Flowable.create(new FlowableOnSubscribe<Integer>() {
@Override
public void subscribe(@NonNull FlowableEmitter<Integer> e) throws Exception {
Log.e(TAG,"start send data ");
for(int i=0;i<100;i++){
e.onNext(i);
}
e.onComplete();
}
}, BackpressureStrategy.DROP)//指定背壓策略
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new FlowableSubscriber<Integer>() {
@Override
public void onSubscribe(@NonNull Subscription s) {
//1， onSubscribe 是2.x新添加的方法刻盐，在發(fā)射數(shù)據(jù)前被調(diào)用，相當(dāng)于1.x的onStart方法
//2劳翰， 參數(shù)為 Subscription 敦锌，Subscription 可用于向上游請(qǐng)求發(fā)射多少個(gè)元素，也可用于取笑請(qǐng)求
//3, 必須要調(diào)用Subscription 的request來(lái)請(qǐng)求發(fā)射數(shù)據(jù)佳簸，不然上游是不會(huì)發(fā)射數(shù)據(jù)的乙墙。
Log.e(TAG,"onSubscribe...");
subscription = s;
s.request(100);
}
@Override
public void onNext(Integer integer) {
Log.e(TAG,"onNext:"+integer);
}
@Override
public void onError(Throwable t) {
Log.e(TAG,"onError..."+t);
}
@Override
public void onComplete() {
Log.e(TAG,"onComplete...");
}
});
Flowable和Observable的使用基本相同，只不過(guò)Observable不支持背壓生均，而Flowable支持背壓听想。使用的時(shí)候，還是要注意幾個(gè)小細(xì)節(jié)：
1马胧，創(chuàng)建Flowable的時(shí)候需要指定一個(gè)背壓策略汉买，本文使用的是PBackpressureStrategy.DROP（丟棄策略），RxJava 2.x中佩脊，內(nèi)置了5種背壓策略蛙粘，由于篇幅有限，背壓和背壓策略下一篇拿出來(lái)單獨(dú)講威彰。
2出牧，onSubscribe 回調(diào)方法中，參數(shù)是Subscription而不是Disposable,前文說(shuō)過(guò)歇盼，RxJava 2.x 中舔痕，訂閱的管理?yè)Q成了Disposable，但是Flowable使用的是Subscription，這個(gè)Subscription不是1.x 版本中的Subscription伯复，雖然它有取消訂閱的能力盈咳。主要用于請(qǐng)求上游元素和取消訂閱。
3边翼，在使用Flowable的時(shí)候鱼响，必須調(diào)用Subscription 的requsest方法請(qǐng)求，不然上游是不會(huì)發(fā)射數(shù)據(jù)的组底≌苫看request的方法解釋：
3 . Single、Completable 和 Maybe 示例
Single债鸡、Completable和Maybe就比較簡(jiǎn)單江滨，Single用于只發(fā)射一個(gè)數(shù)據(jù)，Completable不發(fā)送數(shù)據(jù)厌均，它給下游發(fā)射一個(gè)信號(hào)唬滑。而Maybe則是Single和Completable的結(jié)合，根據(jù)名字就可以看出棺弊，它的結(jié)果是不確定的晶密，可能發(fā)發(fā)射0(Completable)或1(Single) 個(gè)元素,或者收到一個(gè)Error信號(hào)。
Single示例：
Single.create(new SingleOnSubscribe<Boolean>() {
@Override
public void subscribe(@NonNull SingleEmitter<Boolean> e) throws Exception {
Log.e(TAG,"subscribe...");
e.onSuccess(true);
}
})
.observeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new SingleObserver<Boolean>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
Log.e(TAG,"onSubscribe...");
}
@Override
public void onSuccess(@NonNull Boolean aBoolean) {
Log.e(TAG,"onSuccess...");
}
@Override
public void onError(@NonNull Throwable e) {
Log.e(TAG,"onError...");
}
});
Single只發(fā)射一個(gè)元素模她，所以沒(méi)有complete 方法稻艰，不像Observable或者Flowable，數(shù)據(jù)發(fā)射完成之后侈净，需要調(diào)用complete告訴下游已經(jīng)完成尊勿。
Completable示例：
Completable.create(new CompletableOnSubscribe() {
@Override
public void subscribe(@NonNull CompletableEmitter e) throws Exception {
Log.e(TAG,"start send data");
e.onComplete();
}
}).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new CompletableObserver() {
@Override
public void onSubscribe(@NonNull Disposable d) {
Log.e(TAG,"onSubscribe");
}
@Override
public void onComplete() {
Log.e(TAG,"onComplete");
}
@Override
public void onError(@NonNull Throwable e) {
Log.e(TAG,"onError");
}
});
Completable 不會(huì)發(fā)射數(shù)據(jù)，只會(huì)給下游發(fā)送一個(gè)信號(hào)畜侦≡樱回調(diào) onComplete方法。
Maybe示例：
Maybe.create(new MaybeOnSubscribe<Boolean>() {
@Override
public void subscribe(@NonNull MaybeEmitter<Boolean> e) throws Exception {
Log.e(TAG,"start send data");
e.onSuccess(true);
e.onComplete();
}
}).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new MaybeObserver<Boolean>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
Log.e(TAG,"onSubscribe");
}
@Override
public void onSuccess(@NonNull Boolean aBoolean) {
Log.e(TAG,"1->onSuccess:"+aBoolean);
}
@Override
public void onError(@NonNull Throwable e) {
Log.e(TAG,"onError");
}
@Override
public void onComplete() {
Log.e(TAG,"onComplete");
}
});
Maybe是Single和Completable的結(jié)合旋膳，需要注意的是
onSuccess和onComplete方法只會(huì)執(zhí)行其中一個(gè)澎语，這不同于Observable和Flowable最后是以onComplete()結(jié)尾.
如上面的代碼，先調(diào)用onSuccess發(fā)射一個(gè)元素溺忧，再調(diào)用onComplete咏连，
e.onComplete();
e.onSuccess(true);
最后打印結(jié)果如下：
E/MainActivity: onSubscribe
E/MainActivity: start send data
E/MainActivity: onComplete
可以看到只回調(diào)了 onComplete,我們把調(diào)用的順序調(diào)換一下:
e.onSuccess(true);
e.onComplete();
打印結(jié)果如下：
E/MainActivity: onSubscribe
E/MainActivity: start send data
E/MainActivity: 1->onSuccess:true
可以看到調(diào)換了之后打印OnSucces()而沒(méi)有打印onComplete(),這也印證了只會(huì)回調(diào)其中之一。
三鲁森、總結(jié)
RxJava 2.x 相比于 1.x 還是有很大的變化祟滴，雖然操作符基本不變，但是很多類和接口都是基于Reactive-Streams 規(guī)范重新設(shè)計(jì)的歌溉，命名也發(fā)生了變換垄懂，要想玩轉(zhuǎn)RxJava 2.x 骑晶，你得了解這些變化和使用場(chǎng)景，本文介紹了RxJava 2.x 的5種基礎(chǔ)響應(yīng)類型草慧，希望對(duì)才開(kāi)始學(xué)習(xí)RxJava 2.x 的同學(xué)有所幫助桶蛔。
