深入RxJava2 源碼解析(一)

本文作者JasonChen，原文地址： http://chblog.me/2018/12/19/rxjava2 源碼解析(一)/

ReactiveX 響應(yīng)式編程庫，這是一個(gè)程序庫，通過使用可觀察的事件序列來構(gòu)成異步和事件驅(qū)動(dòng)的程序。

其簡(jiǎn)化了異步多線程編程，在以前多線程編程的世界中，鎖、可重入鎖、同步隊(duì)列器、信號(hào)量、并發(fā)同步器、同步計(jì)數(shù)器、并行框架等都是具有一定的使用門檻，稍有不慎或者使用不成熟或?qū)ζ湓创a理解不深入都會(huì)造成相應(yīng)的程序錯(cuò)誤和程序性能的低下。

觀察者模型

24種設(shè)計(jì)模式的一種，觀察者Observer和主題Subject之間建立組合關(guān)系：Subject類實(shí)例中包含觀察者Observer的引用，增加引用的目的就是為了通知notify，重要點(diǎn)就是要在Subject的notify功能中調(diào)用Observer的接受處理函數(shù)receiveAndHandle。

個(gè)人理解：觀察者模型其實(shí)是一種異步回調(diào)通知，將數(shù)據(jù)的處理者先注冊(cè)到數(shù)據(jù)的輸入者那邊，這樣通過數(shù)據(jù)輸入者執(zhí)行某個(gè)函數(shù)去調(diào)用數(shù)據(jù)處理者的某個(gè)處理方法。

RxJava2

Rx有很多語言的實(shí)現(xiàn)庫，目前比較出名的就是RxJava2。這里主講Rxjava2的部門源碼解讀，內(nèi)部設(shè)計(jì)機(jī)制和內(nèi)部執(zhí)行的線程模型。

RxJava是近兩年來越來越流行的一個(gè)異步開發(fā)框架，其使用起來十分簡(jiǎn)單方便，功能包羅萬象，十分強(qiáng)大。

基本使用

使用RxJava2大致分為四個(gè)操作：

建立數(shù)據(jù)發(fā)布者

添加數(shù)據(jù)變換函數(shù)

設(shè)置數(shù)據(jù)發(fā)布線程池機(jī)制，訂閱線程池機(jī)制

添加數(shù)據(jù)訂閱者

// 創(chuàng)建flowable Flowable<Map<String, Map<String,Object>>> esFlowable = Flowable.create(new ElasticSearchAdapter(), BackpressureStrategy.BUFFER); Disposable disposeable = esFlowable// map操作 1.采集、2.清洗.map(DataProcess::dataProcess).subscribeOn(Schedulers.single())//計(jì)算任務(wù)調(diào)度器.observeOn(Schedulers.computation())// 訂閱者 consumer 執(zhí)行運(yùn)算.subscribe(keyMaps -> new PredictEntranceForkJoin().predictLogic(keyMaps));

以上就是一個(gè)實(shí)際的例子，里面的ElasticSearchAdapter實(shí)際隱藏了一個(gè)用戶自定義實(shí)現(xiàn)數(shù)據(jù)生產(chǎn)的subscribe接口：

FlowableOnSubscribe<T> source

用戶需要實(shí)現(xiàn)這個(gè)接口函數(shù)：

void subscribe(@NonNull FlowableEmitter<T> emitter) throws Exception

這個(gè)接口主要用于內(nèi)部回調(diào)，后面會(huì)有具體分析，
emitter 英文翻譯發(fā)射器，很形象，數(shù)據(jù)就是由它產(chǎn)生的，也是業(yè)務(wù)系統(tǒng)需要對(duì)接的地方，一般業(yè)務(wù)代碼實(shí)現(xiàn)這個(gè)接口類然后發(fā)射出需要處理的原始數(shù)據(jù)。

map函數(shù)作為數(shù)據(jù)變換處理的功能函數(shù)將原來的數(shù)據(jù)輸入變換為另外的數(shù)據(jù)集合，然后設(shè)置發(fā)布的線程池機(jī)制subscribeOn(Schedulers.single())，訂閱的線程池機(jī)制observeOn(Schedulers.computation())，最后添加數(shù)據(jù)訂閱函數(shù)，也就是業(yè)務(wù)系統(tǒng)需要實(shí)現(xiàn)另外一個(gè)地方，從而實(shí)現(xiàn)數(shù)據(jù)的自定義處理消費(fèi)。

rxjava2支持的lambda語法

• 創(chuàng)建操作符：just fromArray empty error never fromIterable timer interval intervalRange range/rangeLong defer
• 變換操作符：map flatMap flatmapIterable concatMap switchmap cast scan buffer toList groupBy toMap
• 過濾操作符：filter take takeLast firstElement/lastElement first/last firstOrError/lastOrError elementAt/elementAtOrError ofType skip/skipLast
  ignoreElements distinct/distinctUntilChanged timeout throttleFirst throttleLast/sample throttleWithTimeout/debounce
• 合并聚合操作符：startWith/startWithArray concat/concatArray merge/mergeArray concatDelayError/mergeDelayError zip combineLatest combineLatestDelayError
  reduce count collect
• 條件操作符：all ambArray contains any isEmpty defaultIfEmpty switchIfEmpty sequenceEqual takeUntil takeWhile skipUntil skipWhile

有一篇博客詳細(xì)介紹了rxjava的各種操作符，鏈接https://maxwell-nc.github.io/android/rxjava2-1.html

RxJava2 源碼解析

閱讀源碼個(gè)人比較喜歡帶著疑惑去看，這樣與目標(biāo)有方向。接下來的分析以Flowable為例，這里所有的例子都是按照Flowable為例，因?yàn)镕lowable在實(shí)際項(xiàng)目中比Observable可能用的多，因?yàn)閷?shí)際場(chǎng)景中數(shù)據(jù)生產(chǎn)速度和數(shù)據(jù)消費(fèi)速度都會(huì)有一定的不一致甚至數(shù)據(jù)生產(chǎn)速度遠(yuǎn)大于數(shù)據(jù)消費(fèi)速度。

數(shù)據(jù)發(fā)布和訂閱

首先從數(shù)據(jù)訂閱者開始，點(diǎn)進(jìn)源碼看進(jìn)一步解析，里面有很多subscribe重載接口：

public final Disposable subscribe(Consumer<? super T> onNext) {return subscribe(onNext, Functions.ON_ERROR_MISSING,Functions.EMPTY_ACTION, FlowableInternalHelper.RequestMax.INSTANCE);}public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError,Action onComplete, Consumer<? super Subscription> onSubscribe) {ObjectHelper.requireNonNull(onNext, "onNext is null");ObjectHelper.requireNonNull(onError, "onError is null");ObjectHelper.requireNonNull(onComplete, "onComplete is null");ObjectHelper.requireNonNull(onSubscribe, "onSubscribe is null");//組裝成FlowableSubscriberLambdaSubscriber<T> ls = new LambdaSubscriber<T>(onNext, onError, onComplete, onSubscribe);//調(diào)用核心的訂閱方法subscribe(ls);return ls;}public final void subscribe(FlowableSubscriber<? super T> s) {ObjectHelper.requireNonNull(s, "s is null");try {//注冊(cè)一些鉤子這里對(duì)此不進(jìn)行講解，主要不是核心方法Subscriber<? super T> z = RxJavaPlugins.onSubscribe(this, s);ObjectHelper.requireNonNull(z, "The RxJavaPlugins.onSubscribe hook returned a null FlowableSubscriber. Please check the handler provided to RxJavaPlugins.setOnFlowableSubscribe for invalid null returns. Further reading: https://github.com/ReactiveX/RxJava/wiki/Plugins");//核心訂閱方法，從名字也能讀出是指訂閱實(shí)際調(diào)用處//不同的數(shù)據(jù)產(chǎn)生類也就是實(shí)現(xiàn)Flowable抽象類的類//比如FlowableCreate，FlowSingle，FlowMap等等去實(shí)現(xiàn)自己的實(shí)際方法subscribeActual(z);} catch (NullPointerException e) { // NOPMDthrow e;} catch (Throwable e) {Exceptions.throwIfFatal(e);// can't call onError because no way to know if a Subscription has been set or not// can't call onSubscribe because the call might have set a Subscription alreadyRxJavaPlugins.onError(e);NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");npe.initCause(e);throw npe;}}

下面選擇FlowCreate的subscribeActual(Subscriber<? super T> t)方法進(jìn)行剖析。

public void subscribeActual(Subscriber<? super T> t) {BaseEmitter<T> emitter;//根據(jù)不同的回壓模式選擇不一樣的數(shù)據(jù)發(fā)射類//神奇的回壓模式其實(shí)本質(zhì)上就是一個(gè)個(gè)數(shù)據(jù)發(fā)射-消費(fèi)模式switch (backpressure) {case MISSING: {emitter = new MissingEmitter<T>(t);break;}//...default: {emitter = new BufferAsyncEmitter<T>(t, bufferSize());break;}}//回調(diào)注冊(cè)的FlowableSubscriber的onSubscribe方法//這里非常重要，因?yàn)檫@里涉及了rxjava特有的 request請(qǐng)求再消費(fèi)數(shù)據(jù)的模式//也就是說如果沒有request數(shù)據(jù)，那么就不會(huì)調(diào)用數(shù)據(jù)發(fā)射（發(fā)布）者的onNext方法，//那么數(shù)據(jù)訂閱者也就不會(huì)消費(fèi)到數(shù)據(jù)t.onSubscribe(emitter);try {//回調(diào)注冊(cè)的FlowableOnSubscribe<T> source的subscribe方法//這個(gè)source其實(shí)就是在創(chuàng)建Flow流時(shí)注冊(cè)的數(shù)據(jù)產(chǎn)生類，進(jìn)一步驗(yàn)證了上文中//提及的其需要實(shí)現(xiàn)FlowableOnSubscribe<T>接口source.subscribe(emitter);} catch (Throwable ex) {Exceptions.throwIfFatal(ex);emitter.onError(ex);}}//重點(diǎn)分析BufferAsyncEmitter這個(gè)類，看字面意思這是一個(gè)switch的默認(rèn)選擇類，//但其實(shí)它是回壓策略為BUFFER時(shí)的數(shù)據(jù)發(fā)射類//首先這個(gè)類的構(gòu)造函數(shù)具有兩個(gè)參數(shù)，很明顯這是 actul就是前面的t這個(gè)變量，也就是//注冊(cè)的數(shù)據(jù)消費(fèi)（訂閱）者，capacityHint則是設(shè)置容量大小的，默認(rèn)是128，如果需要擴(kuò)大需要//自行設(shè)置環(huán)境變量 rx2.buffer-sizeBufferAsyncEmitter(Subscriber<? super T> actual, int capacityHint) {super(actual);this.queue = new SpscLinkedArrayQueue<T>(capacityHint);this.wip = new AtomicInteger();}public void onNext(T t) {if (done || isCancelled()) {return;}if (t == null) {onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));return;}// queue 是存儲(chǔ)元素的隊(duì)列，也就是buffer的核心存儲(chǔ)。// 當(dāng)我們開始向下游發(fā)送數(shù)據(jù)的時(shí)候首先存入隊(duì)列，然后下面的drain則是進(jìn)行核心的queue.offer(t);drain();}//核心的類void drain() {//關(guān)鍵的地方 解決生產(chǎn)速率和消費(fèi)速率不一致的關(guān)鍵地方，也是我們寫并發(fā)程序值得借鑒的地方。//當(dāng)數(shù)據(jù)的產(chǎn)生者（發(fā)布）頻繁調(diào)用onNext方法時(shí)，這里產(chǎn)生并發(fā)調(diào)用關(guān)系，wip變量是atomic變量，//當(dāng)?shù)谝淮螆?zhí)行drain函數(shù)時(shí)，為0繼續(xù)執(zhí)行后面的流程，當(dāng)快速的繼續(xù)調(diào)用onNext方法時(shí)，wip不為0然后返回//那么后面的流程我們其實(shí)已經(jīng)很大概率會(huì)猜測(cè)到應(yīng)該是去取隊(duì)列的數(shù)據(jù)然后做一些操作if (wip.getAndIncrement() != 0) {return;}int missed = 1;//這里的downstream其實(shí)就是注冊(cè)的數(shù)據(jù)訂閱者，它是基類BaseEmitter的變量，前面初始化時(shí)調(diào)用了基類的構(gòu)造函數(shù)final Subscriber<? super T> a = downstream;final SpscLinkedArrayQueue<T> q = queue;for (;;) {long r = get();long e = 0L;while (e != r) {if (isCancelled()) {q.clear();return;}boolean d = done;//取隊(duì)列中的數(shù)據(jù)T o = q.poll();boolean empty = o == null;if (d && empty) {Throwable ex = error;if (ex != null) {error(ex);} else {complete();}return;}if (empty) {break;}//此處回調(diào)訂閱者的onNext方法去真正的執(zhí)行數(shù)據(jù)實(shí)例程序//到此數(shù)據(jù)從產(chǎn)生到消費(fèi)其生命周期已經(jīng)走完a.onNext(o);e++;}if (e == r) {if (isCancelled()) {q.clear();return;}boolean d = done;boolean empty = q.isEmpty();if (d && empty) {Throwable ex = error;if (ex != null) {error(ex);} else {complete();}return;}}if (e != 0) {//標(biāo)記已經(jīng)消費(fèi)的個(gè)數(shù)BackpressureHelper.produced(this, e);}//前面說過wip會(huì)原子性的增加，而且是每調(diào)用一次onNext增加一次//missed從其名解釋是指錯(cuò)過的意思，個(gè)人理解是錯(cuò)過消費(fèi)的數(shù)據(jù)個(gè)數(shù)，錯(cuò)過消費(fèi)//的意思其實(shí)就是指沒有進(jìn)行a.onNext數(shù)據(jù)消費(fèi)處理的數(shù)據(jù)missed = wip.addAndGet(-missed);if (missed == 0) {//如果沒有錯(cuò)過的數(shù)據(jù)也就是全部都消費(fèi)完那就跳出for循環(huán)//此處for循環(huán)方式和JUC源碼中Doug Lea的做法都有類似之處break;}}}

操作符與線程池機(jī)制原理剖析

首先在進(jìn)行源碼分析之前講述一下一種模式：裝飾者模式 24種模式中的一種，在java io源碼包中廣泛應(yīng)用
簡(jiǎn)單的來說是與被裝飾者具有相同接口父類同時(shí)又對(duì)被裝飾者進(jìn)行一層封裝（持有被裝飾者的引用），以此用來加上自身的特性。

回歸主題，當(dāng)我們使用操作符和線程池機(jī)制的時(shí)候做法都是在數(shù)據(jù)發(fā)布者后面進(jìn)行相應(yīng)的函數(shù)操作：

Disposable disposeable = scheduleObservable.map(aLong -> dataAdapter.handlerDpti()).map(DataProcess::dataProcess).subscribeOn(Schedulers.single())

那么為何這么做，接下來我們進(jìn)行源碼分析：

subscribeOn map 方法都在Flowable類中：

public final <R> Flowable<R> map(Function<? super T, ? extends R> mapper) {ObjectHelper.requireNonNull(mapper, "mapper is null");return RxJavaPlugins.onAssembly(new FlowableMap<T, R>(this, mapper));} public final Flowable<T> subscribeOn(@NonNull Scheduler scheduler, boolean requestOn) {ObjectHelper.requireNonNull(scheduler, "scheduler is null");return RxJavaPlugins.onAssembly(new FlowableSubscribeOn<T>(this, scheduler, requestOn)); }

這里是實(shí)例方法調(diào)用，傳進(jìn)了this對(duì)象這個(gè)很關(guān)鍵，這里其實(shí)就是我們前面提到的裝修者模式，持有上游對(duì)象也就是數(shù)據(jù)源source的引用。

以FlowableSubscribeOn為例進(jìn)行分析，這個(gè)類經(jīng)常會(huì)用到，因?yàn)槠鋬?nèi)部設(shè)置了線程池的機(jī)制所以在實(shí)際使用項(xiàng)目中會(huì)大量使用，那么是如何做到線程池方式的呢？進(jìn)一步利用源碼進(jìn)行分析。

2.裝飾者的內(nèi)部代碼分析

以subscribeOn 為例:

//很明顯 實(shí)現(xiàn)的抽象類其實(shí)是裝修者抽象類public final class FlowableSubscribeOn<T> extends AbstractFlowableWithUpstream<T , T>// 這個(gè)在前面我們重點(diǎn)分析過這是實(shí)際訂閱執(zhí)行的類方法，其實(shí)也就是我們說的裝飾方法，里面實(shí)現(xiàn)了每個(gè)類自己的特定“裝修”方法@Overridepublic void subscribeActual(final Subscriber<? super T> s) {// 獲取訂閱者，下一篇文章會(huì)重點(diǎn)講述rxjava的線程池分配機(jī)制Scheduler.Worker w = scheduler.createWorker();final SubscribeOnSubscriber<T> sos = new SubscribeOnSubscriber<T>(s, w, source, nonScheduledRequests);// 跟前面一樣調(diào)用數(shù)據(jù)訂閱者的onSubscribe方法s.onSubscribe(sos);// 由分配的調(diào)度者進(jìn)行訂閱任務(wù)的執(zhí)行w.schedule(sos);}// 開始分析SubscribeOnSubscriber這個(gè)靜態(tài)內(nèi)部類的內(nèi)部代碼// 實(shí)現(xiàn)了Runable用來異步執(zhí)行static final class SubscribeOnSubscriber<T> extends AtomicReference<Thread>implements FlowableSubscriber<T>, Subscription, Runnable// 下游訂閱引用final Subscriber<? super T> downstream;// 上游發(fā)射類引用final AtomicReference<Subscription> upstream;// 上游數(shù)據(jù)源引用 跟上游引用有區(qū)別，簡(jiǎn)單的說每個(gè)上游數(shù)據(jù)源引用有自己的上游發(fā)射類Publisher<T> source;// 這里是裝飾的核心代碼@Overridepublic void run() {lazySet(Thread.currentThread());// source即為上游，表示其所裝飾的源Publisher<T> src = source;source = null;// 調(diào)用上游的自身的subscribe方法，在上面一開始我們說這個(gè)方法內(nèi)部會(huì)去調(diào)用自身實(shí)現(xiàn)的subscribeActual方法// 從而實(shí)現(xiàn)上游自己的特定方法，比如假設(shè)source是FlowCreate那么此處就會(huì)調(diào)用前面一開始我們所講到的數(shù)據(jù)的發(fā)射src.subscribe(this);}// 既然已經(jīng)保證了數(shù)據(jù)的發(fā)射那么數(shù)據(jù)的處理是不是也要處理// 很明顯這是調(diào)用了下游訂閱者的onNext方法@Overridepublic void onNext(T t) {downstream.onNext(t);}

本文總結(jié)

筆者喜歡總結(jié)，總結(jié)意味著我們反思和學(xué)習(xí)前面的知識(shí)點(diǎn)，應(yīng)用點(diǎn)以及自身的不足。

• 設(shè)計(jì)模式：觀察者模式和裝修者模式
• 并發(fā)處理技巧：回壓策略（其實(shí)本質(zhì)是緩存）的實(shí)現(xiàn)原理以及細(xì)節(jié)點(diǎn)

訂閱最新文章，歡迎關(guān)注我的公眾號(hào)

總結(jié)

以上是生活随笔為你收集整理的深入RxJava2 源码解析(一)的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。