深入RxJava2 源碼解析(一)

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

ReactiveX 響應式編程庫，這是一個程序庫，通過使用可觀察的事件序列來構成異步和事件驅動的程序。

其簡化了異步多線程編程，在以前多線程編程的世界中，鎖、可重入鎖、同步隊列器、信號量、并發同步器、同步計數器、并行框架等都是具有一定的使用門檻，稍有不慎或者使用不成熟或對其源碼理解不深入都會造成相應的程序錯誤和程序性能的低下。

觀察者模型

24種設計模式的一種，觀察者Observer和主題Subject之間建立組合關系：Subject類實例中包含觀察者Observer的引用，增加引用的目的就是為了通知notify，重要點就是要在Subject的notify功能中調用Observer的接受處理函數receiveAndHandle。

個人理解：觀察者模型其實是一種異步回調通知，將數據的處理者先注冊到數據的輸入者那邊，這樣通過數據輸入者執行某個函數去調用數據處理者的某個處理方法。

RxJava2

Rx有很多語言的實現庫，目前比較出名的就是RxJava2。這里主講Rxjava2的部門源碼解讀，內部設計機制和內部執行的線程模型。

RxJava是近兩年來越來越流行的一個異步開發框架，其使用起來十分簡單方便，功能包羅萬象，十分強大。

基本使用

使用RxJava2大致分為四個操作：

建立數據發布者

添加數據變換函數

設置數據發布線程池機制，訂閱線程池機制

添加數據訂閱者

// 創建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())//計算任務調度器.observeOn(Schedulers.computation())// 訂閱者 consumer 執行運算.subscribe(keyMaps -> new PredictEntranceForkJoin().predictLogic(keyMaps));

以上就是一個實際的例子，里面的ElasticSearchAdapter實際隱藏了一個用戶自定義實現數據生產的subscribe接口：

FlowableOnSubscribe<T> source

用戶需要實現這個接口函數：

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

這個接口主要用于內部回調，后面會有具體分析，
emitter 英文翻譯發射器，很形象，數據就是由它產生的，也是業務系統需要對接的地方，一般業務代碼實現這個接口類然后發射出需要處理的原始數據。

map函數作為數據變換處理的功能函數將原來的數據輸入變換為另外的數據集合，然后設置發布的線程池機制subscribeOn(Schedulers.single())，訂閱的線程池機制observeOn(Schedulers.computation())，最后添加數據訂閱函數，也就是業務系統需要實現另外一個地方，從而實現數據的自定義處理消費。

rxjava2支持的lambda語法

• 創建操作符：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

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

RxJava2 源碼解析

閱讀源碼個人比較喜歡帶著疑惑去看，這樣與目標有方向。接下來的分析以Flowable為例，這里所有的例子都是按照Flowable為例，因為Flowable在實際項目中比Observable可能用的多，因為實際場景中數據生產速度和數據消費速度都會有一定的不一致甚至數據生產速度遠大于數據消費速度。

數據發布和訂閱

首先從數據訂閱者開始，點進源碼看進一步解析，里面有很多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);//調用核心的訂閱方法subscribe(ls);return ls;}public final void subscribe(FlowableSubscriber<? super T> s) {ObjectHelper.requireNonNull(s, "s is null");try {//注冊一些鉤子這里對此不進行講解，主要不是核心方法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");//核心訂閱方法，從名字也能讀出是指訂閱實際調用處//不同的數據產生類也就是實現Flowable抽象類的類//比如FlowableCreate，FlowSingle，FlowMap等等去實現自己的實際方法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)方法進行剖析。

public void subscribeActual(Subscriber<? super T> t) {BaseEmitter<T> emitter;//根據不同的回壓模式選擇不一樣的數據發射類//神奇的回壓模式其實本質上就是一個個數據發射-消費模式switch (backpressure) {case MISSING: {emitter = new MissingEmitter<T>(t);break;}//...default: {emitter = new BufferAsyncEmitter<T>(t, bufferSize());break;}}//回調注冊的FlowableSubscriber的onSubscribe方法//這里非常重要，因為這里涉及了rxjava特有的 request請求再消費數據的模式//也就是說如果沒有request數據，那么就不會調用數據發射（發布）者的onNext方法，//那么數據訂閱者也就不會消費到數據t.onSubscribe(emitter);try {//回調注冊的FlowableOnSubscribe<T> source的subscribe方法//這個source其實就是在創建Flow流時注冊的數據產生類，進一步驗證了上文中//提及的其需要實現FlowableOnSubscribe<T>接口source.subscribe(emitter);} catch (Throwable ex) {Exceptions.throwIfFatal(ex);emitter.onError(ex);}}//重點分析BufferAsyncEmitter這個類，看字面意思這是一個switch的默認選擇類，//但其實它是回壓策略為BUFFER時的數據發射類//首先這個類的構造函數具有兩個參數，很明顯這是 actul就是前面的t這個變量，也就是//注冊的數據消費（訂閱）者，capacityHint則是設置容量大小的，默認是128，如果需要擴大需要//自行設置環境變量 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 是存儲元素的隊列，也就是buffer的核心存儲。// 當我們開始向下游發送數據的時候首先存入隊列，然后下面的drain則是進行核心的queue.offer(t);drain();}//核心的類void drain() {//關鍵的地方 解決生產速率和消費速率不一致的關鍵地方，也是我們寫并發程序值得借鑒的地方。//當數據的產生者（發布）頻繁調用onNext方法時，這里產生并發調用關系，wip變量是atomic變量，//當第一次執行drain函數時，為0繼續執行后面的流程，當快速的繼續調用onNext方法時，wip不為0然后返回//那么后面的流程我們其實已經很大概率會猜測到應該是去取隊列的數據然后做一些操作if (wip.getAndIncrement() != 0) {return;}int missed = 1;//這里的downstream其實就是注冊的數據訂閱者，它是基類BaseEmitter的變量，前面初始化時調用了基類的構造函數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;//取隊列中的數據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;}//此處回調訂閱者的onNext方法去真正的執行數據實例程序//到此數據從產生到消費其生命周期已經走完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) {//標記已經消費的個數BackpressureHelper.produced(this, e);}//前面說過wip會原子性的增加，而且是每調用一次onNext增加一次//missed從其名解釋是指錯過的意思，個人理解是錯過消費的數據個數，錯過消費//的意思其實就是指沒有進行a.onNext數據消費處理的數據missed = wip.addAndGet(-missed);if (missed == 0) {//如果沒有錯過的數據也就是全部都消費完那就跳出for循環//此處for循環方式和JUC源碼中Doug Lea的做法都有類似之處break;}}}

操作符與線程池機制原理剖析

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

回歸主題，當我們使用操作符和線程池機制的時候做法都是在數據發布者后面進行相應的函數操作：

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

那么為何這么做，接下來我們進行源碼分析：

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)); }

這里是實例方法調用，傳進了this對象這個很關鍵，這里其實就是我們前面提到的裝修者模式，持有上游對象也就是數據源source的引用。

以FlowableSubscribeOn為例進行分析，這個類經常會用到，因為其內部設置了線程池的機制所以在實際使用項目中會大量使用，那么是如何做到線程池方式的呢？進一步利用源碼進行分析。

2.裝飾者的內部代碼分析

以subscribeOn 為例:

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

本文總結

筆者喜歡總結，總結意味著我們反思和學習前面的知識點，應用點以及自身的不足。

• 設計模式：觀察者模式和裝修者模式
• 并發處理技巧：回壓策略（其實本質是緩存）的實現原理以及細節點

訂閱最新文章，歡迎關注我的公眾號

總結

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

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。