Android RxJava操作符一覽

前言

把現在接觸到的操作符全部整理進來，方便查閱，遇到新的也會添加進來。和JavaLearn">RxJavaLearn 的README.md同步更新。

操作符決策樹

直接創建一個Observable（創建操作） 組合多個Observable（組合操作） 對Observable發射的數據執行變換操作（變換操作） 從Observable發射的數據中取特定的值（過濾操作） 轉發Observable的部分值（條件/布爾/過濾操作） 對Observable發射的數據序列求值（算術/聚合操作）

創建操作

用於創建Observable的操作符

Create
通過調用觀察者的方法從頭創建一個Observable

Empty/Never/Throw
創建行為受限的特殊Observable

Defer
在觀察者訂閱之前不創建這個Observable，為每一個觀察者創建一個新的Observable

Just
將對象或者對象集合轉換為一個會發射這些對象的Observable

From
將其它的對象或數據結構轉換為Observable

Range
創建發射指定范圍的整數序列的Observable,range操作符,發射從start開始的count個數

Interval
間隔一定時間發送一個數字,從0開始.本身運行在Schedulers.computation() 線程內

Repeat
Repeat作用在Observable上,會對其重復發射count次

Timer
Timer會在指定時間後發射一個數字0，注意其也是運行在computation Scheduler

變換操作

對Observable發射的數據進行變，能將數據轉化為我們想要的格式，

Buffer
緩存，可以簡單的理解為緩存，它定期從Observable收集數據到一個集合，然後把這些數據集合打包發射，而不是一次發射一個

FlatMap
扁平映射，將Observable發射的數據變換為Observables集合，然後將這些Observable發射的數據平坦化的放進一個單獨的Observable，可以認為是一個將嵌套的數據結構展開的過程。

GroupBy
分組，將原來的Observable分拆為Observable集合，將原始Observable發射的數據按Key分組，每一個Observable發射一組不同的數據

Map
映射，通過對序列的每一項都應用一個函數變換Observable發射的數據，實質是對序列中的每一項執行一個函數，函數的參數就是這個數據項

Scan
掃描， 連續地對數據序列的每一項應用一個函數，然後連續發射結果,每一項結果基於之前的結果.累加器函數.

Window
窗口，定期將來自Observable的數據分拆成一些Observable窗口，然後發射這些窗口，而不是每次發射一項。類似於Buffer，但Buffer發射的是數據，Window發射的是Observable，每一個Observable發射原始Observable的數據的一個子集

toList
Returns an Observable that emits a single item, a list composed of all the items emitted by the source Observable. 將一個Observable轉換為一個List.

過濾操作

濾掉數據集合中我們不想要的數據。用於從Observable發射的數據中進行選擇。

throttleWithTimeout
源Observable每次發射出來一個數據後就會進行計時,如果在設定好的時間結束前源Observable有新的數據發射出來，這個數據就會被丟棄，同時重新開始計時。

Debounce
只有在空閒了一段時間後才發射數據，通俗的說，就是如果一段時間沒有操作，就執行一次操作.還可以根據一個函數來進行限流。這個函數的返回值是一個臨時Observable，
如果源Observable在發射一個新的數據的時候，上一個數據根據函數所生成的臨時Observable還沒有結束,沒有調用onComplete，那麼上一個數據就會被過濾掉。如果是最後一個,還是會發射.

Distinct
去重，過濾掉所有的重復數據項

DistinctUtilChanged
過濾掉相鄰的重復項

ElementAt
取值，取特定位置的數據項

Filter
過濾，過濾掉沒有通過謂詞測試的數據項，只發射通過測試的

First
取滿足條件的第一個,如無滿足條件數據拋出異常.可以使用takeFisrt(),僅會調動onComplete.
有null數據時會拋空指針異常,要判空處理.
只取滿足條件的第一個數據.可以和BlockingObservable連用。可以Observable.toBlocking或者BlockingObservable.from方法來將一個Observable對象轉化為BlockingObservable對象

Last
末項，只發射最後一條數據.

IgnoreElements
忽略所有的數據，只保留終止通知(onError或onCompleted)

Sample
取樣，定期發射最新的數據，等於是數據抽樣，有的實現裡叫ThrottleFirst

throttleFirst
會定期發射這個時間段裡源Observable發射的第一個數據

Skip
跳過前面的若干項數據

SkipLast
跳過後面的若干項數據

Take
只保留前面的若干項數據

TakeLast
只保留後面的若干項數據

takeFirst
有null數據時會拋空指針異常,要判空處理
和first一樣，但在所有數據不滿足條件時不會拋出異常，僅僅調用onComplete.

組合操作

And/Then/When
通過模式(And條件)和計劃(Then次序)組合兩個或多個Observable發射的數據集

CombineLatest
當兩個Observables中的任何一個發射了一個數據時，通過一個指定的函數組合每個Observable發射的最新數據（一共兩個數據），然後發射這個函數的結果

必須滿足兩個條件:
1)所有的Observable都發射過數據。
2)滿足條件1的時候任何一個Observable發射一個數據，就將所有Observable最新發射的數據按照提供的函數組裝起來發射出去。

在這兩個條件下,可能會忽略掉一些發射的數據.

Join
無論何時，如果一個Observable發射了一個數據項，只要在另一個Observable發射的數據項定義的時間窗口內，就將兩個Observable發射的數據合並發射

參數說明:
1)源Observable所要組合的目標Observable
2)一個函數，就收從源Observable發射來的數據，並返回一個Observable，這個Observable的生命周期決定了源Observable發射出來數據的有效期
3)一個函數，就收從目標Observable發射來的數據，並返回一個Observable，這個Observable的生命周期決定了目標Observable發射出來數據的有效期
4)一個函數，接收從源Observable和目標Observable發射來的數據，並返回最終組合完的數據。

Merge
將兩個Observable發射的數據組合並成一個
Merge可能會讓合並的Observables發射的數據交錯（可以使用Concat操作符,不會讓數據交錯，它會按順序一個接著一個發射多個Observables的發射物）。

StartWith
在發射原來的Observable的數據序列之前，先發射一個指定的數據序列或數據項

在數據序列的開頭插入一條指定的項
你也可以傳遞一個Observable給startWith，
它會將那個Observable的發射物插在原始Observable發射的數據序列之前.這可以看作是Concat的反轉。

Switch，在RxJava的實現為SwitchOnNext
將一個發射Observable序列的Observable轉換為這樣一個Observable：它逐個發射那些Observable最近發射的數據
用來將一個發射多個小Observable的源Observable轉化為一個Observable，然後發射這多個小Observable所發射的數據。
需要注意的就是，如果一個小的Observable正在發射數據的時候，源Observable又發射出一個新的小Observable，則前一個Observable發射的數據會被拋棄，直接發射新的小Observable所發射的數據。

Zip

zip zipWith
打包，使用一個指定的函數將多個Observable發射的數據組合在一起，然後將這個函數的結果作為單項數據發射
Zip操作符將多個Observable發射的數據按順序組合起來，每個數據只能組合一次，而且都是有序的。
最終組合的數據的數量由發射數據最少的Observable來決定。

錯誤處理

這些操作符用於從錯誤通知中恢復

Catch
捕獲，繼續序列操作，將錯誤替換為正常的數據，從onError通知中恢復

onErrorReturn
當發生錯誤的時候，讓Observable發射一個預先定義好的數據並正常地終止,不會拋出異常 onErrorResumeNext
當發生錯誤的時候，由另外一個Observable來代替當前的Observable並繼續發射數據 onExceptionResumeNext
類似於OnErrorResume,不同之處在於其會對onError拋出的數據類型做判斷，
如果是Exception，也會使用另外一個Observable代替原Observable繼續發射數據，
否則會將錯誤分發給Subscriber。

Retry
重試，如果Observable發射了一個錯誤通知，重新訂閱它，期待它正常終止

retry
Retry操作符在發生錯誤的時候會重新進行訂閱,而且可以重復多次，
所以發射的數據可能會產生重復。如果重復指定次數還有錯誤的話就會將錯誤返回給觀察者,會掉onError retryWhen
當錯誤發生時，retryWhen會接收onError的throwable作為參數，並根據定義好的函數返回一個Observable，如果這個Observable發射一個數據，就會重新訂閱。
需要注意的是使用retryWhen的時候,因為每次重新訂閱都會產生錯誤，所以作為參數的obserbvable會不斷地發射數據，使用zipWith操作符可以限制重新訂閱的次數，否則會無限制地重新訂閱。
會正常結束,調用onCompleted

輔助操作

Delay
延遲一段時間發射結果數據

DelaySubscription
延遲注冊到Observer上

Do
Do操作符就是給Observable的生命周期的各個階段加上一系列的回調監聽，當Observable執行到這個階段的時候，這些回調就會被觸發.

doOnEach
Observable每發射一個數據的時候就會觸發這個回調，不僅包括onNext還包括onError和onCompleted。

doOnNext
只有onNext的時候才會被觸發。

doOnSubscribe,doOnUnSubscribe
會在Subscriber進行訂閱和反訂閱的時候觸發回調。
當一個Observable通過OnError或者OnCompleted結束的時候，會反訂閱所有的Subscriber。在Android中和生命周期綁定起來,因為有些Observable執行不完啊.

doOnSubscribeOn
在開始注冊前做一些工作。處於當前線程，而不是subscribeOn指定的線程。

DoOnError
在OnError發生的時候觸發回調，並將Throwable對象作為參數傳進回調函數裡。

DoOnComplete
會在OnCompleted發生的時候觸發回調。

DoOnTerminate
會在Observable結束前觸發回調，無論是正常還是異常終止。

finallyDo,doAfterTerminate
會在Observable結束後觸發回調，無論是正常還是異常終止。

Materialize,dematerialize
Meterialize操作符將OnNext/OnError/OnComplete都轉化為一個Notification對象並按照原來的順序發射出來,dematerialize相反
使用integerNotification.getValue() +”, ” + integerNotification.getKind()可以看到打印值和類型.

ObserveOn
指定Subscriber的調度程序（工作線程）

SubscribeOn
指定Observable應該在哪個調度程序上執行

Serialize
強制Observable按次序發射數據並且功能是有效的

Subscribe
收到Observable發射的數據和通知後執行的操作

TimeInterval
將一個Observable轉換為發射兩個數據之間所耗費時間的Observable
TimeInterval會攔截發射出來的數據，取代為前後兩個發射兩個數據的間隔時間。對於第一個發射的數據，其時間間隔為訂閱後到首次發射的間隔。

Timeout
添加超時機制，如果過了指定的一段時間沒有發射數據，就發射一個錯誤通知
Timeout操作符給Observable加上超時時間，每發射一個數據後就重置計時器，當超過預定的時間還沒有發射下一個數據，就拋出一個超時的異常。
Rxjava將Timeout實現為很多不同功能的操作符，比如說超時後用一個備用的Observable繼續發射數據等。

Timestamp
給Observable發射的每個數據項添加一個時間戳
TimeStamp會將每個數據項給重新包裝一下，加上了一個時間戳來標明每次發射的時間

Using
創建一個只在Observable的生命周期內存在的一次性資源

Using操作符創建一個在Observable生命周期內存活的資源，也可以這樣理解：
我們創建一個資源並使用它，用一個Observable來限制這個資源的使用時間，當這個Observable終止的時候，這個資源就會被銷毀。
Using需要使用三個參數，分別是：
1)創建這個一次性資源的函數
2)創建Observable的函數
3)釋放資源的函數

條件和布爾操作

這些操作符可用於單個或多個數據項，也可用於Observable

All
判斷Observable發射的所有的數據項是否都滿足某個條件

All操作符根據一個函數對源Observable發射的所有數據進行判斷，最終返回的結果就是這個判斷結果。
對發射的所有數據應用這個函數,如果全部都滿足則返回true，否則就返回false。

Amb
Amb操作符可以將至多9個Observable結合起來，讓他們競爭。
哪個Observable首先發射了數據（包括onError和onComplete)就會繼續發射這個Observable的數據，其他的Observable所發射的數據都會別丟棄。

Contains
判斷Observable是否會發射一個指定的數據項
Contains操作符用來判斷源Observable所發射的數據是否包含某一個數據，如果包含會返回true，如果源Observable已經結束了卻還沒有發射這個數據則返回false。

IsEmpty
IsEmpty操作符用來判斷源Observable是否發射過數據，沒有發射過數據返回true.
Null也是一個數據

DefaultIfEmpty
發射來自原始Observable的數據，如果原始Observable沒有發射數據，就發射一個默認數據
會判斷源Observable是否發射數據，如果源Observable發射了數據則正常發射這些數據，如果沒有則發射一個默認的數據

SequenceEqual
判斷兩個Observable是否按相同的數據序列
SequenceEqual操作符用來判斷兩個Observable發射的數據序列是否相同（發射的數據相同，數據的序列相同，結束的狀態相同），如果相同返回true，否則返回false

SkipUntil
SkipUnitl是根據一個標志Observable來判斷的，當這個標志Observable沒有發射數據的時候，所有源Observable發射的數據都會被跳過；當標志Observable發射了一個數據，則開始正常地發射數據。
一直等到skipUntil發射了數據才能發射源Observable的數據,並忽略了此段時間內的數據

SkipWhile
SkipWhile則是根據一個函數來判斷是否跳過數據，當函數返回值為true的時候則一直跳過源Observable發射的數據；當函數返回false的時候則開始正常發射數據。

TakeUntil
和SkipUtil恰好相反,只獲取takeUntil裡的Observable之前的數據

TakeWhile
和SkipWhile相反,獲取滿足skipWhile的數據

算術和聚合操作

Concat
將多個Observable結合成一個Observable並發射數據，並且嚴格按照先後順序發射數據，前一個Observable的數據沒有發射完，不發射後面Observable的數據

Count
Count操作符用來統計源Observable發射了多少個數據，最後將數目給發射出來；
如果源Observable發射錯誤，則會將錯誤直接報出來；在源Observable沒有終止前，count是不會發射統計數據的。

Reduce
Reduce操作符應用一個函數接收Observable發射的數據和函數的計算結果作為下次計算的參數，輸出最後的結果。
跟前面我們了解過的scan操作符很類似，只是scan會輸出每次計算的結果，而reduce只會輸出最後的結果。

Collect
collect用來將源Observable發射的數據給收集到一個數據結構裡面，需要使用兩個參數：
一個產生收集數據結構的函數
一個接收第一個函數產生的數據結構和源Observable發射的數據作為參數的函數。

連接操作

一些有精確可控的訂閱行為的特殊Observable

什麼是Connectable Observable:
就是一種特殊的Observable對象，並不是Subscrib的時候就發射數據，而是只有對其應用connect操作符的時候才開始發射數據，
所以可以用來更靈活的控制數據發射的時機。

使用Publish操作符將Observable轉換為Connectable Observable,然後可以通過connect控制何時發射.

Publish
將一個普通的Observable轉換為可連接的
Publish操作符就是用來將一個普通的Observable對象轉化為一個Connectable Observable。需要注意的是如果發射數據已經開始了再進行訂閱只能接收以後發射的數據。

Connect
Connect操作符就是用來觸發Connectable Observable發射數據的。
應用Connect操作符後會返回一個Subscription對象，通過這個Subscription對象，我們可以調用其unsubscribe方法來終止數據的發射。
另外，如果還沒有訂閱者訂閱的時候就應用Connect操作符也是可以使其開始發射數據的。

RefCount
RefCount操作符就是將一個Connectable Observable 對象再重新轉化為一個普通的Observable對象，這時候訂閱者進行訂閱時就會觸發數據的發射。

Replay
Replay操作符返回一個Connectable Observable 對象並且可以緩存其發射過的數據，這樣即使有訂閱者在其發射數據之後進行訂閱也能收到其之前發射過的數據。
不過使用Replay操作符我們最好還是限定其緩存的大小，否則緩存的數據太多了可會占用很大的一塊內存。
對緩存的控制可以從空間和時間兩個方面來實現。

直接返回一個connectable observable,不用publish

轉換操作

to

自定義操作符

lift compose