RxJava2 系列 -1：一篇的比較全面的 RxJava2 方法總結
RxJava2 系列 -1：一篇的比較全面的 RxJava2 方法總結
看了許多講解RxJava的文章，有些文章講解的內容是基於第一個版本的，有些文章的講解是通過比較常用的一些API和基礎的概念進行講解的。
但是每次看到RxJava的類中的幾十個方法的時候，總是感覺心裏沒底。所以，我打算自己去專門寫篇文章來從API的角度系統地梳理一下RxJava的各種方法和用法。
1、RxJava 基本
1.1 RxJava 簡介
RxJava是一個在Java VM上使用可觀測的序列來組成非同步的、基於事件的程式的庫。
雖然，在Android中，我們可以使用AsyncTask來完成非同步任務操作，但是當任務的梳理比較多的時候，我們要爲每個任務定義一個AsyncTask就變得非常繁瑣。
RxJava能幫助我們在實現非同步執行的前提下保持程式碼的清晰。
它的原理就是建立一個Observable來完成非同步任務，組合使用各種不同的鏈式操作，來實現各種複雜的操作，最終將任務的執行結果發射給Observer進行處理。
當然，RxJava不僅適用於Android，也適用於伺服器端等各種場景。
我們總結以下RxJava的用途：
- 簡化非同步程式的流程；
- 使用近似於Java8的流的操作進行程式設計：因爲想要在Android中使用Java8的流程式設計有諸多的限制，所以我們可以使用RxJava來實現這個目的。
在使用RxJava之前，我們需要先在自己的專案中新增如下的依賴：
compile 'io.reactivex.rxjava2:rxjava:2.2.0'
compile 'io.reactivex.rxjava2:rxandroid:2.0.2'
這裏我們使用的是RxJava2，它與RxJava的第一個版本有些許不同。在本文中，我們所有的關於RxJava的範例都將基於RxJava2.
注：如果想瞭解關於Java8的流程式設計的內容的內容，可以參考我之前寫過的文章五分鐘學習Java8的流程式設計。
1.2 概要
下面 下麪是RxJava的一個基本的用例，這裏我們定義了一個Observable，然後在它內部使用emitter發射了一些數據和資訊（其實就相當於呼叫了被觀察物件內部的方法，通知所有的觀察者）。
然後，我們用Consumer介面的範例作爲subscribe()方法的參數來觀察發射的結果。（這裏的介面的方法都已經被使用Lambda簡化過，應該學着適應它。）
Observable<Integer> observable = Observable.create(emitter -> {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
});
observable.subscribe(System.out::println);
這樣，我們就完成了一個基本的RxJava的範例。從上面的例子中，你或許沒法看出Observable中隱藏的流的概念，看下面 下麪的例子：
Observable.range(0, 10).map(String::valueOf).forEach(System.out::println);
這裏我們先用Observable.range()方法產生一個序列，然後用map方法將該整數序列對映成一個字元序列，最後將得到的序列輸出來。從上面看出，這種操作和Java8裏面的Stream程式設計很像。但是兩者之間是有區別的：
- 所謂的「推」和「拉」的區別：Stream中是通過從流中讀取數據來實現鏈式操作，而RxJava除了Stream中的功能之外，還可以通過「發射」數據，來實現通知的功能，即RxJava在Stream之上又多了一個觀察者的功能。
- Java8中的Stream可以通過
parall()來實現並行，即基於分治演算法將任務分解並計算得到結果之後將結果合併起來；而RxJava只能通過subscribeOn()方法將所有的操作切換到某個執行緒中去。 - Stream只能被消費一次，但是
Observable可以被多次進行訂閱；
RxJava除了爲我們提供了Observable之外，在新的RxJava中還提供了適用於其他場景的基礎類，它們之間的功能和主要區別如下：
Flowable: 多個流，響應式流和背壓Observable: 多個流，無背壓Single: 只有一個元素或者錯誤的流Completable: 沒有任何元素，只有一個完成和錯誤信號的流Maybe: 沒有任何元素或者只有一個元素或者只有一個錯誤的流
除了上面的幾個基礎類之外，還有一個Disposable。當我們監聽某個流的時候，就能獲取到一個Disposable物件。它提供了兩個方法，一個是isDisposed，可以被用來判斷是否停止了觀察指定的流；另一個是dispose方法，用來放棄觀察指定的流，我們可以使用它在任意的時刻停止觀察操作。
1.3 總結
上面我們介紹了了關於RxJava的基本的概念和使用方式，在下面 下麪的文章中我們會按照以上定義的順序從API的角度來講解以下RxJava各個模組的使用方法。
2、RxJava 的使用
2.1 Observable
從上面的文章中我們可以得知，Observable和後面3種操作功能近似，區別在於Flowable加入了背壓的概念，Observable的大部分方法也適用於其他3個操作和Flowable。
因此，我們這裏先從Observable開始梳理，然後我們再專門對Flowable和背壓的進行介紹。
Observable爲我們提供了一些靜態的構造方法來建立一個Observable物件，還有許多鏈式的方法來完成各種複雜的功能。
這裏我們按照功能將它的這些方法分成各個類別並依次進行相關的說明。
2.1.1 建立操作
1.interval & intervalRange
下面 下麪的操作可以每個3秒的時間發送一個整數，整數從0開始：
Observable.interval(3, TimeUnit.SECONDS).subscribe(System.out::println);
如果想要設定從指定的數位開始也是可以的，實際上interval提供了許多過載方法供我們是使用。下面 下麪我們連同與之功能相近的intervalRange方法也一同給出：
public static Observable<Long> interval(long initialDelay, long period, TimeUnit unit, Scheduler scheduler)public static Observable<Long> interval(long period, TimeUnit unit, Scheduler scheduler)public static Observable<Long> intervalRange(long start, long count, long initialDelay, long period, TimeUnit unit, Scheduler scheduler)
這裏的initialDelay參數用來指示開始發射第一個整數的之前要停頓的時間，時間的單位與peroid一樣，都是通過unit參數來指定的；period參數用來表示每個發射之間停頓多少時間；unit表示時間的單位，是TimeUnit型別的；scheduler參數指定數據發射和等待時所在的執行緒。
intervalRange方法可以用來將發射的整數序列限制在一個範圍之內，這裏的start用來表示發射的數據的起始值，count表示總共要發射幾個數位，其他參數與上面的interval方法一致。
2.range & rangeLong
下面 下麪的操作可以產生一個從5開始的連續10個整數構成的序列：
Observable.range(5, 10).subscribe(i -> System.out.println("1: " + i));
該方法需要傳入兩個參數，與之有相同功能的方法還有rangeLong：
public static Observable<Integer> range(final int start, final int count)public static Observable<Long> rangeLong(long start, long count)
這裏的兩個參數start用來指定用於生成的序列的開始值，count用來指示要生成的序列總共包含多少個數字，上面的兩個方法的主要區別在於一個是用來生成int型整數的，一個是用來生成long型整數的。
3.create
create方法用於從頭開始建立一個Observable，像下面 下麪顯示的那樣，你需要使用create方法並傳一個發射器作爲參數，在該發射器內部呼叫onNext、onComplete和onError方法就可以將數據發送給監聽者。
Observable.create((ObservableOnSubscribe<Integer>) observableEmitter -> {
observableEmitter.onNext(1);
observableEmitter.onNext(2);
observableEmitter.onComplete();
}).subscribe(System.out::println);
4.defer
defer直到有觀察者訂閱時才建立Observable，並且爲每個觀察者建立一個新的Observable。defer操作符會一直等待直到有觀察者訂閱它，然後它使用Observable工廠方法生成一個Observable。比如下面 下麪的程式碼兩個訂閱輸出的結果是不一致的：
Observable<Long> observable = Observable.defer((Callable<ObservableSource<Long>>) () -> Observable.just(System.currentTimeMillis()));
observable.subscribe(System.out::print);
System.out.println();
observable.subscribe(System.out::print);
下面 下麪是該方法的定義，它接受一個Callable物件，可以在該物件中返回一個Observable的範例：
public static <T> Observable<T> defer(Callable<? extends ObservableSource<? extends T>> supplier)
5.empty & never & error
public static <T> Observable<T> empty()：建立一個不發射任何數據但是正常終止的Observable；public static <T> Observable<T> never()：建立一個不發射數據也不終止的Observable；public static <T> Observable<T> error(Throwable exception)：建立一個不發射數據以一個錯誤終止的Observable，它有幾個過載版本，這裏給出其中的一個。
測試程式碼：
Observable.empty().subscribe(i->System.out.print("next"),i->System.out.print("error"),()->System.out.print("complete"));
Observable.never().subscribe(i->System.out.print("next"),i->System.out.print("error"),()->System.out.print("complete"));
Observable.error(new Exception()).subscribe(i->System.out.print("next"),i->System.out.print("error"),()->System.out.print("complete"));
輸出結果：completeerror
6.from 系列
from系列的方法用來從指定的數據源中獲取一個Observable：
public static <T> Observable<T> fromArray(T... items)：從陣列中獲取；public static <T> Observable<T> fromCallable(Callable<? extends T> supplier)：從Callable中獲取；public static <T> Observable<T> fromFuture(Future<? extends T> future)：從Future中獲取，有多個過載版本，可以用來指定執行緒和超時等資訊；public static <T> Observable<T> fromIterable(Iterable<? extends T> source)：從Iterable中獲取；public static <T> Observable<T> fromPublisher(Publisher<? extends T> publisher)：從Publisher中獲取。
7.just 系列
just系列的方法的一個參數的版本爲下面 下麪的形式：public static <T> Observable<T> just(T item)，它還有許多個過載的版本，區別在於接受的參數的個數不同，最少1個，最多10個。
8.repeat
該方法用來表示指定的序列要發射多少次，下面 下麪的方法會將該序列無限次進行發送：
Observable.range(5, 10).repeat().subscribe(i -> System.out.println(i));
repeat方法有以下幾個相似方法：
public final Observable<T> repeat()public final Observable<T> repeat(long times)public final Observable<T> repeatUntil(BooleanSupplier stop)public final Observable<T> repeatWhen(Function<? super Observable<Object>, ? extends ObservableSource<?>> handler)
第1個無參的方法會無限次地發送指定的序列（實際上內部呼叫了第2個方法並傳入了Long.MAX_VALUE），第2個方法會將指定的序列重複發射指定的次數；第3個方法會在滿足指定的要求的時候停止重複發送，否則會一直髮送。
9.timer
timer操作符建立一個在給定的時間段之後返回一個特殊值的Observable，它在延遲一段給定的時間後發射一個簡單的數位0。比如下面 下麪的程式會在500毫秒之後輸出一個數字0。
Observable.timer(500, TimeUnit.MILLISECONDS).subscribe(System.out::print);
下面 下麪是該方法及其過載方法的定義，過載方法還可以指定一個排程器：
public static Observable<Long> timer(long delay, TimeUnit unit)public static Observable<Long> timer(long delay, TimeUnit unit, Scheduler scheduler)
2.1.2 變換操作
1.map & cast
map操作符對原始Observable發射的每一項數據應用一個你選擇的函數，然後返回一個發射這些結果的Observable。預設不在任何特定的排程器上執行。cast操作符將原始Observable發射的每一項數據都強制轉換爲一個指定的型別（多型），然後再發射數據，它是map的一個特殊版本：
下面 下麪的第一段程式碼用於將生成的整數序列轉換成一個字串序列之後並輸出；第二段程式碼用於將Date型別轉換成Object型別並進行輸出，這裏如果前面的Class無法轉換成第二個Class就會出現異常：
Observable.range(1, 5).map(String::valueOf).subscribe(System.out::println);
Observable.just(new Date()).cast(Object.class).subscribe(System.out::print);
這兩個方法的定義如下：
public final <R> Observable<R> map(Function<? super T, ? extends R> mapper)public final <U> Observable<U> cast(Class<U> clazz)
這裏的mapper函數接受兩個泛型，一個表示原始的數據型別，一個表示要轉換之後的數據型別，轉換的邏輯寫在該介面實現的方法中即可。
2.flatMap & contactMap
flatMap將一個發送事件的上遊Observable變換爲多個發送事件的Observables，然後將它們發射的事件合併後放進一個單獨的Observable裡。需要注意的是, flatMap並不保證事件的順序，也就是說轉換之後的Observables的順序不必與轉換之前的序列的順序一致。比如下面 下麪的程式碼用於將一個序列構成的整數轉換成多個單個的Observable，然後組成一個OBservable，並被訂閱。下面 下麪輸出的結果仍將是一個字串數位序列，只是順序不一定是增序的。
Observable.range(1, 5)
.flatMap((Function<Integer, ObservableSource<String>>) i -> Observable.just(String.valueOf(i)))
.subscribe(System.out::println);
與flatMap對應的方法是contactMap，後者能夠保證最終輸出的順序與上遊發送的順序一致。下面 下麪是這兩個方法的定義：
public final <R> Observable<R> flatMap(Function<? super T, ? extends ObservableSource<? extends R>> mapper)public final <R> Observable<R> concatMap(Function<? super T, ? extends ObservableSource<? extends R>> mapper)
flatMap的過載方法數量過多，它們在數據源方面略有不同，有的支援錯誤等可選參數，具體可以參考原始碼。
3.flatMapIterable
flatMapIterable可以用來將上流的任意一個元素轉換成一個Iterable物件，然後我們可以對其進行消費。在下面 下麪的程式碼中，我們先生成一個整數的序列，然後將每個整數對映成一個Iterable<string>型別，最後，我們對其進行訂閱和消費：
Observable.range(1, 5)
.flatMapIterable((Function<Integer, Iterable<String>>) integer -> Collections.singletonList(String.valueOf(integer)))
.subscribe(s -> System.out.println("flatMapIterable : " + s));
下面 下麪是該方法及其過載方法的定義：
public final <U> Observable<U> flatMapIterable(Function<? super T, ? extends Iterable<? extends U>> mapper)public final <U, V> Observable<V> flatMapIterable(Function<? super T, ? extends Iterable<? extends U>> mapper, BiFunction<? super T, ? super U, ? extends V> resultSelector)
4.buffer
該方法用於將整個流進行分組。以下面 下麪的程式爲例，我們會先生成一個7個整數構成的流，然後使用buffer之後，這些整數會被3個作爲一組進行輸出，所以當我們訂閱了buffer轉換之後的Observable之後得到的是一個列表構成的OBservable：
Observable.range(1, 7).buffer(3)
.subscribe(integers -> System.out.println(Arrays.toString(integers.toArray())));
下面 下麪是這個方法及其過載方法的定義，它的過載方法太多，這裏我們只給出其中的兩個，其他的可以參考RxJava的原始碼。這裏的buffer應該理解爲一個緩衝區，當緩衝區滿了或者剩餘的數據不夠一個緩衝區的時候就將數據發射出去。
public final Observable<List<T>> buffer(int count)public final Observable<List<T>> buffer(int count, int skip)- ...
5.groupBy
groupBy用於分組元素，它可以被用來根據指定的條件將元素分成若幹組。它將得到一個Observable<GroupedObservable<T, M>>型別的Observable。如下面 下麪的程式所示，這裏我們使用concat方法先將兩個Observable拼接成一個Observable，然後對其元素進行分組。這裏我們的分組依據是整數的值，這樣我們將得到一個Observable<GroupedObservable<Integer, Integer>>型別的Observable。然後，我們再將得到的序列拼接成一個並進行訂閱輸出：
Observable<GroupedObservable<Integer, Integer>> observable = Observable.concat(
Observable.range(1,4), Observable.range(1,6)).groupBy(integer -> integer);
Observable.concat(observable).subscribe(integer -> System.out.println("groupBy : " + integer));
該方法有多個過載版本，這裏我們用到的一個的定義是：
public final <K> Observable<GroupedObservable<K, T>> groupBy(Function<? super T, ? extends K> keySelector)
6.scan
scan操作符對原始Observable發射的第一項數據應用一個函數，然後將那個函數的結果作爲自己的第一項數據發射。它將函數的結果同第二項數據一起填充給這個函數來產生它自己的第二項數據。它持續進行這個過程來產生剩餘的數據序列。這個操作符在某些情況下被叫做accumulator。
以下面 下麪的程式爲例，該程式的輸結果是2 6 24 120 720，可以看出這裏的計算規則是，我們把傳入到scan中的函數記爲f，序列記爲x，生成的序列記爲y，那麼這裏的計算公式是y(0)=x(0); y(i)=f(y(i-1), x(i)), i>0：
Observable.range(2, 5).scan((i1, i2) -> i1 * i2).subscribe(i -> System.out.print(i + " "));
除了上面的這種形式，scan方法還有一個過載的版本，我們可以使用這個版本的方法來在生成序列的時候指定一個初始值。以下面 下麪的程式爲例，它的輸出結果是3 6 18 72 360 2160，可以看出它的輸出比上面的形式多了1個，這是因爲當指定了初始值之後，生成的第一個數位就是那個初始值，剩下的按照我們上面的規則進行的。所以，用同樣的函數語言來描述的話，那麼它就應該是下面 下麪的這種形式：y(0)=initialValue; y(i)=f(y(i-1), x(i)), i>0。
Observable.range(2, 5).scan(3, (i1, i2) -> i1 * i2).subscribe(i -> System.out.print(i + " "));
以上方法的定義是：
public final Observable<T> scan(BiFunction<T, T, T> accumulator)public final <R> Observable<R> scan(R initialValue, BiFunction<R, ? super T, R> accumulator)
7.window
windowWindow和Buffer類似，但不是發射來自原始Observable的數據包，它發射的是Observable，這些Observables中的每一個都發射原始Observable數據的一個子集，最後發射一個onCompleted通知。
以下面 下麪的程式爲例，這裏我們首先生成了一個由10個數字組成的整數序列，然後使用window函數將它們每3個作爲一組，每組會返回一個對應的Observable物件。
這裏我們對該返回的結果進行訂閱並進行消費，因爲10個數字，所以會被分成4個組，每個對應一個Observable：
Observable.range(1, 10).window(3).subscribe(
observable -> observable.subscribe(integer -> System.out.println(observable.hashCode() + " : " + integer)));
除了對數據包進行分組，我們還可以根據時間來對發射的數據進行分組。該方法有多個過載的版本，這裏我們給出其中的比較具有代表性的幾個：
public final Observable<Observable<T>> window(long count)public final Observable<Observable<T>> window(long timespan, long timeskip, TimeUnit unit)public final <B> Observable<Observable<T>> window(ObservableSource<B> boundary)public final <B> Observable<Observable<T>> window(Callable<? extends ObservableSource<B>> boundary)
2.1.3 過濾操作
1.filter
filter用來根據指定的規則對源進行過濾，比如下面 下麪的程式用來過濾整數1到10中所有大於5的數位：
Observable.range(1,10).filter(i -> i > 5).subscribe(System.out::println);
下面 下麪是該方法的定義：
public final Observable<T> filter(Predicate<? super T> predicate)
2.elementAt & firstElement & lastElement
elementAt用來獲取源中指定位置的數據，它有幾個過載方法，這裏我們介紹一下最簡單的一個方法的用法。下面 下麪是elementAt的一個範例，它將獲取源數據中索引爲1的元素並交給觀察者訂閱。下面 下麪的程式將輸出1
Observable.range(1, 10).elementAt(0).subscribe(System.out::print);
這裏我們給出elementAt及其相關的方法的定義，它們的使用相似。注意一下這裏的返回型別：
public final Maybe<T> elementAt(long index)public final Single<T> elementAt(long index, T defaultItem)public final Single<T> elementAtOrError(long index)
除了獲取指定索引的元素的方法之外，RxJava中還有可以用來直接獲取第一個和最後一個元素的方法，這裏我們直接給出方法的定義：
public final Maybe<T> firstElement()public final Single<T> first(T defaultItem)public final Single<T> firstOrError()public final Maybe<T> lastElement()public final Single<T> last(T defaultItem)public final Single<T> lastOrError()
3.distinct & distinctUntilChanged
distinct用來對源中的數據進行過濾，以下面 下麪的程式爲例，這裏會把重複的數位7過濾掉：
Observable.just(1,2,3,4,5,6,7,7).distinct().subscribe(System.out::print);
與之類似的還有distinctUntilChanged方法，與distinct不同的是，它只當相鄰的兩個元素相同的時候纔會將它們過濾掉。比如下面 下麪的程式會過濾掉其中的2和5，所以最終的輸出結果是12345676：
Observable.just(1,2,2,3,4,5,5,6,7,6).distinctUntilChanged().subscribe(System.out::print);
該方法也有幾個功能相似的方法，這裏給出它們的定義如下：
public final Observable<T> distinct()public final <K> Observable<T> distinct(Function<? super T, K> keySelector)public final <K> Observable<T> distinct(Function<? super T, K> keySelector, Callable<? extends Collection<? super K>> collectionSupplier)public final Observable<T> distinctUntilChanged()public final <K> Observable<T> distinctUntilChanged(Function<? super T, K> keySelector)public final Observable<T> distinctUntilChanged(BiPredicate<? super T, ? super T> comparer)
4.skip & skipLast & skipUntil & skipWhile
skip方法用於過濾掉數據的前n項，比如下面 下麪的程式將會過濾掉前2項，因此輸出結果是345：
Observable.range(1, 5).skip(2).subscribe(System.out::print);
與skip方法對應的是take方法，它用來表示只選擇數據源的前n項，該方法的範例就不給出了。這裏，我們說一下與之類功能類似的過載方法。skip還有一個過載方法接受兩個參數，用來表示跳過指定的時間，也就是在指定的時間之後纔開始進行訂閱和消費。下面 下麪的程式會在3秒之後纔開始不斷地輸出數位：
Observable.range(1,5).repeat().skip(3, TimeUnit.SECONDS).subscribe(System.out::print);
與skip功能相反的方法的還有skipLast，它用來表示過濾掉後面的幾項，以及最後的一段時間不進行發射等。比如下面 下麪的方法，我們會在程式開始之前進行計時，然後會不斷重複輸出數位，直到5秒之後結束。然後，我們用skipLast方法表示最後的2秒不再進行發射。所以下面 下麪的程式會先不斷輸出數位3秒，3秒結束後停止輸出，並在2秒之後結束程式：
long current = System.currentTimeMillis();
Observable.range(1,5)
.repeatUntil(() -> System.currentTimeMillis() - current > TimeUnit.SECONDS.toMillis(5))
.skipLast(2, TimeUnit.SECONDS).subscribe(System.out::print);
與上面的這些方法類似的還有一些，這裏我們不再一一列舉。因爲這些方法的過載方法比較多，下面 下麪我們給出其中的具有代表性的一部分：
public final Observable<T> skip(long count)public final Observable<T> skip(long time, TimeUnit unit, Scheduler scheduler)public final Observable<T> skipLast(int count)public final Observable<T> skipLast(long time, TimeUnit unit, Scheduler scheduler, boolean delayError, int bufferSize)public final <U> Observable<T> skipUntil(ObservableSource<U> other)public final Observable<T> skipWhile(Predicate<? super T> predicate)
5.take & takeLast & takeUntil & takeWhile
與skip方法對應的是take方法，它表示按照某種規則進行選擇操作。我們以下面 下麪的程式爲例，這裏第一段程式表示只發射序列中的前2個數據：
Observable.range(1, 5).take(2).subscribe(System.out::print);
下面 下麪的程式表示只選擇最後2秒中輸出的數據：
long current = System.currentTimeMillis();
Observable.range(1,5)
.repeatUntil(() -> System.currentTimeMillis() - current > TimeUnit.SECONDS.toMillis(5))
.takeLast(2, TimeUnit.SECONDS).subscribe(System.out::print);
下面 下麪是以上相關的方法的定義，同樣的，我們只選擇其中比較有代表性的幾個：
public final Observable<T> take(long count)public final Observable<T> takeLast(long count, long time, TimeUnit unit, Scheduler scheduler, boolean delayError, int bufferSize)public final <U> Observable<T> takeUntil(ObservableSource<U> other)public final Observable<T> takeUntil(Predicate<? super T> stopPredicate)public final Observable<T> takeWhile(Predicate<? super T> predicate)
6.ignoreElements
該方法用來過濾所有源Observable產生的結果，只會把Observable的onComplete和onError事件通知給訂閱者。下面 下麪是該方法的定義：
public final Completable ignoreElements()
7.throttleFirst & throttleLast & throttleLatest & throttleWithTimeout
這些方法用來對輸出的數據進行限制，它們是通過時間的」視窗「來進行限制的，你可以理解成按照指定的參數對時間進行分片，然後根據各個方法的要求選擇第一個、最後一個、最近的等進行發射。下面 下麪是throttleLast方法的用法範例，它會輸出每個500毫秒之間的數位中最後一個數字：
Observable.interval(80, TimeUnit.MILLISECONDS)
.throttleLast(500, TimeUnit.MILLISECONDS)
.subscribe(i -> System.out.print(i + " "));
其他的幾個方法的功能大致列舉如下：
throttleFirst只會發射指定的Observable在指定的事件範圍內發射出來的第一個數據；throttleLast只會發射指定的Observable在指定的事件範圍內發射出來的最後一個數據；throttleLatest用來發射距離指定的時間分片最近的那個數據;throttleWithTimeout僅在過了一段指定的時間還沒發射數據時才發射一個數據，如果在一個時間片達到之前，發射的數據之後又緊跟着發射了一個數據，那麼這個時間片之內之前發射的數據會被丟掉，該方法底層是使用debounce方法實現的。如果數據發射的頻率總是快過這裏的timeout參數指定的時間，那麼將不會再發射出數據來。
下面 下麪是這些方法及其過載方法的定義（選擇其中一部分）：
public final Observable<T> throttleFirst(long skipDuration, TimeUnit unit, Scheduler scheduler)public final Observable<T> throttleLast(long intervalDuration, TimeUnit unit, Scheduler scheduler)public final Observable<T> throttleLatest(long timeout, TimeUnit unit, Scheduler scheduler, boolean emitLast)public final Observable<T> throttleWithTimeout(long timeout, TimeUnit unit, Scheduler scheduler)
8.debounce
debounce也是用來限制發射頻率過快的，它僅在過了一段指定的時間還沒發射數據時才發射一個數據。我們通過下面 下麪的圖來說明這個問題：
debounce
這裏紅、綠、藍三個球發射出來的原因都是因爲當反射了這個球之後的一定的時間內沒有其他的球發射出來，這個時間是我們可以通過參數來指定的。
該方法的用法與throttle之類的方法類似，上面也說過throttle那些方法底層用了debounce實現，所以，這裏我們不再爲該方法專門編寫相關的測試程式碼。
9.sample
實際上throttleLast的實現中內部呼叫的就是sample。
2.1.4 組合操作
1.startWith & startWithArray
startWith方法可以用來在指定的數據源的之前插入幾個數據，它的功能類似的方法有startWithArray，另外還有幾個過載方法。這裏我們給出一個基本的用法範例，下面 下麪的程式會在原始的數位流1-5的前面加上0，所以最終的輸出結果是012345：
Observable.range(1,5).startWith(0).subscribe(System.out::print);
下面 下麪是startWith及其幾個功能相關的方法的定義：
public final Observable<T> startWith(Iterable<? extends T> items)public final Observable<T> startWith(ObservableSource<? extends T> other)public final Observable<T> startWith(T item)public final Observable<T> startWithArray(T... items)
2.merge & mergeArray
merge可以讓多個數據源的數據合併起來進行發射，當然它可能會讓merge之後的數據交錯發射。下面 下麪是一個範例，這個例子中，我們使用merge方法將兩個Observable合併到了一起進行監聽：
Observable.merge(Observable.range(1,5), Observable.range(6,5)).subscribe(System.out::print);
鑑於merge方法及其功能類似的方法太多，我們這裏挑選幾個比較有代表性的方法，具體的可以檢視RxJava的原始碼：
public static <T> Observable<T> merge(Iterable<? extends ObservableSource<? extends T>> sources)public static <T> Observable<T> mergeArray(ObservableSource<? extends T>... sources)public static <T> Observable<T> mergeDelayError(Iterable<? extends ObservableSource<? extends T>> sources)public static <T> Observable<T> mergeArrayDelayError(ObservableSource<? extends T>... sources)
這裏的mergeError方法與merge方法的表現一致，只是在處理由onError觸發的錯誤的時候有所不同。mergeError方法會等待所有的數據發射完畢之後才把錯誤發射出來，即使多個錯誤被觸發，該方法也只會發射出一個錯誤資訊。而如果使用merger方法，那麼當有錯誤被觸發的時候，該錯誤會直接被拋出來，並結束髮射操作。下面 下麪是該方法的一個使用的範例，這裏我們主執行緒停頓4秒，然後所有merge的Observable中的一個會線上程開始的第2秒的時候觸發一個錯誤，該錯誤最終會在所有的數據發射完畢之後被髮射出來：
Observable.mergeDelayError(Observable.range(1,5),
Observable.range(1,5).repeat(2),
Observable.create((ObservableOnSubscribe<String>) observableEmitter -> {
Thread.sleep(2000);
observableEmitter.onError(new Exception("error"));
})
).subscribe(System.out::print, System.out::print);
Thread.sleep(4000);
3.concat & concatArray & concatEager
該方法也是用來將多個Observable拼接起來，但是它會嚴格按照傳入的Observable的順序進行發射，一個Observable沒有發射完畢之前不會發射另一個Observable裏面的數據。下面 下麪是一個程式範例，這裏傳入了兩個Observable，會按照順序輸出12345678910：
Observable.concat(Observable.range(1, 5), Observable.range(6, 5)).subscribe(System.out::print);
下面 下麪是該方法的定義，鑑於該方法及其過載方法太多，這裏我們選擇幾個比較有代表性的說明：
public static <T> Observable<T> concat(Iterable<? extends ObservableSource<? extends T>> sources)public static <T> Observable<T> concatDelayError(Iterable<? extends ObservableSource<? extends T>> sources)public static <T> Observable<T> concatArray(ObservableSource<? extends T>... sources)public static <T> Observable<T> concatArrayDelayError(ObservableSource<? extends T>... sources)public static <T> Observable<T> concatEager(ObservableSource<? extends ObservableSource<? extends T>> sources)public static <T> Observable<T> concatArrayEager(ObservableSource<? extends T>... sources)
對於concat方法，我們之前已經介紹過它的用法；這裏的conactArray的功能與之類似；對於concatEager方法，當一個觀察者訂閱了它的結果，那麼就相當於訂閱了它拼接的所有ObservableSource，並且會先快取這些ObservableSource發射的數據，然後再按照順序將它們發射出來。而對於這裏的concatDelayError方法的作用和前面的mergeDelayError類似，只有當所有的數據都發射完畢纔會處理異常。
4.zip & zipArray & zipIterable
zip操作用來將多個數據項進行合併，可以通過一個函數指定這些數據項的合併規則。比如下面 下麪的程式的輸出結果是6 14 24 36 50，顯然這裏的合併的規則是相同索引的兩個數據的乘積。不過仔細看下這裏的輸出結果，可以看出，如果一個數據項指定的位置沒有對應的值的時候，它是不會參與這個變換過程的：
Observable.zip(Observable.range(1, 6), Observable.range(6, 5), (integer, integer2) -> integer * integer2)
.subscribe(i -> System.out.print(i + " "));
zip方法有多個過載的版本，同時也有功能近似的方法，這裏我們挑選有代表性的幾個進行說明：
public static <T, R> Observable<R> zip(Iterable<? extends ObservableSource<? extends T>> sources, Function<? super Object[], ? extends R> zipper)ublic static <T, R> Observable<R> zipArray(Function<? super Object[], ? extends R> zipper, boolean delayError, int bufferSize, ObservableSource... sources)public static <T, R> Observable<R> zipIterable(Iterable<? extends ObservableSource<? extends T>> sources, Function<? super Object[], ? extends R> zipper, boolean delayError, int bufferSize)
實際上上面幾個方法的用法和功能基本類似，區別在於傳入的ObservableSource的參數的形式。
5.combineLastest
與zip操作類似，但是這個操作的輸出結果與zip截然不同，以下面 下麪的程式爲例，它的輸出結果是36 42 48 54 60：
Observable.combineLatest(Observable.range(1, 6), Observable.range(6, 5), (integer, integer2) -> integer * integer2)
.subscribe(i -> System.out.print(i + " "));
利用下面 下麪的這張圖可以比較容易來說明這個問題：
[圖片上傳失敗...(image-5b184-1534261018262)]
上圖中的上面的兩條橫線代表用於拼接的兩個數據項，下面 下麪的一條橫線是拼接之後的結果。combineLatest的作用是拼接最新發射的兩個數據。下面 下麪我們用上圖的過程來說明該方法是如何執行的：開始第一條只有1的時候無法拼接，；當第二條出現A的時候，此時最新的數據是1和A，故組合成一個1A；第二個數據項發射了B，此時最新的數據是1和B，故組合成1B；第一條橫線發射了2，此時最新的數據是2和B，因此得到了2B，依次類推。然後再回到我們上面的問題，第一個數據項連續發射了5個數據的時候，第二個數據項一個都沒有發射出來，因此沒有任何輸出；然後第二個數據項開始發射數據，當第二個數據項發射了6的時候，此時最新的數據組合是6和6，故得36；然後，第二個數據項發射了7，此時最新的數據組合是6和7，故得42，依次類推。
該方法也有對應的combineLatestDelayError方法，用途也是隻有當所有的數據都發射完畢的時候纔去處理錯誤邏輯。
2.1.5 輔助操作
1.delay
delay方法用於在發射數據之前停頓指定的時間，比如下面 下麪的程式會在真正地發射數據之前停頓1秒：
Observable.range(1, 5).delay(1000, TimeUnit.MILLISECONDS).subscribe(System.out::print);
Thread.sleep(1500);
同樣delay方法也有幾個過載的方法，可以供我們用來指定觸發的執行緒等資訊，這裏給出其中的兩個，其他的可以參考原始碼和文件：
public final Observable<T> delay(long delay, TimeUnit unit)public final Observable<T> delay(long delay, TimeUnit unit, Scheduler scheduler)
2.do系列
RxJava中還有一系列的方法可以供我們使用，它們共同的特點是都是以do開頭，下面 下麪我們列舉一下這些方法並簡要說明一下它們各自的用途：
public final Observable<T> doAfterNext(Consumer<? super T> onAfterNext)，會在onNext方法之後觸發；public final Observable<T> doAfterTerminate(Action onFinally)，會在Observable終止之後觸發；public final Observable<T> doFinally(Action onFinally)，當onComplete或者onError的時候觸發；public final Observable<T> doOnDispose(Action onDispose)，當被dispose的時候觸發；public final Observable<T> doOnComplete(Action onComplete)，當complete的時候觸發；public final Observable<T> doOnEach(final Observer<? super T> observer)，當每個onNext呼叫的時候觸發；public final Observable<T> doOnError(Consumer<? super Throwable> onError)，當呼叫onError的時候觸發；public final Observable<T> doOnLifecycle(final Consumer<? super Disposable> onSubscribe, final Action onDispose)public final Observable<T> doOnNext(Consumer<? super T> onNext)，，會在onNext的時候觸發；public final Observable<T> doOnSubscribe(Consumer<? super Disposable> onSubscribe)，會在訂閱的時候觸發；public final Observable<T> doOnTerminate(final Action onTerminate)，當終止之前觸發。
這些方法可以看作是對操作執行過程的一個監聽，當指定的操作被觸發的時候會同時觸發這些監聽方法：
Observable.range(1, 5)
.doOnEach(integerNotification -> System.out.println("Each : " + integerNotification.getValue()))
.doOnComplete(() -> System.out.println("complete"))
.doFinally(() -> System.out.println("finally"))
.doAfterNext(i -> System.out.println("after next : " + i))
.doOnSubscribe(disposable -> System.out.println("subscribe"))
.doOnTerminate(() -> System.out.println("terminal"))
.subscribe(i -> System.out.println("subscribe : " + i));
3.subscribeOn & observeOn
subscribeOn用於指定Observable自身執行的執行緒，observeOn用於指定發射數據所處的執行緒，比如Android中的非同步任務需要用observeOn指定發射數據所在的執行緒是非主執行緒，然後執行完畢之後將結果發送給主執行緒，就需要用subscribeOn來指定。比如下面 下麪的程式，我們用這兩個方法來指定所在的執行緒：
Observable.create((ObservableOnSubscribe<Integer>) observableEmitter -> {
System.out.println(Thread.currentThread());
observableEmitter.onNext(0);
}).observeOn(Schedulers.newThread()).subscribeOn(Schedulers.computation())
.subscribe(integer -> System.out.println(Thread.currentThread()));
最終的輸出結果如下所示：
Thread[RxComputationThreadPool-1,5,main]
Thread[RxNewThreadScheduler-1,5,main]
4.timeout
用來設定一個超時時間，如果指定的時間之內沒有任何數據被髮射出來，那麼就會執行我們指定的數據項。如下面 下麪的程式所示，我們先爲設定了一個間隔200毫秒的數位產生器，開始發射數據之前要停頓1秒鐘，因爲我們設定的超時時間是500毫秒，因而在第500毫秒的時候會執行我們傳入的數據項：
Observable.interval(1000, 200, TimeUnit.MILLISECONDS)
.timeout(500, TimeUnit.MILLISECONDS, Observable.rangeLong(1, 5))
.subscribe(System.out::print);
Thread.sleep(2000);
timeout方法有多個過載方法，可以爲其指定執行緒等參數，可以參考原始碼或者文件瞭解詳情。
2.1.6 錯誤處理操作符
錯誤處理操作符主要用來提供給Observable，用來對錯誤資訊做統一的處理，常用的兩個是catch和retry。
1.catch
catch操作會攔截原始的Observable的onError通知，將它替換爲其他數據項或者數據序列，讓產生的Observable能夠正常終止或者根本不終止。在RxJava中該操作有3終型別：
onErrorReturn：這種操作會在onError觸發的時候返回一個特殊的項替換錯誤，並呼叫觀察者的onCompleted方法，而不會將錯誤傳遞給觀察者；onErrorResumeNext：會在onError觸發的時候發射備用的數據項給觀察者；onExceptionResumeNext：如果onError觸發的時候onError收到的Throwable不是Exception，它會將錯誤傳遞給觀察者的onError方法，不會使用備用的Observable。
下面 下麪是onErrorReturn和onErrorResumeNext的程式範例，這裏第一段程式碼會在出現錯誤的時候輸出666，而第二段會在出現錯誤的時候發射數位12345：
Observable.create((ObservableOnSubscribe<Integer>) observableEmitter -> {
observableEmitter.onError(null);
observableEmitter.onNext(0);
}).onErrorReturn(throwable -> 666).subscribe(System.out::print);
Observable.create((ObservableOnSubscribe<Integer>) observableEmitter -> {
observableEmitter.onError(null);
observableEmitter.onNext(0);
}).onErrorResumeNext(Observable.range(1,5)).subscribe(System.out::print);
2.retry
retry使用了一種錯誤重試機制 機製，它可以在出現錯誤的時候進行重試，我們可以通過參數指定重試機制 機製的條件。以下面 下麪的程式爲例，這裏我們設定了當出現錯誤的時候會進行2次重試，因此，第一次的時候出現錯誤會呼叫onNext，重試2次又會呼叫2次onNext，第二次重試的時候因爲重試又出現了錯誤，因此此時會觸發onError方法。也就是說，下面 下麪這段程式碼會觸發onNext3次，觸發onError()1次：
Observable.create(((ObservableOnSubscribe<Integer>) emitter -> {
emitter.onNext(0);
emitter.onError(new Throwable("Error1"));
emitter.onError(new Throwable("Error2"));
})).retry(2).subscribe(i -> System.out.println("onNext : " + i), error -> System.out.print("onError : " + error));
retry有幾個過載的方法和功能相近的方法，下面 下麪是這些方法的定義（選取部分）：
public final Observable<T> retry()：會進行無限次地重試；public final Observable<T> retry(BiPredicate<? super Integer, ? super Throwable> predicate)public final Observable<T> retry(long times)：指定重試次數；public final Observable<T> retry(long times, Predicate<? super Throwable> predicate)public final Observable<T> retryUntil(final BooleanSupplier stop)public final Observable<T> retryWhen(Function<? super Observable<Throwable>, ? extends ObservableSource<?>> handler)
2.1.7 條件操作符和布爾操作符
1.all & any
all用來判斷指定的數據項是否全部滿足指定的要求，這裏的「要求」可以使用一個函數來指定；any用來判斷指定的Observable是否存在滿足指定要求的數據項。
在下面 下麪的程式中，我們用該函數來判斷指定的數據項是否全部滿足大於5的要求，顯然是不滿足的，因此下面 下麪的程式將會輸出false：
Observable.range(5, 5).all(i -> i>5).subscribe(System.out::println); // false
Observable.range(5, 5).any(i -> i>5).subscribe(System.out::println); // true
以下是該方法的定義：
public final Single<Boolean> all(Predicate<? super T> predicate)public final Single<Boolean> any(Predicate<? super T> predicate)
2.contains & isEmpty
這兩個方法分別用來判斷數據項中是否包含我們指定的數據項，已經判斷數據項是否爲空：
Observable.range(5, 5).contains(4).subscribe(System.out::println); // false
Observable.range(5, 5).isEmpty().subscribe(System.out::println); // false
以下是這兩個方法的定義：
public final Single<Boolean> isEmpty()public final Single<Boolean> contains(final Object element)
3.sequenceEqual
sequenceEqual用來判斷兩個Observable發射出的序列是否是相等的。比如下面 下麪的方法用來判斷兩個序列是否相等：
Observable.sequenceEqual(Observable.range(1,5), Observable.range(1, 5)).subscribe(System.out::println);
4.amb
amb作用的兩個或多個Observable，但是隻會發射最先發射數據的那個Observable的全部數據：
Observable.amb(Arrays.asList(Observable.range(1, 5), Observable.range(6, 5))).subscribe(System.out::print)
該方法及其功能近似的方法的定義，這裏前兩個是靜態的方法，第二個屬於實體方法：
public static <T> Observable<T> amb(Iterable<? extends ObservableSource<? extends T>> sources)public static <T> Observable<T> ambArray(ObservableSource<? extends T>... sources)public final Observable<T> ambWith(ObservableSource<? extends T> other)
5.defaultIfEmpty
defaultIfEmpty用來當指定的序列爲空的時候指定一個用於發射的值。下面 下麪的程式中，我們直接呼叫發射器的onComplete方法，因此序列是空的，結果輸出一個整數6：
Observable.create((ObservableOnSubscribe<Integer>) Emitter::onComplete).defaultIfEmpty(6).subscribe(System.out::print);
下面 下麪是該方法的定義：
public final Observable<T> defaultIfEmpty(T defaultItem)
2.1.8 轉換操作符
1.toList & toSortedList
toList和toSortedList用於將序列轉換成列表，後者相對於前者增加了排序的功能：
Observable.range(1, 5).toList().subscribe(System.out::println);
Observable.range(1, 5).toSortedList(Comparator.comparingInt(o -> -o)).subscribe(System.out::println);
下面 下麪是它們的定義，它們有多個過載版本，這裏選擇其中的兩個進行說明：
public final Single<List<T>> toList()public final Single<List<T>> toSortedList(final Comparator<? super T> comparator)
注意一下，這裏的返回結果是Single型別的，不過這並不妨礙我們繼續使用鏈式操作，因爲Single的方法和Observable基本一致。
另外還要注意這裏的Single中的參數是一個List<T>，也就是說，它把整個序列轉換成了一個列表物件。因此，上面的兩個範例程式的輸出是：
[1, 2, 3, 4, 5]
[5, 4, 3, 2, 1]
2.toMap & toMultimap
toMap用於將發射的數據轉換成另一個型別的值，它的轉換過程是針對每一個數據項的。以下面 下麪的程式碼爲例，它會將原始的序列中的每個數位轉換成對應的十六進制。但是，toMap轉換的結果不一定是按照原始的序列的發射的順序來的：
Observable.range(8, 10).toMap(Integer::toHexString).subscribe(System.out::print);
與toMap近似的是toMultimap方法，它可以將原始序列的每個數據項轉換成一個集合型別：
Observable.range(8, 10).toMultimap(Integer::toHexString).subscribe(System.out::print);
上面的兩段程式的輸出結果是：
{11=17, a=10, b=11, c=12, d=13, e=14, f=15, 8=8, 9=9, 10=16}
{11=[17], a=[10], b=[11], c=[12], d=[13], e=[14], f=[15], 8=[8], 9=[9], 10=[16]}
上面的兩個方法的定義是（多個過載，選擇部分）：
public final <K> Single<Map<K, T>> toMap(final Function<? super T, ? extends K> keySelector)public final <K> Single<Map<K, Collection<T>>> toMultimap(Function<? super T, ? extends K> keySelector)
3.toFlowable
該方法用於將一個Observable轉換成Flowable型別，下面 下麪是該方法的定義，顯然這個方法使用了策略模式，這裏面涉及背壓相關的內容，我們後續再詳細介紹。
public final Flowable<T> toFlowable(BackpressureStrategy strategy)
4.to
相比於上面的方法，to方法的限制更加得寬泛，你可以將指定的Observable轉換成任意你想要的型別（如果你可以做到的話），下面 下麪是一個範例程式碼，用來將指定的整數序列轉換成另一個整數型別的Observable，只不過這裏的每個數據項都是原來的列表中的數據總數的值：
Observable.range(1, 5).to(Observable::count).subscribe(System.out::println);
下面 下麪是該方法的定義：
public final <R> R to(Function<? super Observable<T>, R> converter)
2.2 執行緒控制
之前有提到過RxJava的執行緒控制是通過subscribeOn和observeOn兩個方法來完成的。
這裏我們梳理一下RxJava提供的幾種執行緒排程器以及RxAndroid爲Android提供的排程器的使用場景和區別等。
Schedulers.io()：代表適用於io操作的排程器，增長或縮減來自適應的執行緒池，通常用於網路、讀寫檔案等io密集型的操作。重點需要注意的是執行緒池是無限制的，大量的I/O排程操作將建立許多個執行緒並佔用記憶體。Schedulers.computation()：計算工作預設的排程器，代表CPU計算密集型的操作，與I/O操作無關。它也是許多RxJava方法，比如buffer(),debounce(),delay(),interval(),sample(),skip()，的預設排程器。Schedulers.newThread()：代表一個常規的新執行緒。Schedulers.immediate()：這個排程器允許你立即在當前執行緒執行你指定的工作。它是timeout(),timeInterval()以及timestamp()方法預設的排程器。Schedulers.trampoline()：當我們想在當前執行緒執行一個任務時，並不是立即，我們可以用trampoline()將它入隊。這個排程器將會處理它的佇列並且按序執行佇列中每一個任務。它是repeat()和retry()方法預設的排程器。
以及RxAndroid提供的執行緒排程器：
AndroidSchedulers.mainThread()用來指代Android的主執行緒
2.3 總結
上面的這些操作也基本適用於Flowable、Single、Completable和Maybe。
我們花費了很多的時間和精力來梳理了這些方法，按照上面的內容，使用RxJava實現一些基本的或者高階的操作都不是什麼問題。
但是，Observable更適用於處理一些數據規模較小的問題，當數據規模比較多的時候可能會出現MissingBackpressureException異常。
因此，我們還需要瞭解背壓和Flowable的相關內容才能 纔能更好地理解和應用RxJava.