經(jīng)過上一篇文章 Android拾萃 - RxJava最簡單的入門（一） 的學(xué)習(xí)，我們對(duì)整個(gè)流程有了一定的把握。今天我們就來看看操作符。RxJava中提供了大量不同種類，不同場景的Operators(操作符)，RxJava的強(qiáng)大性就來自于它所定義的操作符。
- 操作符讓你可以變換、組合、操縱和處理Observable發(fā)射的數(shù)據(jù)。
- Rx的操作符讓你可以用聲明式的風(fēng)格組合異步操作序列，它擁有回調(diào)的所有效率優(yōu)勢(shì)，同時(shí)又避免了典型的異步系統(tǒng)中嵌套回調(diào)的缺點(diǎn)。
主要分類：
| RxJava 的操作符 | 說明 | 例如 |
|---|---|---|
| 創(chuàng)建操作 | 用于創(chuàng)建Observable的操作符 | create、defer、from、just、start、repeat、range |
| 變換操作 | 用于對(duì)Observable發(fā)射的數(shù)據(jù)進(jìn)行變換 | buffer、window、map、flatMap、groupBy、scan |
| 過濾操作 | 用于從Observable發(fā)射的數(shù)據(jù)中進(jìn)行選擇 | debounce、distinct、filter、sample、skip、take |
| 組合操作 | 用于將多個(gè)Observable組合成一個(gè)單一的Observable | and、startwith、join、merge、switch、zip |
| 異常處理 | 用于從錯(cuò)誤通知中恢復(fù) | catch、retry |
| 輔助操作 | 用于處理Observable的操作符 | delay、do、observeOn、subscribeOn、subscribe |
| 條件和布爾操作 | all、amb、contains、skipUntil、takeUntil | |
| 算法和聚合操作 | average、concat、count、max、min、sum、reduce | |
| 異步操作 | start、toAsync、startFuture、FromAction、FromCallable、runAsync | |
| 連接操作 | connect、publish、refcount、replay | |
| 轉(zhuǎn)換操作 | toFuture、toList、toIterable、toMap、toMultiMap | |
| 阻塞操作 | forEach、first、last、mostRecent、next、single | |
| 字符串操作 | byLine、decode、encode、from、join、split、stringConcat |
其中有一些高頻使用的操作符如下：
| 常用操作符 | 說明 |
|---|---|
| interval | 延時(shí)幾秒，每隔幾秒開始執(zhí)行 |
| take | 超過多少秒停止執(zhí)行 |
| map | 類型轉(zhuǎn)換 |
| observeOn | 在主線程運(yùn)行 |
| doOnSubscribe | 在執(zhí)行的過程中 |
| subscribe | 訂閱 |
RxJava 2.x的5種響應(yīng)類型。
上游響應(yīng)類型可以簡單理解為被觀察者的回調(diào)函數(shù)，下游消費(fèi)類型就是觀察者，這樣上面的圖就很清晰了。
在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 |
ObservableOnSubscribe： Observable 對(duì)應(yīng)的上游類型是ObservableOnSubscribe 我們需要發(fā)射數(shù)據(jù)，所以實(shí)現(xiàn)了ObservableEmitter方法
ObservableEmitter： ObservableEmitter可以理解為發(fā)射器，這個(gè)就是用來發(fā)出事件的，它可以發(fā)出三種類型的事件。
Observer: 觀察者訂閱者
Consumer ： 消費(fèi)者 如果我們并不關(guān)心OnComplete和OnError，我們只需要在onNext的時(shí)候做一些處理，這時(shí)候就可以使用Consumer類。所以我們可以認(rèn)為Consumer是一個(gè)不完整的Observer(其實(shí)和rxjava1.0的action一樣，但是rxjava2.0開始，actiOn表示的無參數(shù)返回，具體見下面對(duì)比)
Single/SingleObserver： 是只發(fā)射一次，一次性的
Completable/CompletableObserver：只關(guān)注完成狀態(tài)的
Maybe/MaybeObserver：可以說是前兩者的復(fù)合體
Action 在Rx1.0是從Action0，Action1...往后排（數(shù)字代表可接受的參數(shù)），Rx2.0做出了改動(dòng)
Rx1.0-----------Rx2.0
Action0--------Action（無參的onComplete()）
Action1--------Consumer（單個(gè)參數(shù)的onError(Throwable t)和onNext(T t)）
Action2--------BiConsumer
后面的Action都去掉了，只保留了ActionN
或者理解為 無參數(shù)的Action0 被操作符io.reactivex.functions.Action和Scheduler代替。Action1被重命名為Consumer。Action2 被重命名為BiConsumer。 ActionN 被Consumer<Object[]> 代替。
Observable使用場景:
- 數(shù)據(jù)流最長不超過1000個(gè)元素，即隨著時(shí)間的流逝，應(yīng)用沒有機(jī)會(huì)報(bào)OOME(OutOfMemoryError)錯(cuò)誤。
- 處理諸如鼠標(biāo)移動(dòng)或觸摸事件之類的GUI事件
Flowable使用場景:
- 處理超過10K+ 的元素流
- 從磁盤讀?。ń馕鑫募?/li>
- 從數(shù)據(jù)庫讀取數(shù)據(jù)
- 從網(wǎng)絡(luò)獲取數(shù)據(jù)流
在onSubscribe/onStart中調(diào)用request
注意，在Subscriber.onSubscribe或ResourceSubscriber.onStart中調(diào)用request去請(qǐng)求資源，參數(shù)就是要請(qǐng)求的數(shù)量，一般如果不限制請(qǐng)求數(shù)量，可以寫成Long.MAX_VALUE。如果你不調(diào)用request，Subscriber的onNext和onComplete方法將不會(huì)被調(diào)用。當(dāng)調(diào)用request(n)時(shí)，將會(huì)立即調(diào)用onNext。
實(shí)例代碼如下：
Flowable.range(1, 3).subscribe(new Subscriber<Integer>() {
//rx2 新加入的方法，在訂閱后發(fā)送數(shù)據(jù)之前，
//回首先調(diào)用這個(gè)方法，而Disposable可用于取消訂閱
@Override
public void onSubscribe(Subscription s) {
System.out.println("OnSubscribe start");
s.request(Long.MAX_VALUE);//用request(n)將會(huì)立即調(diào)用onNext,批量發(fā)數(shù)據(jù)
System.out.println("OnSubscribe end");
}
@Override
public void onNext(Integer v) {
System.out.println(v);
}
@Override
public void onError(Throwable e) {
e.printStackTrace();
}
@Override
public void onComplete() {
System.out.println("Done");
}
});
打印結(jié)果
OnSubscribe start
1
2
3
Done
OnSubscribe end
###### Single 使用（監(jiān)聽單一事件）
是Observable的變種，只能發(fā)射一個(gè)值或一個(gè)錯(cuò)誤通知，有3個(gè)方法：
Single只會(huì)調(diào)用onSuccess onError這兩個(gè)方法中的一個(gè)，而且只會(huì)調(diào)用一次，調(diào)用了任何一個(gè)方法之后，訂閱關(guān)系終止。
interface SingleObserver<T> {
void onSubscribe(Disposable d);
void onSuccess(T value);
void onError(Throwable error);
}
###### Completable使用（只監(jiān)聽 completed or failed 事件）
Completable 本質(zhì)上來說和 Observable 與 Single 不一樣，因?yàn)樗?*不發(fā)射數(shù)據(jù)**。因此 Completable 的操作符也有所區(qū)別，最常用的是andThen。在這個(gè)操作符中你可以傳任何Observable、Single、Flowable、Maybe或者其他Completable，它們會(huì)在原來的 Completable 結(jié)束后執(zhí)行。
這里不發(fā)射數(shù)據(jù)了，也就沒有sumer發(fā)射器了，但是多了一個(gè)Action，當(dāng)然，它也是不發(fā)射數(shù)據(jù)的。
###### Maybe
Maybe 是 RxJava2.x 之后才有的新類型，可以看成是Single和Completable的結(jié)合。
Maybe創(chuàng)建之后，MaybeEmitter 和 SingleEmitter 一樣并沒有onNext()方法，同樣需要通過onSuccess()方法來發(fā)射數(shù)據(jù)。
Maybe也只能發(fā)射0或者1個(gè)數(shù)據(jù)，即使發(fā)射多個(gè)數(shù)據(jù)，后面發(fā)射的數(shù)據(jù)也不會(huì)處理。
####參考
+ [ReactiveX/RxJava文檔中文版](https://mcxiaoke.gitbooks.io/rxdocs/content/https://mcxiaoke.gitbooks.io/rxdocs/content/)