反應式編程 RxJava 設計原理解析

本文首發於 vivo互聯網技術 微信公衆號 
連接： https://mp.weixin.qq.com/s/duO1pAfaKUI2_x_GVvZHMg
做者：Yunjie Ma

1、ReactiveX 與 RxJava

ReactiveX 的全稱爲Reactive Extension，通常縮寫爲 Rx，即咱們日常所說的反應式編程。其設計原理主要使用了觀察者模式，區分數據的生產者和消費者，經過事件流的方式進行數據的異步處理。

RxJava 是 ReactiveX Java語言的實現，其編程體驗與Java 8中的函數式編程和流（Stream）有很大的類似之處，在掌握了Java8的相關知識後，你能夠很輕鬆的就上手使用 RxJava。

本篇文章主要聚焦對RxJava中幾種主要的設計模式的理解，經過梳理Observable的相關類圖以及講解這些類之間的關係，讓你們可以更清晰的理解RxJava中事件驅動的工做原理。

2、RxJava中的概念

首先咱們寫一個簡單的RxJava的程序，把數組中的元素做爲事件發送，最終由消費者打印在控制檯：

咱們以這段簡單的代碼爲基礎，講解下貫穿整個ReactiveX設計的四個概念：觀察者，被觀察者，事件，訂閱。

• 觀察者：對事件進行響應的對象，也能夠稱做消費者，在上述的代碼中，subscirbe方法的參數是一個Consumer對象，該對象後續會被包裝成一個LambdaObserver對象，即爲這段代碼中的觀察者（消費者）。
• 被觀察者：產生事件的對象，也能夠稱做生產者，在上述代碼中，Observable.fromArray(...)返回的是一個Observable對象，即爲這段程序的被觀察者（生產者）。
• 事件：RxJava中存在四種事件流：onSubscribe（訂閱事件），onNext（正常事件），onError（異常事件），onComplete（完成事件）。在上述代碼中，是將數組中的元素做爲onNext事件中的數據進行發送。
• 訂閱：建立觀察者與被觀察者之間觀察關係，對應着上述代碼中的subscribe()方法。RxJava的事件驅動模型是一種「拉模型」，在觀察者沒有進行事件訂閱以前是不會有事件產生的，只有觀察者進行訂閱後，纔會觸發被觀察者生產事件。

對上述代碼進行時序分析，能夠清晰的看出這一段代碼的運行過程，最終由FromArrayDisposable生產了onNext和onComplete事件，並通知Observer進行消費。

與此同時，咱們也看到，簡單的一行代碼，居然涉及這麼多類的交互，若是增長一些其餘的操做符，咱們對整個程序把控起來就沒那麼容易了，下面咱們將經過分析RxJava中的一些主要的設計模式，剖析類與類的關聯關係，來更清晰地理解RxJava的工做原理。

3、 集大成者Observable

在整個數據處理的過程當中，Observable能夠說是最重要的一個對象。從上面的時序圖能夠看出，客戶端（消息的生產者或者消費者）只和Observable進行交互，觀察者和被觀察者之間關係的建立也是由Observable去實現，而不用咱們顯示的編碼實現，這大大下降了咱們使用觀察者模式的成本。

那麼Observable主要有哪些做用呢，咱們首先來看下和Observable相關的類圖：

從圖中咱們能夠看出：

• Observable實現了ObservableSource接口，從字面意思就能夠理解，這是一個提供觀察能力的接口，因此Observable的一大能力是供觀察者進行事件訂閱，而進行事件訂閱的方法實現就是調用Observable的subscribe()方法
• Observable是一個抽象類，它提供了subscribeActual模板方法供子類實現，從源碼中能夠看出，Observable的subscribe()方法最終會委託子類的subscribeActual()方法實現，這個方法會創建生產者與消費者之間的關聯關係。
• 除此以外，Observable仍是一個工廠類，它提供了靜態方法fromArray()、create()等用來建立具體的可觀察對象，同時還提供了flatMap()、concatMap()等操做方法對可觀察對象進行包裝。

Observable的存在讓生產者和消費者徹底的解耦了，生產者只需關注本身生成何種Observable對象，而消費者也只需關注本身觀察的是哪一種Observable。

在實際的應用中，Rxjava已經提供了各類各樣的操做符供咱們使用，生產者只須要調用Observable中相應的方法便可以生成所需的可觀察對象，供消費者進行事件訂閱。消費者只需調用可觀察對象的subscribe()方法便可與生產者創建觀察關係，極其方便。

4、 真實的觀察

觀察者模式是RxJava設計的核心思想，在觀察者模式中老是存在觀察的對象和被觀察的對象，從上文的解析中也能夠看出Observable更多的是一個控制器的做用，而並不是真正的事件的來源。那麼在RxJava中，什麼纔是真正的生產者，什麼纔是真正的消費者呢。

咱們來分析下如下三種常見的Observable：

先簡單介紹下這幾個Observable的做用，fromArray的做用是將數組中的元素做爲onNext事件發送，create的做用是發送自定義事件，just的做用是發送單個事件。

上一小節有講到實際的訂閱行爲是由各個Observable類中subscribeActual()方法實現的，咱們來看下這三個類的subscribeActual()方法。

除去細枝末節，這三個方法均可以分紅如下三步

1. 建立被觀察者對象，並傳入觀察者observer，創建二者的關聯關係；
2. 觸發onSubscribe事件，觀察者響應該事件；
3. 進行事件的拉取，咱們能夠進入到d.run()，source.subscribe(parent)，sd.run()這些方法的內部看一些，能夠看到這些方法就是在發送onNext(),onError(),onComplete()等事件。

下圖是整個流程中的相關類圖。實際事件的發送者是FromArrayDisposable等對象，而實際的觀察者，則是一個實現了Observer接口的實體類。若是咱們在subscribe時傳入的是一個lambda表達式，以後會被包裝成一個默認的LambdaObserver對象，進行事件消費。

5、 包裝的必要

RxJava中提供了豐富的操做符，好比flatMap，concatMap等能夠對事件轉換，subscribeOn，observableOn等能夠對生產和消費的線程進行控制。這些操做符實際上調用了Observable中的包裝方法對原有的可觀察對象進行包裝，返回了一個加強了的可觀察對象。

操做符種類繁多，在這就不一一舉例，咱們以flatMap爲例，分析一下這些操做符是如何工做的。

首先，flatMap操做會返回一個ObservableFlatMap對象，在建立這個對象時，會將原始的Observable對象做爲構造函數的參數傳入。

查看其核心方法subscribeActual，

能夠看到這一類對象的subscribeActual方法和上一節中的方法不太同樣，這裏面並無去實際的建立觀察關係，而是作了兩件事：

1. 對觀察者進行加強，將其包裝成爲MergeObserver對象，由其對產生的時間進行響應。
2. 再調用source的subscribe方法，這裏source就是前面構造函數中傳入的Observable對象，由其再進行觀察關係的創建。
   下圖是RxJava中裝飾器模式的相關類圖：全部的包裝類都繼承了AbstractObservableWithUpstream類，該抽象類有一個類型爲ObservableSource的成員函數，用來持有被裝飾的對象。

Observable是支持鏈式操做的，就和Java 8中的Stream同樣，咱們來考慮這樣一行代碼。

咱們在分析上面這串代碼時，必定會凌亂很是，在看源碼時也會看到前面忘掉後面，可是若是咱們對RxJava的包裝流程足夠了解的話，就能夠很輕鬆的對上述代碼進行分析。

6、 小結

RxJava的封裝足夠強大，可讓咱們很方便的進行使用和擴展，但這也給咱們理解其真實的工做原理帶來了難度，若是咱們對整個事件的處理過程處於只知其一;不知其二的狀態，那咱們就沒法從容的對服務進行異步編排，在實際開發過程當中也難以發現問題的根源。

本文主要分析了RxJava中主要的設計模式，其中有模板模式、工廠模式、觀察者模式、裝飾器模式，理解了這些設計模式，理解了RxJava中類與類的關係，咱們就可以對整個事件的處理流程瞭然於胸，分析代碼時也可以事半功倍。