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

這篇文章給大家介紹如何解析反應式編程 RxJava 設計原理，內容非常詳細，感興趣的小伙伴們可以參考借鑒，希望對大家能有所幫助。

一、ReactiveX 與 RxJava

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

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

本篇文章主要聚焦對RxJava中幾種主要的設計模式的理解，通過梳理Observable的相關類圖以及講解這些類之間的關系，讓大家能夠更清晰的理解RxJava中事件驅動的工作原理。

二、RxJava中的概念

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

從圖中我們可以看出：

• Observable實現了ObservableSource接口，從字面意思就可以理解，這是一個提供觀察能力的接口，所以Observable的一大能力是供觀察者進行事件訂閱，而進行事件訂閱的方法實現就是調用Observable的subscribe()方法

• Observable是一個抽象類，它提供了subscribeActual模板方法供子類實現，從源碼中可以看出，Observable的subscribe()方法最終會委托子類的subscribeActual()方法實現，這個方法會建立生產者與消費者之間的關聯關系。

• 除此之外，Observable還是一個工廠類，它提供了靜態方法fromArray()、create()等用來創建具體的可觀察對象，同時還提供了flatMap()、concatMap()等操作方法對可觀察對象進行包裝。

Observable的存在讓生產者和消費者完全的解耦了，生產者只需關注自己生成何種Observable對象，而消費者也只需關注自己觀察的是哪種Observable。

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

四、 真實的觀察

觀察者模式是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對象，進行事件消費。

五、 包裝的必要

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的包裝流程足夠了解的話，就可以很輕松的對上述代碼進行分析。

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

關于如何解析反應式編程 RxJava 設計原理就分享到這里了，希望以上內容可以對大家有一定的幫助，可以學到更多知識。如果覺得文章不錯，可以把它分享出去讓更多的人看到。