Rxjava2(一)、基礎概念及使用

Android進階系列之第三方庫知識點整理。

知識點總結，整理也是學習的過程，若有錯誤，歡迎批評指出。

寫這篇文章的時候，Rxjava3已經出來了，兩者差異不是特別大，因此仍是基於Rxjava2進行總結，後續會花時間來總結Rxjava3的一些改變。

基礎概念

一、什麼是Rx

ReactiveX是Reactive Extensions的縮寫，通常簡寫爲Rx，最初是LINQ的一個擴展，Rx是由由微軟開發並開源的一個編程模型，目標是提供一致的編程接口，幫助開發者更方便的處理異步數據流，Rx庫支持不少語言，而 Rxjava 是Rx庫基於Java語言提供的函數庫。

官網顯示支持的語言列表：

官網Rx支持的語言

二、觀察者模式

觀察者模式（Observer），又叫發佈-訂閱模式（Publish/Subscribe），定義對象間一種一對多的依賴關係，使得每當一個對象改變狀態，則全部依賴於它的對象都會獲得通知並自動更新,觀察者模式屬於行爲型模式。

我須要打電話跟小明說他被戴綠帽這事，可是我又怕打電話過去他可能在忙，因此我直接給他發信息，讓他有時間了再給我回電話，這樣，當小明有時間了可能就會給我回電話，他給我回電話，我再含着淚的告訴他要堅強。

在上面這個例子中，小明就是被觀察者，我是觀察者，我發短信告訴他有時間回我電話這個操做就是訂閱，這樣我跟他有了依賴關係，當他有時間後，就會回我帶電話，我這邊響應式的執行後續的操做。

三、響應式(Reactive Programming)編程

通常來講，編程就是爲了處理問題，解決問題就能夠有不一樣的視角和思路，其中具備普適性的模式就會歸結一種編程思想，咱們常說的面向對象，面向過程都是一種編程思想。

響應式編程是一種經過異步和數據流來構建事務關係的編程模式

這裏麪包含的兩個核心概念：

異步：不須要等待處理完成便馬上返回，經過回調將處理結果帶回。能提升吞吐量，提升性能和效率。

數據流：就是按照時間線的事件序列，任何東西均可以看作是一個stream。在這個基礎上經過函數來過濾(filter)、選擇(select)、變換(transform)、結合(combine)這些stream。經過定義事件處理器，來異步捕獲感興趣的stream，即監聽（訂閱）stream，也叫就是註冊觀察者。

再對比咱們前面的觀察者模式看，觀察者模式中被觀察者和觀察者之間經過訂閱產生依賴，訂閱這層關係至關於在兩者以前挖了一道水渠，被觀察者是上游，觀察者是下游，這樣被觀察者發生改變時候，會將數據流源源不斷的發送到產生依賴關係的觀察者，一般觀察者和被觀察者是處於同一個線程中，可是響應式編程重點在異步上，即上游和下游能夠在不一樣的線程中工做。

咱們能夠看一下官網的動圖：

四、Rxjava概念及簡單使用

Rxjava採用了響應式編程模式，它在觀察者模式基礎上進行擴展，方便咱們的異步操做，觀察者與被觀察者產生訂閱關係後，被觀察者將數據流發送給觀察者的時候，Rxjava提供了豐富的操做符來對數據流進行過濾(filter)、選擇(select)、變換(transform)、結合(combine)等等操做，而後再將處理後的數據發送給觀察者，又因爲其特別棒的鏈式調用，讓邏輯在很複雜的狀況下，也能很清晰的經過代碼表達出來。

咱們先引入rxjava

implementation "io.reactivex.rxjava2:rxjava:2.2.8"
implementation 'io.reactivex.rxjava2:rxandroid:2.1.1'
複製代碼

直接先來一個簡單的例子：

  // 經過 Observable 建立一個被觀察者
  Observable mObservable = Observable.create(new ObservableOnSubscribe() {
         @Override
         public void subscribe(ObservableEmitter emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();
            }
        });

        // 經過Observer建立一個觀察者
        Observer mObserver = new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                LogUtil.d(TAG + "--onSubscribe");
            }

            @Override
            public void onNext(Integer o) {
                LogUtil.d(TAG + "--onNext  o=" + o);
            }

            @Override
            public void onError(Throwable e) {
                LogUtil.d(TAG + "--onError");
            }

            @Override
            public void onComplete() {
                LogUtil.d(TAG + "--onComplete");
            }
        };
        // 產生訂閱關係
        mObservable.subscribe(mObserver);
複製代碼

再看看Rxjava強大的鏈式調用：

Observable.create(new ObservableOnSubscribe() {
       @Override
       public void subscribe(ObservableEmitter emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();
            }
            // 這裏直接經過subscribe將兩者鏈式調用起來
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                LogUtil.d(TAG + "--Observer--onSubscribe");
            }

            @Override
            public void onNext(Integer o) {
                LogUtil.d(TAG + "--Observer--onNext  o=" + o);
            }

            @Override
            public void onError(Throwable e) {
                LogUtil.d(TAG + "--Observer--onError");
            }

            @Override
            public void onComplete() {
                LogUtil.d(TAG + "--Observer--onComplete");
            }
        });
複製代碼

咱們前面說，RxJava是拓展於觀察者模式的響應式編程，因此在Rxjava中，就有觀察者、被觀察者以及兩者經過訂閱產生的依賴關係。

經過上面代碼，能夠看到：

被觀察者 ——————> Observable

觀察者 ——————> Observer

訂閱關係：Observable ————> subscribe <———— Observer

在上游的 Observable （被觀察者）中經過 emitter（發射器）就能夠不斷地給下游的 Observer （觀察者）傳送數據流，咱們在上游發射了 一、二、3還有發射了一個 onComplete 事件。

咱們經過日誌能夠看到，在下游的 Observer （觀察者）成功的接收到了上游發送的數字一、二、3包括經過調用 onComplete()方法來接收上游發送的 onComplete 事件。

經過上面一個簡單的例子，咱們能夠總結出如下幾點

4.一、解耦

咱們能夠看到被觀察者和觀察者都是獨立存在的。

Observable（被觀察者）：

  // 經過 Observable 建立一個被觀察者
  Observable mObservable = Observable.create(new ObservableOnSubscribe() {
         @Override
         public void subscribe(ObservableEmitter emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();
            }
        });
複製代碼

Observer（觀察者）：

 // 經過Observer建立一個觀察者
        Observer mObserver = new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                LogUtil.d(TAG + "--onSubscribe");
            }

            @Override
            public void onNext(Integer o) {
                LogUtil.d(TAG + "--onNext  o=" + o);
            }

            @Override
            public void onError(Throwable e) {
                LogUtil.d(TAG + "--onError");
            }

            @Override
            public void onComplete() {
                LogUtil.d(TAG + "--onComplete");
            }
        };
複製代碼

能夠看到，兩者各自獨立存在，各幹各的事，只是經過 subscribe 來創建依賴關係，這樣的話，能夠很好地實現代碼的複用，好比咱們能夠建立多個Observer（觀察者），和Observable（被觀察者）之間經過訂閱創建依賴關係後，就能實現Observable（被觀察者）發送的數據流被多個Observer（觀察者）接收。

4.二、數據流轉換（操做符的使用）

第2、Observable（被觀察者）經過 emitter（發射器）將數據以流的形式發送給Observer（觀察者），前面講了，既然是以流的形式，那咱們是否是能夠在觀察者接收數據以前，對數據進行處理呢？答案是確定的，這也是Rxjava的強大之一，它能夠經過各類操做符來對數據進行處理，咱們先講一個簡單的來。

這裏只是簡單的演示一些Rxjava對數據流的操做，後面會具體講操做符，圖片中能夠看到，咱們被觀察者明明發射的是一個int型的數值100，可是經過map操做符進行數據流進行轉換後，下游的消費者接收到的居然是String的，這裏只是一個小Demo,在Rxjava中，還有各類很強大的操做符來對數據進行各類變態式的處理。

上游被觀察者Observable發射數據，下游除了使用Observer來接收數據外，還可使用消費者Consumer來接收數據，能夠看Consumer只有一個accept方法。

先貼一下Rxjava強大的操做符列表，是否是不少？不要被嚇到，越多因此咱們能夠越方便的對數據流進行各類各樣的操做，這個後面會單獨再講解。

4.三、線程切換

咱們前面說了，Rxjava能更好的異步操做，咱們先用上面的例子打印一下他們當前所在的線程狀況。

結果：

能夠到，統一戰線的都在主線程，咱們來看看，怎麼實現異步。

一樣是那個例子，咱們添加了下面這兩個代碼

subscribeOn(Schedulers.io())
observeOn(AndroidSchedulers.mainThread())
複製代碼

subscribeOn : 指定上游執行的線程

observeOn：指定觀察者或者消費者執行線程

結果：

代碼中，咱們指定上游在io線程，而下游在主線程執行，就這麼簡單的兩行代碼，就實現了線程切換，實現了異步操做，爽不爽？

固然，咱們不光能夠指定io線程，Schedulers還提供了其餘的線程供咱們選擇。

咱們簡單看一個Single線程的源碼：

是否是很熟悉？就是經過線程池來建立了一個單線程，其餘相似，都是經過線程池來建立，您能夠簡單去分析一下，固然，若是你對線程池還不熟悉的話，這篇 多線程(三)、線程池 ThreadPoolExecutor 知識點總結 可能適合你看看。

總結

Rxjava的東西還挺多的，這篇先簡單的走了一遍Rxjava的基礎概念還有簡單的使用，關於Rxjava裏面還有其餘的觀察者模式的建立，背壓，各類操做符的使用、實戰等等騷操做，下篇繼續。