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

一、基礎(chǔ)概述

RxJava的關(guān)鍵是異步，即使隨著程序的邏輯變得復(fù)雜，它依然能夠保持簡潔。

二、API介紹和原理剖析

觀察者模式面向的需求是：A對(duì)象（觀察者）對(duì)B對(duì)象（被觀察者）的某種變化高度敏感，需要在B變化的一瞬間做出反應(yīng)，觀察者采用注冊(cè)Register或者訂閱Subscribe的方式，告訴觀察者，我需要你的某某狀態(tài)，并在它變化的時(shí)候通知我，在RxJava當(dāng)中，Observable是被觀察者，Observer就是觀察者。

RxJava有四個(gè)基本概念：

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

Observable和Observer通過subscribe方法實(shí)現(xiàn)訂閱關(guān)系，Observable可以在需要的時(shí)候發(fā)出事件來通知Observer。

RxJava有以下三種事件：

  • onNext：普通事件。
  • onCompleted：RxJava不僅把每個(gè)事件單獨(dú)處理，還會(huì)把它們看作一個(gè)隊(duì)列，當(dāng)不會(huì)再有新的onNext事件發(fā)出時(shí)，需要觸發(fā)onCompleted事件作為標(biāo)志。
  • onError：onCompleted和有且僅有一個(gè)，并且是事件序列中的最后一個(gè)。

三、基本實(shí)現(xiàn)

RxJava的基本實(shí)現(xiàn)有以下三點(diǎn)：
1）創(chuàng)建觀察者 - Observer

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還內(nèi)置了一個(gè)實(shí)現(xiàn)了Observer的抽象類：Subscriber，它對(duì)Observer接口進(jìn)行了一些擴(kuò)展，實(shí)質(zhì)上在RxJava的subscribe過程中，Observer也總是被轉(zhuǎn)換成為一個(gè)Subscriber再使用，他們的區(qū)別在與：

  • onStart：這是新增的方法，它會(huì)在subscribe剛開始，而事件還未發(fā)送之前被調(diào)用，它總是在subscribe所發(fā)生的線程被調(diào)用。
  • unsubscribe：這是它實(shí)現(xiàn)的另一個(gè)接口Subscription的方法，用于取消訂閱，在這個(gè)方法被調(diào)用后，Subscriber將不再接收事件，一般在調(diào)用這個(gè)方法前，可以使用isUnsubscribed判斷一下狀態(tài)，Observable在訂閱之后會(huì)持有Subscriber的引用，因此不釋放會(huì)有內(nèi)存泄漏的危險(xiǎn)。

2）創(chuàng)建被觀察者 - Observable
RxJava用create方法來創(chuàng)建一個(gè)observable，

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();
    }
});

這里傳入了一個(gè)Observable.OnSubscribe<T>對(duì)象作為參數(shù)，它會(huì)被存儲(chǔ)在返回的Observable對(duì)象當(dāng)中，它的作用相當(dāng)于一個(gè)計(jì)劃表，當(dāng)Observable被訂閱的時(shí)候，OnSubscribe的call方法會(huì)自動(dòng)被調(diào)用，事件序列被依次觸發(fā)。
create是RxJava最基本的創(chuàng)造事件序列的方法，基于這個(gè)方法，還提供了一些快捷方法來創(chuàng)建事件隊(duì)列：

  • 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）訂閱 - subscribe

observable.subscribe(observer);
observable.subscribe(subscriber);

其內(nèi)部核心的代碼類似于：

public Subscription subscribe(Subscriber subscriber) {
    //準(zhǔn)備方法。
    subscriber.onStart();
    //事件發(fā)送的邏輯開始執(zhí)行，這個(gè)onSubscribe就是創(chuàng)建Observable時(shí)新建的OnSubscribe對(duì)象。
    onSubscribe.call(subscriber);
    //把傳入的Subscriber轉(zhuǎn)換為Subscription并返回，方便unsubscribe。
    return subscriber;
}

Observable.subscribe方法除了支持傳入Observer和Subscriber，還支持傳入Action0、Action1這樣不完整定義的回調(diào)，RxJava會(huì)自動(dòng)根據(jù)定義創(chuàng)建出Subscriber。

四、線程控制

在不指定線程的情況下，RxJava遵循這樣的原則，在哪個(gè)線程調(diào)用subscribe，就在哪個(gè)線程產(chǎn)生事件，在哪個(gè)線程產(chǎn)生事件，就在哪個(gè)線程消費(fèi)事件，如果需要消費(fèi)線程，那么就需要用到Scheduler， RxJava內(nèi)置了幾個(gè)Scheduler：

  • Schedulers.immediate：直接在當(dāng)前線程運(yùn)行。
  • Schedulers.newThread：總是啟用新線程，并在線程執(zhí)行操作。
  • Schedulers.io：其內(nèi)部實(shí)現(xiàn)是一個(gè)無數(shù)量上限的的線程池，可以重用空閑的線程，不要把計(jì)算工作放在io，可以避免創(chuàng)建不必要的線程。
  • Schedulers.computation：使用固定的線程池，大小為CPU核數(shù)。
  • AndroidSchedulers.mainThread：指定的操作將在Android主線程中運(yùn)行。

對(duì)線程控制有以下兩個(gè)方法：

  • subscribeOn：指定subscribe發(fā)生的線程，即Observable.OnSubscribe被激活時(shí)所處的線程，也就是call方法執(zhí)行時(shí)所處的線程。
  • observeOn：指定Subscriber所運(yùn)行在的線程。

observeOn指定的是Subscriber的線程，而這個(gè)Subscriber并不一定是subscribe()參數(shù)中的Subscriber，而是observeOn執(zhí)行時(shí)的當(dāng)前Observable所對(duì)應(yīng)的Subscriber，即它的直接下級(jí)Subscriber，也就是它之后的操作所在的線程，因此，如果有多次切換線程的要求，只要在每個(gè)想要切換線程的位置調(diào)用依次observeOn即可。
和observeOn不同，subscribeOn只能調(diào)用一次，下面我們來分析一下它的內(nèi)部實(shí)現(xiàn)，首先是subscribeOn的原理：
subscribeOn和ObserveOn都做了線程切換的工作：

  • subscribeOn的線程切換發(fā)生在OnSubscribe中，即在它通知上一級(jí)的OnSubscribe時(shí)，這時(shí)事件還沒有發(fā)送，因此subscribeOn的線程控制可以從事件發(fā)出的開端造成影響。
Paste_Image.png
  • observeOn的線程切換則發(fā)生在它內(nèi)建的Subscriber中，即發(fā)生在它即將給下一級(jí)Subscriber發(fā)送事件時(shí)，因此控制的是它后面的線程。
Paste_Image.png

五、變換

變換，就是將事件序列中的對(duì)象或整個(gè)序列進(jìn)行加工處理，轉(zhuǎn)換不同的事件或者序列。

5.1 map()

通過FuncX，把參數(shù)中的Integer轉(zhuǎn)換成為String，是最常用的變換，這個(gè)變換是發(fā)生在subscribeOn所指定的線程當(dāng)中的。

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有一個(gè)共同點(diǎn)，就是把傳入的參數(shù)轉(zhuǎn)化之后返回另一個(gè)對(duì)象，但是和map不同的是，flatMap返回的是一個(gè)Observable對(duì)象，而且它并不直接把這個(gè)對(duì)象傳給Subscriber，而是通過這個(gè)新建的Observable來發(fā)送事件，其整個(gè)的調(diào)用過程：

  • 使用傳入的事件對(duì)象創(chuàng)建一個(gè)Observable。
  • 激活這個(gè)Observable，通過它來發(fā)送事件。
  • 每一個(gè)創(chuàng)建出來的Observable發(fā)送的事件，被匯入同一個(gè)Observable，它復(fù)雜將這些事件同一交給Subscriber的回調(diào)方法。
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

六、變換的原理

變換的實(shí)質(zhì)是針對(duì)事件序列的處理和再發(fā)送，在RxJava的內(nèi)部，它們是基于同一個(gè)基礎(chǔ)的變換方法lift(operator)

//生成了一個(gè)新的Observable并返回。
public <R> Observable<R> lift(Operator<? extends R, ? super T> operator) {
    //構(gòu)造新的Observable時(shí)，同時(shí)新建了一個(gè)OnSubscribe對(duì)象。
    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創(chuàng)建了一個(gè)Observable后，加上之前的原始Observable，有兩個(gè)Observable。
  • 新的Observable里的OnSubscribe加上原始的，共有兩個(gè)OnSubscribe。
  • 當(dāng)用戶通過調(diào)用lift/map創(chuàng)建的Observable對(duì)象的subscribe方法時(shí)，于是它觸發(fā)了上面的call方法中的內(nèi)容。
  • 在這個(gè)新的OnSubscribe的call方法中，傳入了目標(biāo)的Subscriber，同時(shí)其外部類中還持有了原始的OnSubscribe。我們先通過operator.call(oldSubscriber)方法，生成了新的Subscriber（new Subscriber），然后利用這個(gè)新的Subscriber向原始的Observable進(jìn)行訂閱。

下面我們以前面map實(shí)現(xiàn)的例子來分析一下源碼，上面的例子通過map操作符把Integer類型的Observable和String類型的Subscriber生成了訂閱關(guān)系。

  • map方法，它通過lift方法返回了一個(gè)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這個(gè)對(duì)象，這個(gè)對(duì)象實(shí)現(xiàn)了operator<R,T>接口，而這個(gè)接口繼承于Func1<Subscriber<? super R>, Subscriber<? super T>>，在它實(shí)現(xiàn)的call方法中傳入了String類型的Subscriber（目標(biāo)Subscriber），并返回了Integer類型的Subscriber（代理Subscriber），當(dāng)它的方法被回調(diào)時(shí)，會(huì)調(diào)用目標(biāo)Subscriber的對(duì)應(yīng)方法，其中在調(diào)用onNext時(shí)，就用上了外部傳入的Func1函數(shù)：
    @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 {
                    //返回一個(gè)Integer類型的Subscriber。
                    Subscriber<? super T> st = hook.onLift(operator).call(o);
                    try {
                        st.onStart();
                        //關(guān)鍵方法：Integer類型的OnSubscribe調(diào)用對(duì)應(yīng)的Subscribe，這個(gè)call方法里面寫了我們的邏輯，當(dāng)它調(diào)用onNext(Integer integer)時(shí)，實(shí)際上調(diào)用的是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);
                }
            }
        });
    }
  • 最后就是調(diào)用subscribe方法。
最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時(shí)請(qǐng)結(jié)合常識(shí)與多方信息審慎甄別。
平臺(tái)聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點(diǎn)，簡書系信息發(fā)布平臺(tái)，僅提供信息存儲(chǔ)服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

  • 我從去年開始使用 RxJava ，到現(xiàn)在一年多了。今年加入了 Flipboard 后，看到 Flipboard 的...
    Jason_andy閱讀 5,744評(píng)論 7贊 62
  • 前言我從去年開始使用 RxJava ，到現(xiàn)在一年多了。今年加入了 Flipboard 后，看到 Flipboard...
    占導(dǎo)zqq閱讀 9,299評(píng)論 6贊 151
  • 文章轉(zhuǎn)自：http://gank.io/post/560e15be2dca930e00da1083作者：扔物線在正...
    xpengb閱讀 7,127評(píng)論 9贊 73
  • 前言 今天，沒有風(fēng)和日麗，沒有太陽高照夏夏苦逼的坐著公交車去面試了一把。無論結(jié)果如何，總要漲一波經(jīng)驗(yàn)的說于是乎夏夏...
    __夏至未至閱讀 551評(píng)論 6贊 6
  • 作者/胄寧 空山暮雨渲染著遲夏的凄悲， 劫后重生仍舊殘留著的余味， 你楚楚的眼神在向他傳達(dá)著曖昧， 在這曖昧的背后...
    胄寧閱讀 582評(píng)論 8贊 6

友情鏈接更多精彩內(nèi)容