RxJava 知識梳理(1) RxJava 基本思想

時間  2019-11-21
標籤 rxjava 知識 梳理 基本 思想 欄目 Java 简体版
原文   原文鏈接

1、基礎概述

RxJava的關鍵是異步，即便隨着程序的邏輯變得複雜，它依然可以保持簡潔。bash

2、API介紹和原理剖析

觀察者模式面向的需求是：A對象（觀察者）對B對象（被觀察者）的某種變化高度敏感，須要在B變化的一瞬間作出反應，觀察者採用註冊Register或者訂閱Subscribe的方式，告訴觀察者，我須要你的某某狀態，並在它變化的時候通知我，在RxJava當中，Observable是被觀察者，Observer就是觀察者。異步

RxJava有四個基本概念：ide

  • Observable：被觀察者。
  • Observer：觀察者。
  • Subscribe：訂閱。
  • Event：事件。

Observable和Observer經過subscribe方法實現訂閱關係，Observable能夠在須要的時候發出事件來通知Observer。函數

RxJava有如下三種事件：ui

  • onNext：普通事件。
  • onCompleted：RxJava不只把每一個事件單獨處理，還會把它們看做一個隊列，當不會再有新的onNext事件發出時，須要觸發onCompleted事件做爲標誌。
  • onError：onCompleted和有且僅有一個，而且是事件序列中的最後一個。

3、基本實現

RxJava的基本實現有如下三點： 1）建立觀察者 - Observerspa

Observer<String> observer = new Observer<String>() {

    @Override
    public void onCompleted() {
        Log.d(TAG, "onCompleted");
    }

    @Override
    public void onError(Throwable e) {
        Log.d(TAG, "onError");
    }

    @Override
    public void onNext(String s) {
        Log.d(TAG, "onNext");
    }
};
複製代碼

除了Observer接口以外，RxJava還內置了一個實現了Observer的抽象類：Subscriber，它對Observer接口進行了一些擴展，實質上在RxJava的subscribe過程當中，Observer也老是被轉換成爲一個Subscriber再使用，他們的區別在與：線程

  • onStart：這是新增的方法，它會在subscribe剛開始，而事件還未發送以前被調用，它老是在subscribe所發生的線程被調用。
  • unsubscribe：這是它實現的另外一個接口Subscription的方法，用於取消訂閱，在這個方法被調用後，Subscriber將再也不接收事件，通常在調用這個方法前，可使用isUnsubscribed判斷一下狀態，Observable在訂閱以後會持有Subscriber的引用，所以不釋放會有內存泄漏的危險。

2）建立被觀察者 - Observable RxJava用create方法來建立一個observable，3d

rx.Observable observable = rx.Observable.create(new rx.Observable.OnSubscribe<String>() {

    @Override
    public void call(Subscriber<? super String> subscriber) {
        subscriber.onNext("Hello World!");
        subscriber.onCompleted();
    }
});
複製代碼

這裏傳入了一個Observable.OnSubscribe<T>對象做爲參數，它會被存儲在返回的Observable對象當中，它的做用至關於一個計劃表，當Observable被訂閱的時候，OnSubscribe的call方法會自動被調用，事件序列被依次觸發。 create是RxJava最基本的創造事件序列的方法，基於這個方法，還提供了一些快捷方法來建立事件隊列：代理

  • just(T...)
Observable observable = Observable.just("Hello", "Hi", "Aloha");
複製代碼
  • from(T[]) / from(Iterable<? extends T>)
String[] words = {"Hello", "Hi", "Aloha"};
Observable observable = Observable.from(words);
複製代碼

3）訂閱 - subscribecode

observable.subscribe(observer);
observable.subscribe(subscriber);
複製代碼

其內部核心的代碼相似於：

public Subscription subscribe(Subscriber subscriber) {
    //準備方法。
    subscriber.onStart();
    //事件發送的邏輯開始執行，這個onSubscribe就是建立Observable時新建的OnSubscribe對象。
    onSubscribe.call(subscriber);
    //把傳入的Subscriber轉換爲Subscription並返回，方便unsubscribe。
    return subscriber;
}
複製代碼

Observable.subscribe方法除了支持傳入Observer和Subscriber，還支持傳入Action0、Action1這樣不完整定義的回調，RxJava會自動根據定義建立出Subscriber。

4、線程控制

在不指定線程的狀況下，RxJava遵循這樣的原則，在哪一個線程調用subscribe，就在哪一個線程產生事件，在哪一個線程產生事件，就在哪一個線程消費事件，若是須要消費線程，那麼就須要用到Scheduler， RxJava內置了幾個Scheduler：

  • Schedulers.immediate：直接在當前線程運行。
  • Schedulers.newThread：老是啓用新線程，並在線程執行操做。
  • Schedulers.io：其內部實現是一個無數量上限的的線程池，能夠重用空閒的線程，不要把計算工做放在io，能夠避免建立沒必要要的線程。
  • Schedulers.computation：使用固定的線程池，大小爲CPU核數。
  • AndroidSchedulers.mainThread：指定的操做將在Android主線程中運行。

對線程控制有如下兩個方法：

  • subscribeOn：指定subscribe發生的線程，即Observable.OnSubscribe被激活時所處的線程，也就是call方法執行時所處的線程。
  • observeOn：指定Subscriber所運行在的線程。

observeOn指定的是Subscriber的線程，而這個Subscriber並不必定是subscribe()參數中的Subscriber，而是observeOn執行時的當前Observable所對應的Subscriber，即它的直接下級Subscriber，也就是它以後的操做所在的線程，所以，若是有屢次切換線程的要求，只要在每一個想要切換線程的位置調用依次observeOn便可。 和observeOn不一樣，subscribeOn只能調用一次，下面咱們來分析一下它的內部實現，首先是subscribeOn的原理： subscribeOn和ObserveOn都作了線程切換的工做：

  • subscribeOn的線程切換髮生在OnSubscribe中，即在它通知上一級的OnSubscribe時，這時事件尚未發送，所以subscribeOn的線程控制能夠從事件發出的開端形成影響。

Paste_Image.png

  • observeOn的線程切換則發生在它內建的Subscriber中，即發生在它即將給下一級Subscriber發送事件時，所以控制的是它後面的線程。

Paste_Image.png

5、變換

變換，就是將事件序列中的對象或整個序列進行加工處理，轉換不一樣的事件或者序列。

5.1 map()

經過FuncX，把參數中的Integer轉換成爲String，是最經常使用的變換，這個變換是發生在subscribeOn所指定的線程當中的。

Subscriber<String> subscriber = new Subscriber<String>() {

    @Override
    public void onCompleted() {

    }

    @Override
    public void onError(Throwable e) {

    }

    @Override
    public void onNext(String s) {
        long nextId = Thread.currentThread().getId();
        Log.d(TAG, "onNext:" + s + ", threadId=" + nextId);
    }
};
Observable<Integer> observable = Observable.create(new Observable.OnSubscribe<Integer>() {
    @Override
    public void call(Subscriber<? super Integer> subscriber) {
        long callId = Thread.currentThread().getId();
        subscriber.onNext(5);
        subscriber.onCompleted();
    }
});
observable.map(new Func1<Integer, String>() {

    @Override
    public String call(Integer integer) {
        long mapId = Thread.currentThread().getId();
        return "My Number is:" + integer;
    }
}).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(subscriber);
複製代碼

其示意圖相似於：

Paste_Image.png

5.2 flatMap

它和map有一個共同點，就是把傳入的參數轉化以後返回另外一個對象，可是和map不一樣的是，flatMap返回的是一個Observable對象，並且它並不直接把這個對象傳給Subscriber，而是經過這個新建的Observable來發送事件，其整個的調用過程：

  • 使用傳入的事件對象建立一個Observable。
  • 激活這個Observable，經過它來發送事件。
  • 每個建立出來的Observable發送的事件，被匯入同一個Observable，它複雜將這些事件同一交給Subscriber的回調方法。
Subscriber<String> subscriber = new Subscriber<String>() {

    @Override
    public void onCompleted() {}

    @Override
    public void onError(Throwable e) {}

    @Override
    public void onNext(String s) {
        Log.d(TAG, "onNext, s=" + s);
    }
};
Observable<List<String>> observable = Observable.create(new Observable.OnSubscribe<List<String>>() {

    @Override
    public void call(Subscriber<? super List<String>> subscriber) {
        List<String> list = new ArrayList<>();
        list.add("First");
        list.add("Second");
        list.add("Third");
        subscriber.onNext(list);
    }
});
observable.flatMap(new Func1<List<String>, Observable<String>>() {
    @Override
    public Observable<String> call(List<String> strings) {
        return Observable.from(strings);
    }
}).subscribe(subscriber);
複製代碼

其示意圖：

Paste_Image.png

6、變換的原理

變換的實質是針對事件序列的處理和再發送，在RxJava的內部，它們是基於同一個基礎的變換方法lift(operator)

//生成了一個新的Observable並返回。
public <R> Observable<R> lift(Operator<? extends R, ? super T> operator) {
    //構造新的Observable時，同時新建了一個OnSubscribe對象。
    return Observable.create(new OnSubscribe<R>() {
        @Override
        public void call(Subscriber subscriber) {
            Subscriber newSubscriber = operator.call(subscriber);
            newSubscriber.onStart();
            //原始的onSubscribe。
            onSubscribe.call(newSubscriber);
        }
    });
}
複製代碼

示意圖：

Paste_Image.png

  • lift建立了一個Observable後，加上以前的原始Observable，有兩個Observable。
  • 新的Observable裏的OnSubscribe加上原始的，共有兩個OnSubscribe。
  • 當用戶經過調用lift/map建立的Observable對象的subscribe方法時，因而它觸發了上面的call方法中的內容。
  • 在這個新的OnSubscribe的call方法中，傳入了目標的Subscriber，同時其外部類中還持有了原始的OnSubscribe。咱們先經過operator.call(oldSubscriber)方法，生成了新的Subscriber（new Subscriber），而後利用這個新的Subscriber向原始的Observable進行訂閱。

下面咱們之前面map實現的例子來分析一下源碼，上面的例子經過map操做符把Integer類型的Observable和String類型的Subscriber生成了訂閱關係。

  • map方法，它經過lift方法返回了一個String類型的Observable。
//其中T=Integer，R=String。
public final <R> Observable<R> map(Func1<? super T, ? extends R> func) {
        return lift(new OperatorMap<T, R>(func));
}
複製代碼
  • 下面看下OperatorMap這個對象，這個對象實現了operator<R,T>接口，而這個接口繼承於Func1<Subscriber<? super R>, Subscriber<? super T>>，在它實現的call方法中傳入了String類型的Subscriber（目標Subscriber），並返回了Integer類型的Subscriber（代理Subscriber），當它的方法被回調時，會調用目標Subscriber的對應方法，其中在調用onNext時，就用上了外部傳入的Func1函數：
@Override
    public Subscriber<? super T> call(final Subscriber<? super R> o) {
        return new Subscriber<T>(o) {

            @Override
            public void onCompleted() {
                o.onCompleted();
            }

            @Override
            public void onError(Throwable e) {
                o.onError(e);
            }

            @Override
            public void onNext(T t) {
                try {
                    o.onNext(transformer.call(t));
                } catch (Throwable e) {
                    Exceptions.throwIfFatal(e);
                    onError(OnErrorThrowable.addValueAsLastCause(e, t));
                }
            }

        };
    }
複製代碼
  • 接着再回過頭來看lift方法：
public final <R> Observable<R> lift(final Operator<? extends R, ? super T> operator) {
        return new Observable<R>(new OnSubscribe<R>() {
            @Override
            public void call(Subscriber<? super R> o) {
                try {
                    //返回一個Integer類型的Subscriber。
                    Subscriber<? super T> st = hook.onLift(operator).call(o);
                    try {
                        st.onStart();
                        //關鍵方法：Integer類型的OnSubscribe調用對應的Subscribe，這個call方法裏面寫了咱們的邏輯，當它調用onNext(Integer integer)時，實際上調用的是onNext(String str)。
                        onSubscribe.call(st);
                    } catch (Throwable e) {
                        if (e instanceof OnErrorNotImplementedException) {
                            throw (OnErrorNotImplementedException) e;
                        }
                        st.onError(e);
                    }
                } catch (Throwable e) {
                    if (e instanceof OnErrorNotImplementedException) {
                        throw (OnErrorNotImplementedException) e;
                    }
                    o.onError(e);
                }
            }
        });
    }
複製代碼
  • 最後就是調用subscribe方法。
相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程