Rxjava的思想

? ? 有一個(gè)起點(diǎn)和終點(diǎn),在數(shù)據(jù)的流向過(guò)程中,可用增加數(shù)據(jù)的攔截 攔截時(shí)可以對(duì)數(shù)據(jù)發(fā)生改變,終點(diǎn)只關(guān)心它上一個(gè)攔截

Rxjava的使用:

dependencies{

implementation'io.reactivex.rxjava2:rxandroid:2.0.1'

implementation'io.reactivex.rxjava2:rxjava:2.1.6'

}

1.rxjava設(shè)計(jì)模式

????rxjava準(zhǔn)確來(lái)說(shuō)不是標(biāo)準(zhǔn)的觀察模式,在標(biāo)準(zhǔn)的觀察者模式中是一個(gè)被觀察者對(duì)應(yīng)多個(gè)觀察者,而在rxjava是多個(gè)被觀察者對(duì)應(yīng)一個(gè)觀察者

?????為什么是個(gè)被觀察者對(duì)應(yīng)一個(gè)觀察者? 因?yàn)槲覀僌bservable.create是一個(gè)被觀察者,map操作符也是被觀察者,最后我們只有一個(gè)終點(diǎn),所以是多個(gè)被觀察者對(duì)應(yīng)一個(gè)觀察者

2.rxjava原理分析:主要對(duì)以下源碼進(jìn)行分析,其他操作符號(hào)原理也是差不多的,如果覺(jué)得new?Observer太麻煩了 可以實(shí)現(xiàn)new?Consumer 這個(gè)然后實(shí)現(xiàn)accept方法

圖1

(1)?RxJavaPlugins.onAssembly:默認(rèn)情況下是返回source,如果需要監(jiān)聽(tīng)Rx的操作符,可以調(diào)用setOnObservableAssembly方法,然后調(diào)用RxJavaPlugins.setOnObservableAssembly方法

(1) Create操作符:創(chuàng)建了一個(gè)ObservableCreate對(duì)象,并且把source傳給了ObservableCreate對(duì)象,然后在ObservableCreate構(gòu)造方法里面保存了source

(2) map操符號(hào):創(chuàng)建一個(gè)ObservableMap對(duì)象傳入一個(gè)this和mapper,這里的this的實(shí)例就是ObservableCreate,我們可以看作是ObservableCreate調(diào)用map方法? mapper就是我們自定義的Function,然后ObservableMap里面也是保存了source和Function

?(3) subscribe操作符:直接看核心代碼subscribeActual方法,subscribeActual是一個(gè)抽象方法,但是我們是通過(guò) ObservableMap調(diào)用出來(lái)的,我們直接看ObservableMap的subscribeActual方法

圖4

? ? ? ?調(diào)用的是source.subscribe(new MapObserver(t, function)); 這個(gè)source是什么呢?這個(gè)source就是ObservableCreate,因?yàn)樵贠bservableMap的是我們傳入了一個(gè)this和Function,所以這里就開始調(diào)用了ObservableCreate.subscribe(new MapObserver)方法ObservableCreate并沒(méi)有實(shí)現(xiàn)subscribe方法,但是它的父類實(shí)現(xiàn)了subscribe方法所以也是調(diào)用了ObservableCreate.subscribeActual方法并且把MapObserver傳給了ObservableCreate.subscribeActual方法

?(4) ObservableCreate.subscribeActual:拿到了MapObserver傳給了發(fā)射器CreateEmitter,CreateEmitter可以看作Observer的包裝類,然后回調(diào)了observer.onSubscribe, MapObserver并沒(méi)有實(shí)現(xiàn)onSubscribe方法,但是父類BasicFuseableObserver實(shí)現(xiàn)了Observer接口并且調(diào)用了actual.onSubscribe(this)? ?這個(gè)actual又是什么東西?? 這個(gè)actual是在new MapObserver(t, function)傳入的,這個(gè)t在我們調(diào)用subscribe操作符號(hào)的時(shí)候傳入的new?Observer,然后回調(diào)observer.onSubscribe方法;接著調(diào)用source.subscribe方法,這個(gè)source指的是create傳入的new?ObservableOnSubscribe

?(5) ObservableEmitter.onNext方法:當(dāng)我們調(diào)用onNext方法時(shí)候也就是調(diào)用ObservableEmitter.onNext方法,ObservableEmitter的onNext方法比較簡(jiǎn)單,先是observer.onNext(t);那么這個(gè)observer又是誰(shuí)呢?通過(guò)ObservableEmitter構(gòu)造方法得知傳入了一個(gè)observer,這個(gè)observer就是ObservableMap的MapObserver

(5) MapObserver.onNext方法:直接看核心代碼mapper.apply(t),這個(gè)mapper又是什么呢?回到我們的第二步是不是傳入一個(gè)mapper,這個(gè)mapper就是Function,然后又將Function傳給了這個(gè)MapObserver,所以就將這個(gè)轉(zhuǎn)換后的數(shù)據(jù)apply過(guò)去.然后接著調(diào)用downstream.onNext(v)方法,這個(gè)downstream又是啥呢?這個(gè)downstream回到圖4,在構(gòu)造方法傳入了一個(gè)t,這個(gè)t就是我們的觀察者new?Observer 所以downstream.onNext方法就會(huì)回調(diào)了觀察者的onNext方法

為了方便理解繪畫了一份思維導(dǎo)圖

總結(jié):rxjava的結(jié)構(gòu)是一種u形的結(jié)構(gòu),主要是subscribe開始的,先是調(diào)用的ObservableMap.subscribeActual方法,以打包的形式new MapObserver(t, function) 回調(diào)了ObservableCreate.subscribeActual,然后在ObservableCreate的subscribeActual方法里面進(jìn)行了一個(gè)拆包,拆包是從最外面開始拆的也就說(shuō)調(diào)用了MapObserver.onNext方法進(jìn)行數(shù)據(jù)的回調(diào).

1.創(chuàng)建一個(gè)ObservableCreate對(duì)象 ,并且內(nèi)存存儲(chǔ)了同一個(gè)自定義source

2.調(diào)用map操作符,內(nèi)部創(chuàng)建一個(gè)ObservableMap對(duì)象,內(nèi)存存儲(chǔ)了一個(gè)source和function ,這個(gè)source指的就是ObservableCreate,那是應(yīng)該我們通過(guò)ObservableCreate調(diào)用map操作符

3.subscribe方法會(huì)傳入一個(gè)Observer觀察者,然后調(diào)用subscribeActual方法subscribeActual是一個(gè)抽象方法 但是ObservableMap.實(shí)現(xiàn)了subscribeActual方法

4.ObservableMap.subscribeActual調(diào)用source.subscribe(MapObserver()); 這個(gè)source就是ObservableCreate 也就是說(shuō)調(diào)用了ObservableCreate.subscribeActual方法

5.ObservableCreate.subscribeActual內(nèi)部先創(chuàng)建一個(gè)CreateEmitter發(fā)射器,這個(gè)發(fā)射器相當(dāng)于將MapObserver的包裝類,然后調(diào)用observer.onSubscribe()方法 然后就回調(diào)了onSubscribe方法,接著調(diào)用source.subscribe(CreateEmitter),因?yàn)檫@個(gè)source 是在ObservableCreate的時(shí)候傳入的一個(gè)自定義source,所以就回調(diào)了subscribe方法

5.ObservableEmitter.onNext方法 調(diào)用發(fā)射器的onNext ,內(nèi)部也有一個(gè)observer觀察者,然后調(diào)用?MapObserver.onNext ,然后回到observer.onNext方法,它會(huì)調(diào)用mapper.apply()方法將數(shù)據(jù)轉(zhuǎn)換需要的類型,然后回調(diào)onNext方法,并且返回?cái)?shù)據(jù)