Android ：一文帶你真正入門神祕的Rxjava

時間  2019-11-30
標籤 android 一文 真正 入門 神祕 rxjava 欄目 Android 简体版
原文   原文鏈接

前言

  • Rxjava因爲其基於事件流的鏈式調用、邏輯簡潔 & 使用簡單的特色，深受各大 Android開發者的歡迎。

Github截圖

  • 本文主要：
    1. 面向 剛接觸Rxjava的初學者
    2. 提供了一份 清晰、簡潔、易懂的Rxjava入門教程

    涵蓋 基本介紹、原理 & 具體使用等java

    1. 解決的是初學者不理解Rxjava原理 & 不懂得如何使用的問題

但願大家會喜歡。react

  1. 本文主要基於Rxjava 2.0
  2. 若是讀者還沒學習過Rxjava 1.0也不要緊，由於Rxjava 2.0只是在Rxjava 1.0上增長了一些新特性，本質原理 & 使用基本相同
  3. 接下來的時間，我將持續推出 Android中 Rxjava 2.0 的一系列文章，包括原理、操做符、應用場景、背壓等等 ，有興趣能夠繼續關注Carson_Ho的安卓開發筆記！！

示意圖

目錄

示意圖

1. 定義

  • RxJava 在 GitHub 的介紹：
RxJava：a library for composing asynchronous and event-based programs using observable sequences for the Java VM
// 翻譯：RxJava 是一個在 Java VM 上使用可觀測的序列來組成異步的、基於事件的程序的庫
複製代碼
  • 總結：RxJava 是一個 基於事件流、實現異步操做的庫

2. 做用

實現異步操做android

相似於 Android中的 AsyncTask 、Handler做用git

3. 特色

因爲 RxJava的使用方式是：基於事件流的鏈式調用，因此使得 RxJava：github

  • 邏輯簡潔
  • 實現優雅
  • 使用簡單

更重要的是，隨着程序邏輯的複雜性提升，它依然可以保持簡潔 & 優雅數組

4. 原理

4.1 生活例子引入

  • 我用一個生活例子引入 & 講解 Rxjava原理： 顧客到飯店吃飯

示意圖

流程圖

4.2 Rxjava原理介紹

  • Rxjava原理 基於 一種擴展的觀察者模式bash

  • Rxjava的擴展觀察者模式中有4個角色：微信

角色 做用 類比
被觀察者（Observable） 產生事件 顧客
觀察者（Observer） 接收事件，並給出響應動做 廚房
訂閱（Subscribe） 鏈接 被觀察者 & 觀察者 服務員
事件（Event） 被觀察者 & 觀察者 溝通的載體 菜式
  • 具體原理

請結合上述 顧客到飯店吃飯 的生活例子理解： 異步

示意圖

流程圖

即RxJava原理可總結爲：被觀察者 （Observable） 經過 訂閱（Subscribe） 按順序發送事件 給觀察者 （Observer）， 觀察者（Observer） 按順序接收事件 & 做出對應的響應動做。具體以下圖：async

示意圖

至此，RxJava原理講解完畢。

5. 基本使用

  • 本文只關注 RxJava的基本使用，更深刻的RxJava使用請繼續關注Carson_Ho的安卓開發筆記
  • Rxjava的使用方式有兩種：
    1. 分步驟實現：該方法主要爲了深刻說明Rxjava的原理 & 使用，主要用於演示說明
    2. 基於事件流的鏈式調用：主要用於實際使用

5.1 方式1：分步驟實現

5.1.1 使用步驟

示意圖

5.1.2 步驟詳解
步驟1：建立被觀察者 （Observable ）& 生產事件
  • 即 顧客入飯店 - 坐下餐桌 - 點菜
  • 具體實現
// 1. 建立被觀察者 Observable 對象
        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
          // create() 是 RxJava 最基本的創造事件序列的方法
          // 此處傳入了一個 OnSubscribe 對象參數
          // 當 Observable 被訂閱時，OnSubscribe 的 call() 方法會自動被調用，即事件序列就會依照設定依次被觸發
          // 即觀察者會依次調用對應事件的複寫方法從而響應事件
          // 從而實現被觀察者調用了觀察者的回調方法 & 由被觀察者向觀察者的事件傳遞，即觀察者模式

        // 2. 在複寫的subscribe（）裏定義須要發送的事件
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                // 經過 ObservableEmitter類對象產生事件並通知觀察者
                // ObservableEmitter類介紹
                    // a. 定義：事件發射器
                    // b. 做用：定義須要發送的事件 & 向觀察者發送事件
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();
            }
        });

<--擴展：RxJava 提供了其餘方法用於 建立被觀察者對象Observable -->
// 方法1：just(T...)：直接將傳入的參數依次發送出來
  Observable observable = Observable.just("A", "B", "C");
  // 將會依次調用：
  // onNext("A");
  // onNext("B");
  // onNext("C");
  // onCompleted();

// 方法2：from(T[]) / from(Iterable<? extends T>) : 將傳入的數組 / Iterable 拆分紅具體對象後，依次發送出來
  String[] words = {"A", "B", "C"};
  Observable observable = Observable.from(words);
  // 將會依次調用：
  // onNext("A");
  // onNext("B");
  // onNext("C");
  // onCompleted();

複製代碼
步驟2：建立觀察者 （Observer ）並 定義響應事件的行爲
  • 即 開廚房 - 肯定對應菜式
  • 發生的事件類型包括：Next事件、Complete事件 & Error事件。具體以下：

示意圖

  • 具體實現
<--方式1：採用Observer 接口 -->
        // 1. 建立觀察者 （Observer ）對象
        Observer<Integer> observer = new Observer<Integer>() {
        // 2. 建立對象時經過對應複寫對應事件方法 從而 響應對應事件

            // 觀察者接收事件前，默認最早調用複寫 onSubscribe（）
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG, "開始採用subscribe鏈接");
            }
            
            // 當被觀察者生產Next事件 & 觀察者接收到時，會調用該複寫方法 進行響應
            @Override
            public void onNext(Integer value) {
                Log.d(TAG, "對Next事件做出響應" + value);
            }

            // 當被觀察者生產Error事件& 觀察者接收到時，會調用該複寫方法 進行響應
            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "對Error事件做出響應");
            }
          
            // 當被觀察者生產Complete事件& 觀察者接收到時，會調用該複寫方法 進行響應
            @Override
            public void onComplete() {
                Log.d(TAG, "對Complete事件做出響應");
            }
        };

<--方式2：採用Subscriber 抽象類 -->
// 說明：Subscriber類 = RxJava 內置的一個實現了 Observer 的抽象類，對 Observer 接口進行了擴展

// 1. 建立觀察者 （Observer ）對象
Subscriber<String> subscriber = new Subscriber<Integer>() {

// 2. 建立對象時經過對應複寫對應事件方法 從而 響應對應事件
            // 觀察者接收事件前，默認最早調用複寫 onSubscribe（）
            @Override
            public void onSubscribe(Subscription s) {
                Log.d(TAG, "開始採用subscribe鏈接");
            }

            // 當被觀察者生產Next事件 & 觀察者接收到時，會調用該複寫方法 進行響應
            @Override
            public void onNext(Integer value) {
                Log.d(TAG, "對Next事件做出響應" + value);
            }

            // 當被觀察者生產Error事件& 觀察者接收到時，會調用該複寫方法 進行響應
            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "對Error事件做出響應");
            }

            // 當被觀察者生產Complete事件& 觀察者接收到時，會調用該複寫方法 進行響應
            @Override
            public void onComplete() {
                Log.d(TAG, "對Complete事件做出響應");
            }
        };


<--特別注意：2種方法的區別，即Subscriber 抽象類與Observer 接口的區別 -->
// 相同點：兩者基本使用方式徹底一致（實質上，在RxJava的 subscribe 過程當中，Observer老是會先被轉換成Subscriber再使用）
// 不一樣點：Subscriber抽象類對 Observer 接口進行了擴展，新增了兩個方法：
    // 1. onStart()：在還未響應事件前調用，用於作一些初始化工做
    // 2. unsubscribe()：用於取消訂閱。在該方法被調用後，觀察者將再也不接收 & 響應事件
    // 調用該方法前，先使用 isUnsubscribed() 判斷狀態，肯定被觀察者Observable是否還持有觀察者Subscriber的引用，若是引用不能及時釋放，就會出現內存泄露
複製代碼
步驟3：經過訂閱（Subscribe）鏈接觀察者和被觀察者
  • 即 顧客找到服務員 - 點菜 - 服務員下單到廚房 - 廚房烹調
  • 具體實現
observable.subscribe(observer);
 // 或者 observable.subscribe(subscriber)；
複製代碼
  • 擴展說明
<-- Observable.subscribe(Subscriber) 的內部實現 -->

public Subscription subscribe(Subscriber subscriber) {
    subscriber.onStart();
    // 步驟1中 觀察者  subscriber抽象類複寫的方法，用於初始化工做
    onSubscribe.call(subscriber);
    // 經過該調用，從而回調觀察者中的對應方法從而響應被觀察者生產的事件
    // 從而實現被觀察者調用了觀察者的回調方法 & 由被觀察者向觀察者的事件傳遞，即觀察者模式
    // 同時也看出：Observable只是生產事件，真正的發送事件是在它被訂閱的時候，即當 subscribe() 方法執行時
}
複製代碼

5.2 方式2：優雅的實現方法 - 基於事件流的鏈式調用

  • 上述的實現方式是爲了說明Rxjava的原理 & 使用
  • 在實際應用中，會將上述步驟&代碼連在一塊兒，從而更加簡潔、更加優雅，即所謂的 RxJava基於事件流的鏈式調用
// RxJava的鏈式操做
        Observable.create(new ObservableOnSubscribe<Integer>() {
        // 1. 建立被觀察者 & 生產事件
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();
            }
        }).subscribe(new Observer<Integer>() {
            // 2. 經過經過訂閱（subscribe）鏈接觀察者和被觀察者
            // 3. 建立觀察者 & 定義響應事件的行爲
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG, "開始採用subscribe鏈接");
            }
            // 默認最早調用複寫的 onSubscribe（）

            @Override
            public void onNext(Integer value) {
                Log.d(TAG, "對Next事件"+ value +"做出響應"  );
            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "對Error事件做出響應");
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "對Complete事件做出響應");
            }

        });
    }
}

注：總體方法調用順序：觀察者.onSubscribe（）> 被觀察者.subscribe（）> 觀察者.onNext（）>觀察者.onComplete() 
複製代碼

這種 基於事件流的鏈式調用，使得RxJava：

  • 邏輯簡潔
  • 實現優雅
  • 使用簡單

更重要的是，隨着程序邏輯的複雜性提升，它依然可以保持簡潔 & 優雅。因此，通常建議使用這種基於事件流的鏈式調用方式實現RxJava。

特別注意

RxJava 2.x 提供了多個函數式接口 ，用於實現簡便式的觀察者模式。具體以下：

示意圖

以 Consumer爲例：實現簡便式的觀察者模式

Observable.just("hello").subscribe(new Consumer<String>() {
            // 每次接收到Observable的事件都會調用Consumer.accept（）
            @Override
            public void accept(String s) throws Exception {
                System.out.println(s);
            }
        });
複製代碼

6. 實例說明

我將用一個實際工程實例來演示 Rxjava的使用

6.1 方式1：分步驟實現

步驟1：加入依賴

compile 'io.reactivex.rxjava2:rxjava:2.0.1'
    compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
複製代碼

步驟2：直接在MainActivity.java中實現下述步驟

  1. 建立被觀察者 （Observable ）& 生產事件
  2. 建立觀察者 （Observer ）並 定義響應事件的行爲
  3. 經過訂閱（Subscribe）鏈接觀察者和被觀察者
public class MainActivity extends AppCompatActivity {

    private static final String TAG = "Rxjava";

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);



// 步驟1：建立被觀察者 Observable & 生產事件
// 即 顧客入飯店 - 坐下餐桌 - 點菜
        
        //  1. 建立被觀察者 Observable 對象
        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
            // 2. 在複寫的subscribe（）裏定義須要發送的事件
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                // 經過 ObservableEmitter類對象產生事件並通知觀察者
                // ObservableEmitter類介紹
                    // a. 定義：事件發射器
                    // b. 做用：定義須要發送的事件 & 向觀察者發送事件
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();
            }
        });

// 步驟2：建立觀察者 Observer 並 定義響應事件行爲
// 即 開廚房 - 肯定對應菜式
        
        Observer<Integer> observer = new Observer<Integer>() {
            // 經過複寫對應方法來 響應 被觀察者
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG, "開始採用subscribe鏈接");
            }
            // 默認最早調用複寫的 onSubscribe（）

            @Override
            public void onNext(Integer value) {
                Log.d(TAG, "對Next事件"+ value +"做出響應"  );
            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "對Error事件做出響應");
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "對Complete事件做出響應");
            }
        };

        
        // 步驟3：經過訂閱（subscribe）鏈接觀察者和被觀察者
        // 即 顧客找到服務員 - 點菜 - 服務員下單到廚房 - 廚房烹調
        observable.subscribe(observer);
複製代碼
  • 測試結果

示意圖

6.2 方式2：基於事件流的鏈式調用方式

public class MainActivity extends AppCompatActivity {

    private static final String TAG = "Rxjava";

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);

// RxJava的流式操做
        Observable.create(new ObservableOnSubscribe<Integer>() {
        // 1. 建立被觀察者 & 生產事件
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();
            }
        }).subscribe(new Observer<Integer>() {
            // 2. 經過經過訂閱（subscribe）鏈接觀察者和被觀察者
            // 3. 建立觀察者 & 定義響應事件的行爲
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG, "開始採用subscribe鏈接");
            }
            // 默認最早調用複寫的 onSubscribe（）

            @Override
            public void onNext(Integer value) {
                Log.d(TAG, "對Next事件"+ value +"做出響應"  );
            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "對Error事件做出響應");
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "對Complete事件做出響應");
            }

        });
    }
}
複製代碼

喜歡的麻煩點個star！

7. 額外說明

7.1 觀察者 Observer的subscribe()具有多個重載的方法

public final Disposable subscribe() {}
    // 表示觀察者不對被觀察者發送的事件做出任何響應（但被觀察者仍是能夠繼續發送事件）

    public final Disposable subscribe(Consumer<? super T> onNext) {}
    // 表示觀察者只對被觀察者發送的Next事件做出響應
    public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError) {} 
    // 表示觀察者只對被觀察者發送的Next事件 & Error事件做出響應

    public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete) {}
    // 表示觀察者只對被觀察者發送的Next事件、Error事件 & Complete事件做出響應

    public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete, Consumer<? super Disposable> onSubscribe) {}
    // 表示觀察者只對被觀察者發送的Next事件、Error事件 、Complete事件 & onSubscribe事件做出響應

    public final void subscribe(Observer<? super T> observer) {}
    // 表示觀察者對被觀察者發送的任何事件都做出響應
複製代碼

7.2 可採用 Disposable.dispose() 切斷觀察者 與 被觀察者 之間的鏈接

  • 即觀察者 沒法繼續 接收 被觀察者的事件，但被觀察者仍是能夠繼續發送事件
  • 具體使用
// 主要在觀察者 Observer中 實現
        Observer<Integer> observer = new Observer<Integer>() {
            // 1. 定義Disposable類變量
            private Disposable mDisposable;

            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG, "開始採用subscribe鏈接");
                // 2. 對Disposable類變量賦值
                mDisposable = d;
            }

            @Override
            public void onNext(Integer value) {
                Log.d(TAG, "對Next事件"+ value +"做出響應"  );
                if (value == 2) {
                    // 設置在接收到第二個事件後切斷觀察者和被觀察者的鏈接
                    mDisposable.dispose();
                    Log.d(TAG, "已經切斷了鏈接：" + mDisposable.isDisposed());
                }
            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "對Error事件做出響應");
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "對Complete事件做出響應");
            }
        };
複製代碼
  • 效果圖

示意圖

8. 總結

  • 本文主要對 Rxjava 的入門知識進行講解，包括基本介紹、原理 & 具體使用等
  • 接下來，我將持續推出 Android中 Rxjava 2.0 的一系列文章，包括原理、操做符、應用場景、背壓等等 ，有興趣能夠繼續關注Carson_Ho的安卓開發筆記！！

示意圖

請點贊！由於你的鼓勵是我寫做的最大動力！

歡迎關注carson_ho的微信公衆號

示意圖

示意圖
相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程