作者： @怪盜kidou
如需轉(zhuǎn)載需在明顯位置保留作者信息及原文鏈接
如果博客中有不恰當(dāng)之處歡迎留言交流http://www.itdecent.cn/p/4e78d447394e

1、前言

一年多沒(méi)有寫(xiě)新博客了，今天又來(lái)水一篇，算是對(duì)《你真的會(huì)用RxJava么?RxJava線程變換之observeOn與subscribeOn》 這篇博客做一個(gè)補(bǔ)充，如果你自己有想過(guò)RxJava怎么實(shí)現(xiàn)線程變換以及讀過(guò)RxJava源碼的同學(xué)想必是知道今天主題的答案的。如果你不確定那么請(qǐng)繼續(xù)往下看。

既然之前的文章有講過(guò)為啥我時(shí)隔一年多還要再寫(xiě)一篇文章來(lái)說(shuō)明呢? 原因有兩個(gè)：

1. 我看到部分講 RxJava 線程切換的博客里針對(duì)這點(diǎn)說(shuō)得并不對(duì)
2. 之前的文章里講得并不怎么詳細(xì)

當(dāng)然按照慣列先給結(jié)論：并不是只有第一次才有效。

2、RxJava如何運(yùn)作

最開(kāi)始接觸到RxJava的時(shí)候感覺(jué)到最神奇的是線程變換，也就是 subscribeOn 以及 observeOn 操作符，還是在一條調(diào)用鏈上就搞定了，我當(dāng)時(shí)就在想我能不能自己也實(shí)現(xiàn)一個(gè)這樣的東西呢，所以在本地建一個(gè)叫 DeRxJava 的項(xiàng)目，在沒(méi)有看RxJava源碼的情況下自己來(lái)實(shí)現(xiàn)一下 subscribeOn 和 observeOn ，當(dāng)然實(shí)際上實(shí)現(xiàn)原理也非常簡(jiǎn)單，下面我們就分析一下。

為方便理解后面的內(nèi)容，我以下面代碼為例給大家進(jìn)行說(shuō)明，看看是RxJava是如何運(yùn)作的。

Observable
        .just(0)
        .observeOn(Schedulers.computation())
        .map(i -> i + 1)
        .subscribeOn(Schedulers.newThread())
        .subscribe(System.out::println);

請(qǐng)看下面的示意圖（向上的箭頭表示訂閱操作的方向，向下的箭頭表示數(shù)據(jù)流向，箭頭的顏色表示所在的線程，曲折的箭頭表示發(fā)生了線程切換）。

RxJava線程切換示意圖1

該圖我們應(yīng)該就能看出RxJava的運(yùn)作方式了，如果不能的我作一點(diǎn)提示，這里每調(diào)用一個(gè)操作符時(shí)都會(huì)創(chuàng)建一個(gè)新的Observable出來(lái)，每一個(gè)操作符產(chǎn)生的新Observable都會(huì)向上層的Observable注冊(cè)自己的回調(diào)。 subscribeOn 和 observeOn 的實(shí)現(xiàn)方法：subscribeOn 的原理就是在指定線程中向上游訂閱（白話就是在指定線程中去調(diào)上游的subscribe方法），observeOn 的原理是收到數(shù)據(jù)后在指定的線程中調(diào)用下游的回調(diào)方法（onNext/onError/onComplete等），而數(shù)據(jù)又總是在發(fā)生訂閱關(guān)系之后才被收到，所以 subscribeOn 即使出現(xiàn) observeOn 之后也能保證數(shù)據(jù)源運(yùn)行的線程。

3、為什么subscribeOn不是只有第一次才有效

那 多次調(diào)用subscribeOn只有第一次生效 這種說(shuō)法是怎么來(lái)的呢，我們?cè)僖韵旅娴拇a為例，看看RxJava的運(yùn)作方式。

Observable
        .just(0)
        .observeOn(Schedulers.computation())
        .map(i -> i + 1)
        .subscribeOn(Schedulers.newThread())
        .subscribeOn(Schedulers.io()) // 多了這一行
        .subscribe(System.out::println);

根據(jù)上面的代碼，從表面上看 just(0) 確定是發(fā)生在第一次調(diào)用 subscribeOn 時(shí)指定的 newThread 中，但第二個(gè)
subscribeOn 就真的沒(méi)有效果? 其實(shí)它確實(shí)生效了，只是又被上游的 subscribeOn 切換到了別的線程而已，具體見(jiàn)下圖

RxJava線程切換示意圖2

4、subscribeOn和observeOn的實(shí)現(xiàn)源碼

為讓大家更好的理解上面的圖，我從RxJava 2.1.5 摘取了這兩個(gè)操作符的源碼進(jìn)行了精簡(jiǎn)并添加了說(shuō)明。

subscribeOn 的實(shí)現(xiàn)源碼：

@Override
// 這個(gè)方法會(huì)在下游調(diào)用subscribe方法時(shí)被調(diào)用
public void subscribeActual(final Observer<? super T> s) {
    final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(s);
    s.onSubscribe(parent);
    // 這里并沒(méi)有直接訂閱，而是先進(jìn)行了線程變換（scheduler.scheduleDirect）
    // 大白話：在線程變換后才掉用subscribe方法
    parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
}

static final class SubscribeOnObserver<T> extends AtomicReference<Disposable> implements Observer<T>, Disposable {
    // 其它代碼已經(jīng)被精簡(jiǎn)，只以 onNext為例
    @Override
    public void onNext(T t) {
        // 收到數(shù)據(jù)后不進(jìn)行線程變換
        actual.onNext(t);
    }
}

final class SubscribeTask implements Runnable {
    // 其它代碼已經(jīng)被精簡(jiǎn)
    @Override
    public void run() {
        // 這個(gè)方法會(huì)被指定的scheduler調(diào)用，向上游訂閱時(shí)線程已經(jīng)發(fā)生了變化
        // 所以保證了上游所運(yùn)行的線程
        source.subscribe(parent);
    }
}

observeOn 的實(shí)現(xiàn)源碼：

@Override
protected void subscribeActual(Observer<? super T> observer) {
    if (scheduler instanceof TrampolineScheduler) {
        source.subscribe(observer);
    } else { // 直接向上游訂閱數(shù)據(jù)，不進(jìn)行線程切換,切換操作是在回調(diào)里進(jìn)行的，見(jiàn)下面的onNext
        Scheduler.Worker w = scheduler.createWorker();
        source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
    }
}

static final class ObserveOnObserver<T> extends BasicIntQueueDisposable<T> implements Observer<T>, Runnable {
@Override
public void onNext(T t) {
    if (done) {
        return;
    }
    // 這里選把數(shù)據(jù)放到隊(duì)列中
    if (sourceMode != QueueDisposable.ASYNC) {
        queue.offer(t);
    }
    // 在schedule方法里進(jìn)行線程切換并把數(shù)據(jù)循環(huán)取出
    // 回調(diào)給下游，這樣下游就會(huì)在指定的線程中收到數(shù)據(jù)
    // 大白話：在回調(diào)里切換線程
    schedule();
}

5、結(jié)語(yǔ)

最后就是告訴大家不要被表象所迷惑，就拿這個(gè) subscribeOn 來(lái)說(shuō)，只要你知道實(shí)現(xiàn)的方式就是在指定線程中去向上級(jí)訂閱你就應(yīng)該想到正常情況下 subscribeOn 并不知道上游有沒(méi)有其它的 subscribeOn 操作符，所以每次它都只能老老實(shí)實(shí)的去指定的線程中去向上游訂閱。

Ps:寫(xiě)個(gè)博客還真不容易，就這么一篇小小的文章搞了2、3天，想了很多例子都被去掉了，最后還是直接畫(huà)了兩個(gè)圖來(lái)展示，如果你覺(jué)得有用的話請(qǐng)點(diǎn)贊、分享、打賞，如果還有疑問(wèn)話請(qǐng)直接評(píng)論，我雖然并不常更新博客但還是有一直留意博客上的評(píng)論和私信的，這次就到這兒。

我最近剛剛開(kāi)通了微信公眾號(hào)（怪盜kidou），歡迎關(guān)注