1. RxJava操作符介紹
創(chuàng)建操作
- Create 從頭創(chuàng)建一個(gè)Observable
- Defer 直到有觀察者訂閱時(shí)才創(chuàng)建Observable，并且為每個(gè)觀察者創(chuàng)建一個(gè)新的Observable
- Empty/Never/Throw 創(chuàng)建行為受限的特殊Observable
Empty 創(chuàng)建一個(gè)不發(fā)射任何數(shù)據(jù)但是正常終止的Observable
Never 創(chuàng)建一個(gè)不發(fā)射數(shù)據(jù)也不終止的Observable
Throw 創(chuàng)建一個(gè)不發(fā)射數(shù)據(jù)以一個(gè)錯(cuò)誤終止的Observable - From 將其它對(duì)象或數(shù)據(jù)結(jié)構(gòu)創(chuàng)建為Observable挨個(gè)發(fā)射
- Interval 創(chuàng)建一個(gè)固定間隔發(fā)射整數(shù)序列的Observable幕袱，替代Timer
- Just 創(chuàng)建一個(gè)發(fā)射指定值的Observable
- Range 創(chuàng)建發(fā)射指定范圍的整數(shù)序列的Observable
- Repeat 創(chuàng)建一個(gè)發(fā)射特定數(shù)據(jù)重復(fù)多次的Observable
- Timer 創(chuàng)建在一個(gè)指定的延遲之后發(fā)射單個(gè)數(shù)據(jù)的Observable
變換操作（對(duì)Observable發(fā)射的數(shù)據(jù)進(jìn)行變換）
- Buffer 緩存暴备，定期收集Observable的數(shù)據(jù)放進(jìn)一個(gè)數(shù)據(jù)包裹，然后發(fā)射這些數(shù)據(jù)包裹们豌，而不是一次發(fā)射一個(gè)值
- FlatMap 將一個(gè)發(fā)射數(shù)據(jù)的Observable變換為多個(gè)Observables涯捻，然后將它們發(fā)射的數(shù)據(jù)合并后放進(jìn)一個(gè)單獨(dú)的Observable
- GroupBy 將原來(lái)的Observable分拆為Observable集合，將原始Observable發(fā)射的數(shù)據(jù)按Key分組望迎，每一個(gè)Observable發(fā)射一組不同的數(shù)據(jù)
- Map 通過(guò)對(duì)序列的每一項(xiàng)都應(yīng)用一個(gè)函數(shù)變換Observable發(fā)射的數(shù)據(jù)障癌，實(shí)質(zhì)是對(duì)序列中的每一項(xiàng)執(zhí)行一個(gè)函數(shù)，函數(shù)的參數(shù)就是這個(gè)數(shù)據(jù)項(xiàng)
- Scan 對(duì)Observable發(fā)射的每一項(xiàng)數(shù)據(jù)應(yīng)用一個(gè)函數(shù)辩尊，然后按順序依次發(fā)射這些值
- Window 定期將來(lái)自O(shè)bservable的數(shù)據(jù)分拆成一些Observable窗口涛浙，然后發(fā)射這些窗口，而不是每次發(fā)射一項(xiàng)
過(guò)濾操作
- Debounce 僅在過(guò)了一段指定的時(shí)間還沒(méi)發(fā)射數(shù)據(jù)時(shí)才發(fā)射一個(gè)數(shù)據(jù)
- Distinct 過(guò)濾掉重復(fù)數(shù)據(jù)項(xiàng)
- ElementAt 取特定位置的數(shù)據(jù)項(xiàng)
- Filter 過(guò)濾掉函數(shù)返回false的數(shù)據(jù)項(xiàng)
- First 只發(fā)射滿足條件的第一條數(shù)據(jù)
- IgnoreElements 忽略所有的數(shù)據(jù)摄欲，只保留終止通知(onError或onCompleted)
- Last 只發(fā)射最后一條數(shù)據(jù)
- Sample 定期發(fā)射最新的數(shù)據(jù)轿亮，等于是數(shù)據(jù)抽樣，有的實(shí)現(xiàn)里叫ThrottleFirst
- Skip 跳過(guò)前面的若干項(xiàng)數(shù)據(jù)
- SkipLast 跳過(guò)后面的若干項(xiàng)數(shù)據(jù)
- Take 只保留前面的若干項(xiàng)數(shù)據(jù)
- TakeLast 只保留后面的若干項(xiàng)數(shù)據(jù)
組合操作
- CombineLatest 當(dāng)兩個(gè)Observables中的任何一個(gè)發(fā)射了數(shù)據(jù)時(shí)胸墙，使用一個(gè)函數(shù)結(jié)合每個(gè)Observable發(fā)射的最近數(shù)據(jù)項(xiàng)我注，并且基于這個(gè)函數(shù)的結(jié)果發(fā)射數(shù)據(jù)
- Join 無(wú)論何時(shí)，如果一個(gè)Observable發(fā)射了一個(gè)數(shù)據(jù)項(xiàng)迟隅，只要在另一個(gè)Observable發(fā)射的數(shù)據(jù)項(xiàng)定義的時(shí)間窗口內(nèi)仓手，就將兩個(gè)Observable發(fā)射的數(shù)據(jù)合并發(fā)射
- Merge 將兩個(gè)Observable發(fā)射的數(shù)據(jù)組合并成一個(gè)
- StartWith 在發(fā)射原來(lái)的Observable的數(shù)據(jù)序列之前，先發(fā)射一個(gè)指定的數(shù)據(jù)序列或數(shù)據(jù)項(xiàng)
- Switch 將一個(gè)發(fā)射多個(gè)Observables的Observable轉(zhuǎn)換成另一個(gè)單獨(dú)的Observable玻淑，如果Observable正在發(fā)射數(shù)據(jù)的時(shí)候，源Observable又發(fā)射出一個(gè)新的Observable呀伙，則前一個(gè)Observable發(fā)射的數(shù)據(jù)會(huì)被拋棄补履，直接發(fā)射新的Observable所發(fā)射的數(shù)據(jù)
- Zip 通過(guò)一個(gè)函數(shù)將多個(gè)Observables的發(fā)射物結(jié)合到一起，基于這個(gè)函數(shù)的結(jié)果為每個(gè)結(jié)合體發(fā)射單個(gè)數(shù)據(jù)項(xiàng)剿另。Zip操作符將多個(gè)Observable發(fā)射的數(shù)據(jù)按順序組合起來(lái)箫锤，每個(gè)數(shù)據(jù)只能組合一次，而且都是有序的雨女。最終組合的數(shù)據(jù)的數(shù)量由發(fā)射數(shù)據(jù)最少的Observable來(lái)決定谚攒。
錯(cuò)誤處理
- Catch 繼續(xù)序列操作，將錯(cuò)誤替換為正常的數(shù)據(jù)氛堕，從onError通知中恢復(fù)
- Retry 如果Observable發(fā)射了一個(gè)錯(cuò)誤通知馏臭，重新訂閱它，期待它正常終止讼稚。在發(fā)生錯(cuò)誤的時(shí)候會(huì)重新進(jìn)行訂閱,而且可以重復(fù)多次括儒，所以發(fā)射的數(shù)據(jù)可能會(huì)產(chǎn)生重復(fù)绕沈。如果重復(fù)指定次數(shù)還有錯(cuò)誤的話就會(huì)將錯(cuò)誤返回給觀察者
RetryWhen 指示Observable遇到錯(cuò)誤時(shí)，將錯(cuò)誤傳遞給另一個(gè)Observable來(lái)決定是否要重新給訂閱這個(gè)Observable帮寻，新Observable處理錯(cuò)誤乍狐，老的繼續(xù)流程。
輔助操作
- Delay 延遲一段時(shí)間發(fā)射結(jié)果數(shù)據(jù)
- Do 注冊(cè)一個(gè)動(dòng)作占用一些Observable的生命周期事件固逗，相當(dāng)于Mock某個(gè)操作浅蚪，Do操作符就是給Observable的生命周期的各個(gè)階段加上一系列的回調(diào)監(jiān)聽(tīng)
- Materialize/Dematerialize 將發(fā)射的數(shù)據(jù)和通知都當(dāng)做數(shù)據(jù)發(fā)射，或者反過(guò)來(lái)
- ObserveOn 指定觀察者觀察Observable的調(diào)度程序（工作線程）
- Serialize 強(qiáng)制Observable按次序發(fā)射數(shù)據(jù)并且功能是有效的
- Subscribe 收到Observable發(fā)射的數(shù)據(jù)和通知后執(zhí)行的操作
- SubscribeOn 指定Observable應(yīng)該在哪個(gè)調(diào)度程序上執(zhí)行
- TimeInterval 將一個(gè)Observable轉(zhuǎn)換為發(fā)射兩個(gè)數(shù)據(jù)之間所耗費(fèi)時(shí)間的Observable
- Timeout 添加超時(shí)機(jī)制烫罩，如果過(guò)了指定的一段時(shí)間沒(méi)有發(fā)射數(shù)據(jù)惜傲，就發(fā)射一個(gè)錯(cuò)誤通知
- Timestamp 給Observable發(fā)射的每個(gè)數(shù)據(jù)項(xiàng)添加一個(gè)時(shí)間戳
- Using 創(chuàng)建一個(gè)只在Observable的生命周期內(nèi)存在的一次性資源
我們創(chuàng)建一個(gè)資源并使用它，用一個(gè)Observable來(lái)限制這個(gè)資源的使用時(shí)間嗡髓，當(dāng)這個(gè)Observable終止的時(shí)候操漠，這個(gè)資源就會(huì)被銷(xiāo)毀。
條件和布爾操作
- All 判斷Observable發(fā)射的所有的數(shù)據(jù)項(xiàng)是否都滿足某個(gè)條件
- Amb 給定多個(gè)Observable饿这，只讓第一個(gè)發(fā)射數(shù)據(jù)的Observable發(fā)射浊伙，將至多9個(gè)Observable結(jié)合起來(lái)，讓他們競(jìng)爭(zhēng)长捧。哪個(gè)Observable首先發(fā)射了數(shù)據(jù)（包括onError和onComplete)就會(huì)繼續(xù)發(fā)射這個(gè)Observable的數(shù)據(jù)嚣鄙，其他的Observable所發(fā)射的數(shù)據(jù)都會(huì)被丟棄。
- Contains 判斷Observable是否會(huì)發(fā)射一個(gè)指定的數(shù)據(jù)項(xiàng)串结，如果源Observable已經(jīng)結(jié)束了卻還沒(méi)有發(fā)射這個(gè)數(shù)據(jù)則返回false
- IsEmpty操作符用來(lái)判斷源Observable是否發(fā)射過(guò)數(shù)據(jù)哑子，如果發(fā)射過(guò)就會(huì)返回false，如果源Observable已經(jīng)結(jié)束了卻還沒(méi)有發(fā)射這個(gè)數(shù)據(jù)則返回true肌割。
- DefaultIfEmpty 發(fā)射來(lái)自原始Observable的數(shù)據(jù)卧蜓，如果原始Observable沒(méi)有發(fā)射數(shù)據(jù)，就發(fā)射一個(gè)默認(rèn)數(shù)據(jù)
- SequenceEqual 判斷兩個(gè)Observable是否按相同的數(shù)據(jù)序列
- SkipUntil 丟棄原始Observable發(fā)射的數(shù)據(jù)把敞，直到標(biāo)志的Observable發(fā)射了一個(gè)數(shù)據(jù)弥奸，然后發(fā)射原始Observable的剩余數(shù)據(jù)
- SkipWhile 丟棄原始Observable發(fā)射的數(shù)據(jù)，直到一個(gè)特定的條件為假奋早，然后發(fā)射原始Observable剩余的數(shù)據(jù)
- TakeUntil 發(fā)射來(lái)自原始Observable的數(shù)據(jù)盛霎，直到標(biāo)志的Observable發(fā)射了一個(gè)數(shù)據(jù)，然后跳過(guò)剩余的數(shù)據(jù)
- TakeWhile 發(fā)射原始Observable的數(shù)據(jù)耽装，直到一個(gè)特定的條件為真愤炸，然后跳過(guò)剩余的數(shù)據(jù)
算術(shù)和聚合操作
- Average 計(jì)算Observable發(fā)射的數(shù)據(jù)序列的平均值，然后發(fā)射這個(gè)結(jié)果
- Concat 不交錯(cuò)的連接多個(gè)Observable的數(shù)據(jù)
將多個(gè)Observable結(jié)合成一個(gè)Observable并發(fā)射數(shù)據(jù)掉奄，并且嚴(yán)格按照先后順序發(fā)射數(shù)據(jù)规个，前一個(gè)Observable的數(shù)據(jù)沒(méi)有發(fā)射完，是不能發(fā)射后面Observable的數(shù)據(jù)的。 - Count 計(jì)算Observable發(fā)射的數(shù)據(jù)個(gè)數(shù)绰姻，然后發(fā)射這個(gè)結(jié)果
如果源Observable發(fā)射錯(cuò)誤枉侧，則會(huì)將錯(cuò)誤直接報(bào)出來(lái)；在源Observable沒(méi)有終止前狂芋，count是不會(huì)發(fā)射統(tǒng)計(jì)數(shù)據(jù)的榨馁。 - Max 計(jì)算并發(fā)射數(shù)據(jù)序列的最大值
- Min計(jì)算并發(fā)射數(shù)據(jù)序列的最小值
- Reduce 按順序?qū)?shù)據(jù)序列的每一個(gè)應(yīng)用某個(gè)函數(shù)，然后返回這個(gè)值
接收Observable發(fā)射的數(shù)據(jù)和函數(shù)的計(jì)算結(jié)果作為下次計(jì)算的參數(shù)帜矾，輸出最后的結(jié)果翼虫。 - Sum 計(jì)算并發(fā)射數(shù)據(jù)序列的和
連接操作
- Connect 指示一個(gè)可連接的Observable開(kāi)始發(fā)射數(shù)據(jù)給訂閱者
- Publish 將一個(gè)普通的Observable轉(zhuǎn)換為可連接的
- RefCount 使一個(gè)可連接的Observable表現(xiàn)得像一個(gè)普通的Observable
- Replay 確保所有的觀察者收到同樣的數(shù)據(jù)序列，即使他們?cè)贠bservable開(kāi)始發(fā)射數(shù)據(jù)之后才訂閱
自定義操作符
- lift 如果我們的自定義操作符想要作用到Observable發(fā)射出來(lái)的數(shù)據(jù)上屡萤，使用lift操作符
- compose 如果我們的自定義操作符想要改變整個(gè)的Observable珍剑，使用compose操作符
2. RxJava操作符的使用
3. lift源碼解析
1.先看看Observable的create()
public static <T> Observable<T> create(OnSubscribe<T> f) {
return new Observable<T>(RxJavaHooks.onCreate(f));
}
Observable的構(gòu)造方法
protected Observable(OnSubscribe<T> f) {
this.onSubscribe = f;
}
和RxJavaHooks的onCreate()方法
public static <T> Observable.OnSubscribe<T> onCreate(Observable.OnSubscribe<T> onSubscribe) {
Func1<OnSubscribe, OnSubscribe> f = onObservableCreate;
if (f != null) {
return f.call(onSubscribe);
}
return onSubscribe;
}
hook了Observable.OnSubscribe
2.在看Observable的lift()
public final <R> Observable<R> lift(final Operator<? extends R, ? super T> operator) {
return create(new OnSubscribeLift<T, R>(onSubscribe, operator));
}
OnSubscribeLift類(lèi)
public final class OnSubscribeLift<T, R> implements OnSubscribe<R> {
final OnSubscribe<T> parent;
final Operator<? extends R, ? super T> operator;
public OnSubscribeLift(OnSubscribe<T> parent, Operator<? extends R, ? super T> operator) {
this.parent = parent;
this.operator = operator;
}
@Override
public void call(Subscriber<? super R> o) {
try {
Subscriber<? super T> st = RxJavaHooks.onObservableLift(operator).call(o);
try {
st.onStart();
parent.call(st);
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
st.onError(e);
}
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
o.onError(e);
}
}
}
RxJavaHooks的onObservableLift()方法
public static <T, R> Operator<R, T> onObservableLift(Operator<R, T> operator) {
Func1<Operator, Operator> f = onObservableLift;
if (f != null) {
return f.call(operator);
}
return operator;
}
同樣hook了Observable.OnSubscribe，是新的onSubscribe
這里的parent就是舊onSubscribe
總結(jié)下死陆，新的Observable會(huì)像一個(gè)代理一樣招拙，負(fù)責(zé)接收原始的Observable發(fā)出的事件，并在處理后發(fā)送給Subscriber
4. RxJava應(yīng)用場(chǎng)景
- 取數(shù)據(jù)先檢查緩存的場(chǎng)景
- 界面需要等到多個(gè)接口并發(fā)取完數(shù)據(jù)措译，再更新merge
- 一個(gè)接口的請(qǐng)求依賴(lài)另一個(gè)API請(qǐng)求返回的數(shù)據(jù)
- 界面按鈕需要防止連續(xù)點(diǎn)擊的情況throttleFirst
- 響應(yīng)式的界面checkedChanges
- 復(fù)雜的數(shù)據(jù)變換
- 輪詢(xún)
- 線程切換
- RxBus
- RxBinding Android控件對(duì)RxJava的支持庫(kù)
- rx-preferences 使SharedPreferences支持 RxJava
- RxLifecycle 幫助RxJava在Android中生命周期的控制别凤，避免內(nèi)存溢出等問(wèn)題
- retrofit Retrofit
- storio 數(shù)據(jù)庫(kù)對(duì)RxJava的支持
