RxJava的響應(yīng)流程分析基本調(diào)用流程是怎樣的

本篇文章為大家展示了RxJava的響應(yīng)流程分析基本調(diào)用流程是怎樣的，內(nèi)容簡明扼要并且容易理解，絕對能使你眼前一亮，通過這篇文章的詳細(xì)介紹希望你能有所收獲。

一.用法

首先來看一個簡單的例子:

運(yùn)行結(jié)果為：

從結(jié)果中我們不難看出整體的調(diào)用流程：

首先通過調(diào)用Observable.create()方法生成一個被觀察者，緊接著在這里我們又調(diào)用了map()方法對原被觀察者進(jìn)行數(shù)據(jù)流的變換操作，生成一個新的被觀察者（為何是新的被觀察者后文會講），最后調(diào)用subscribe()方法，傳入我們的觀察者，這里觀察者訂閱的則是調(diào)用map()之后生成的新被觀察者。

在整個過程中我們會注意到三個主角：Observable、OnSubscribe、Subscriber，所有的操作都是圍繞它們進(jìn)行的。不難看出這里三個角色的分工：

• Observable：被觀察者的來源，亦或說是被觀察者本身

• OnSubscribe：用來通知觀察者的不同行為

• Subscriber：觀察者，通過實(shí)現(xiàn)對應(yīng)方法來產(chǎn)生具體的處理。

所以接下來我們以這三個角色為中心來分析具體的流程。

二.分析

1.訂閱過程

首先我們進(jìn)入Observable.create()看看：

這里調(diào)用構(gòu)造函數(shù)生成了一個Observable對象并將傳入的OnSubscribe賦給自己的成員變量onsubscribe，等等，這個hook是從哪里冒出來的？我們向上找：

RxJavaObservableExecutionHook這個抽象Proxy類默認(rèn)對OnSubscribe對象不做任何處理，不過通過繼承該類并重寫onCreate()等方法我們可以對這些方法對應(yīng)的時機(jī)做一些額外處理比如打Log或者一些數(shù)據(jù)收集方面的工作。

到目前最初始的被觀察者已經(jīng)生成了，我們再來看看觀察者這邊。我們知道通過調(diào)用observable.subscribe()方法傳入一個觀察者即構(gòu)成了觀察者與被觀察者之間的訂閱關(guān)系，那么這內(nèi)部又是如何實(shí)現(xiàn)的呢？看代碼：

這里我們略去部分無關(guān)代碼看主要部分，subscribe.onStart()默認(rèn)空實(shí)現(xiàn)我們暫且不用管它，對于傳進(jìn)來的subscriber要包裝成SafeSubscriber，這個SafeSubscriber對原來的subscriber的一系列方法做了更完善的處理，包括：onError()與onCompleted()只會有一個被執(zhí)行；保證一旦onError()或者onCompleted()被執(zhí)行，將不再能再執(zhí)onNext()等情況。這里封裝為SafeSubscriber之后，調(diào)用onSubscribe.call()，并將subscriber傳入，這樣就完成了一次訂閱。

顯而易見，Subscriber作為觀察者，在訂閱行為完成后，其具體行為在整個鏈?zhǔn)秸{(diào)用中起著至關(guān)重要的作用，我們來看看它內(nèi)部的構(gòu)成的主要部分：

每個Subscriber都持有一個SubscriptionList，這個list保存的是所有該觀察者的訂閱事件，同時Subscriber也對應(yīng)實(shí)現(xiàn)了Subscription接口，當(dāng)這個Subscriber取消訂閱的時候會將持有事件列表中的所有Subscription取消訂閱，并且從此不再接受任何訂閱事件。同時，通過Producer可以去限定該Subscriber所接收的數(shù)據(jù)流的總量，這個限制量其實(shí)是加在Subscriber.onNext()方法上的，onComplete()、onError()則不會受到其影響。因?yàn)槭堑讓映橄箢悾?code>onNext()、onComplete()、onError()統(tǒng)一不在這里處理。

2.變換過程

在收到Observable的消息之前我們有可能會對數(shù)據(jù)流進(jìn)行處理，例如map()、flatMap()、deBounce()、buffer()等方法，本例中我們用了map()方法，它接收了原被觀察者發(fā)射的數(shù)據(jù)并將通過該方法返回的結(jié)果作為新的數(shù)據(jù)發(fā)射出去，相當(dāng)于做了一層中間轉(zhuǎn)化：

我們接著看這個轉(zhuǎn)化過程：

這里是通過一個lift()方法實(shí)現(xiàn)的，再查看其他的轉(zhuǎn)化方法發(fā)現(xiàn)內(nèi)部也都使用lift()實(shí)現(xiàn)的，看來這個lift()就是關(guān)鍵所在了，不過不急，我們先來看看這個OperationMap是什么：

OperationMap實(shí)現(xiàn)了Operator接口的call()方法，該方法接受外部傳入的觀察者，并將其作為參數(shù)構(gòu)造出了一個新的觀察者，我們不難發(fā)現(xiàn)o.onNext(transformer.call(t));這一句起了至關(guān)重要的作用，這里的接口transformer將泛型T轉(zhuǎn)化為泛型R：

這樣之后，再將轉(zhuǎn)換后的數(shù)據(jù)傳回至原觀察者的onNext()方法，就完成了觀察數(shù)據(jù)流的轉(zhuǎn)化，但是你應(yīng)該也注意到了，我們用來做轉(zhuǎn)換的這個新的觀察者并沒有實(shí)現(xiàn)訂閱被觀察者的操作，這個訂閱操作又是在哪里實(shí)現(xiàn)的呢？答案就是接下來的lift()：

在這里我們新生成了一個Observable對象，在這個新對象的onSubscribe成員的call()方法中我們通過operator.call()拿到之前生成的未產(chǎn)生訂閱的觀察者st，之后將它作為參數(shù)傳入一開始的onSubscribe.call()中，即完成了這個中間訂閱的過程。
現(xiàn)在我們將整個流程梳理一下：

• 一次map()變換

• 根據(jù)Operator實(shí)例生成新的Subscriber

• 通過lift()生成新的Observable

• 原Subscriber訂閱新的Observavble

• 新的Observable中onSubscribe通知新Subscriber訂閱原Observable

• 新Subscriber將消息傳給原Subscriber。

為了便于理解，這里借用一下扔物線的圖：

以上就是一次map()變換的流程，事實(shí)上多次map()也是同樣道理：最外層的目標(biāo)Subscriber發(fā)生訂閱行為后，onSubscribe.onNext()會逐層嵌套調(diào)用，直至初始Observable被最底層的Subscriber訂閱，通過Operator的一層層變化將消息傳到目標(biāo)Subscriber。再次祭出扔物線的圖：

至于其他的多種變化的實(shí)現(xiàn)流程也都很類似，借助于Operator的不同實(shí)現(xiàn)來達(dá)到變換數(shù)據(jù)流的目的。例如其中的flatMap()，它需要進(jìn)行兩次lift()，其中第二次是OperationMerge，將轉(zhuǎn)換成的每一個Observable數(shù)據(jù)流通過InnerSubscriber這個紐帶訂閱后，在InnerSubscriber的onNext()中拿到R，再通過傳入的parent（也就是原MergeSubscriber）將它們?nèi)堪l(fā)射（emit）出去，由最外層我們傳入的Subscriber統(tǒng)一接收，這樣就完成了 T => Observable<R> => R 的轉(zhuǎn)化。

除此之外，還有許多各式各樣的操作符，如果它們還不能滿足你的需要，你也可以通過實(shí)現(xiàn)Operator接口定制新的操作符。靈活運(yùn)用它們往往能達(dá)到事半功倍的效果，比如通過使用sample()、debounce()等操作符有效避免backpressure的需要等等，這里就不一一介紹了。

上述內(nèi)容就是RxJava的響應(yīng)流程分析基本調(diào)用流程是怎樣的，你們學(xué)到知識或技能了嗎？如果還想學(xué)到更多技能或者豐富自己的知識儲備，歡迎關(guān)注億速云行業(yè)資訊頻道。