你也可以檢視我的其他同類文章，也會讓你有一定的收貨！

為什麼多次呼叫subscribeOn()卻只有第一個起作用？
為什麼多次呼叫observeOn()卻可以切換到不同執行緒
observeOn()後能不能再次呼叫subscribeOn()？

如果你有這些疑問，那接下來的內容必定能解決你心頭的疑惑

subscribeOn()和observeOn()的區別

subscribeOn()和observeOn()都是用來切換執行緒用的

subscribeOn()改變呼叫它之前程式碼的執行緒
observeOn()改變呼叫它之後程式碼的執行緒

一、subscribeOn()

在講解他的原理之前，先來一個簡單的例子，有個感性認識，學起來更輕鬆

先說結論：subscribeOn 作用於該操作符之前的 Observable 的建立操符作以及 doOnSubscribe 操作符，換句話說就是 doOnSubscribe 以及 Observable 的建立操作符總是被其之後最近的 subscribeOn 控制。沒看懂不要緊，看下面程式碼和圖你就懂了。

這裡寫圖片描述

Observable
        .create(new Observable.OnSubscribe<String>() {
            @Override
            public void 
 call(Subscriber<? super String> subscriber) {
                threadInfo("OnSubscribe.call()");
                subscriber.onNext("RxJava");
            }
        })
        .subscribeOn(getNamedScheduler("create之後的subscribeOn"))
        .doOnSubscribe(() -> threadInfo(".doOnSubscribe()-1" 
))
        .subscribeOn(getNamedScheduler("doOnSubscribe1之後的subscribeOn"))
        .doOnSubscribe(() -> threadInfo(".doOnSubscribe()-2"))
        .subscribe(s -> {
            threadInfo(".onNext()");
            System.out.println(s + "-onNext");
        });

結果如下：

.doOnSubscribe()-2 => main
.doOnSubscribe()-1 => doOnSubscribe1之後的subscribeOn
OnSubscribe.call() => create之後的subscribeOn
.onNext() => create之後的subscribeOn
RxJava-onNext

3號框中的.doOnSubscribe(() -> threadInfo(“.doOnSubscribe()-2”)) 的之後由於沒有subscribeOn操作符所以回撥到該段程式碼被呼叫的執行緒（即主執行緒）

由於 subscribe 之前沒有使用observeOn 指定Scheduler，所以.onNext()的執行緒是和OnSubscribe.call()使用相同的Scheduler 。

下面通過原始碼來分析一下：

1、示例程式碼：

 Observable
                .create(new Observable.OnSubscribe<String>() {
                    @Override
                    public void call(Subscriber<? super String> subscriber) {
                        subscriber.onNext("a");
                        subscriber.onNext("b");

                        subscriber.onCompleted();
                    }
                })
                .subscribeOn(Schedulers.io())

                .subscribe(new Observer<String>() {
                    @Override
                    public void onCompleted() {

                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onNext(String integer) {
                        System.out.println(integer);
                    }
                });

執行如下：

a
b

2、subscribeOn()原始碼

public final Observable<T> subscribeOn(Scheduler scheduler) {
        if (this instanceof ScalarSynchronousObservable) {
            return ((ScalarSynchronousObservable<T>)this).scalarScheduleOn(scheduler);
        }
        return create(new OperatorSubscribeOn<T>(this, scheduler));
    }

很明顯，會走if之外的方法。

在這裡我們可以看到，又建立了一個OperatorSubscribeOn物件，但建立時傳入的引數為OperatorSubscribeOn(this,scheduler)，我們看一下此物件以及其對應的構造方法

3、create()的原始碼：

public static <T> Observable<T> create(OnSubscribe<T> f) {
        return new Observable<T>(hook.onCreate(f));
    }

我們看到這個方法，使用OperatorSubscribeOn這個類，來建立一個新的Observable，那就把它叫做Observable_2，把原來的Observable叫做Observable_1

4、OperatorSubscribeOn類的原始碼：

public final class OperatorSubscribeOn<T> implements OnSubscribe<T> {

    final Scheduler scheduler;
    final Observable<T> source;

    public OperatorSubscribeOn(Observable<T> source, Scheduler scheduler) {
        this.scheduler = scheduler;
        this.source = source;
    }

    @Override
    public void call(final Subscriber<? super T> subscriber) {
        final Worker inner = scheduler.createWorker();
        subscriber.add(inner);

        inner.schedule(new Action0() {
            @Override
            public void call() {
                final Thread t = Thread.currentThread();

                Subscriber<T> s = new Subscriber<T>(subscriber) {
                    @Override
                    public void onNext(T t) {
                        subscriber.onNext(t);
                    }

                    @Override
                    public void onError(Throwable e) {
                        try {
                            subscriber.onError(e);
                        } finally {
                            inner.unsubscribe();
                        }
                    }

                    @Override
                    public void onCompleted() {
                        try {
                            subscriber.onCompleted();
                        } finally {
                            inner.unsubscribe();
                        }
                    }

                    @Override
                    public void setProducer(final Producer p) {
                        subscriber.setProducer(new Producer() {
                            @Override
                            public void request(final long n) {
                                if (t == Thread.currentThread()) {
                                    p.request(n);
                                } else {
                                    inner.schedule(new Action0() {
                                        @Override
                                        public void call() {
                                            p.request(n);
                                        }
                                    });
                                }
                            }
                        });
                    }
                };

                source.unsafeSubscribe(s);
            }
        });
    }
}

OperatorSubscribeOn類implements 了Onsubscribe介面，並實現call()方法
OperatorSubscribeOn的構造方法，
- 儲存了Observable物件，就是呼叫了subscribeOn()方法的Observable物件
- 並儲存了Scheduler物件。

這裡做個總結。

把Observable.create()建立的稱之為Observable_1,OnSubscribe_1。
把subscribeOn()建立的稱之為Observable_2，OnSubscribe_2

Observable_1, OnSubscribe_1是由示例程式碼的第1、2行建立的
OperatorSubscribeOn類implements Onsubscribe類，所以可以當做Onsubscribe類使用。（OnSubscribe_2）
並且OnSubscribe_2中儲存了Observable_1的應用，即source。（在OperatorSubscribeOn原始碼的第8行）
在subscribeOn()原始碼的倒數第二行，create(new OperatorSubscribeOn<T>(this, scheduler))返回新建立的Observable_2物件。

4.1、分析call()方法。

inner.schedule()改變了執行緒，此時Action的call()在我們指定的執行緒中執行。
示例程式碼的Subscriber被包裝了一層，賦給物件S（Subscriber_2）。見上面程式碼21行。
source.unsafeSubscribe(s);，
- 注意：source是Observable_1物件，這裡的s就是Subscriber_2
- 因為呼叫過subscribeOn(Schedulers.io())後，返回Observable_2物件，所以示例程式碼第13行程式碼的subscribe()就是Observable_2.subscribe()，也就是執行OnSubscribe_2的call()方法(即OperatorSubscribeOn類的原始碼的第12行)。

4.2 看一下source.unsafeSubscribe(s);（第65行）程式碼都做了什麼

這裡的source就是Observable_1，s是Subscriber_2

unsafeSubscribe()原始碼：

public final Subscription unsafeSubscribe(Subscriber<? super T> subscriber) {
        try {
            // new Subscriber so onStart it
            subscriber.onStart();
            // allow the hook to intercept and/or decorate
            hook.onSubscribeStart(this, onSubscribe).call(subscriber);
            return hook.onSubscribeReturn(subscriber);
        } catch (Throwable e) {
            // special handling for certain Throwable/Error/Exception types
            Exceptions.throwIfFatal(e);
            // if an unhandled error occurs executing the onSubscribe we will propagate it
            try {
                subscriber.onError(hook.onSubscribeError(e));
            } catch (Throwable e2) {
                Exceptions.throwIfFatal(e2);
                // if this happens it means the onError itself failed (perhaps an invalid function implementation)
                // so we are unable to propagate the error correctly and will just throw
                RuntimeException r = new RuntimeException("Error occurred attempting to subscribe [" + e.getMessage() + "] and then again while trying to pass to onError.", e2);
                // TODO could the hook be the cause of the error in the on error handling.
                hook.onSubscribeError(r);
                // TODO why aren't we throwing the hook's return value.
                throw r;
            }
            return Subscriptions.unsubscribed();
        }
    }

關鍵程式碼：

hook.onSubscribeStart(this, onSubscribe).call(subscriber);

該方法即呼叫了OnSubscribe_1.call()方法。注意，此時的call()方法在我們指定的執行緒中執行。那麼就起到了改變執行緒的作用。

對於以上執行緒，我們可以總結，其有如下流程：

Observable.create() : 建立了Observable_1和OnSubscribe_1;
subscribeOn()：建立Observable_2和OperatorSubscribeOn(OnSubscribe_2)，同時OperatorSubscribeOn儲存了Observable_1的引用。
示例程式碼中的subscribe(Observer) 實際上就是呼叫Observable_2.subscribe(Observer):
- 呼叫OperatorSubscribeOn的call()。call()改變了執行緒的執行，並且呼叫了Observable_1.unsafeSubscribe(s);
- Observable_1.unsafeSubscribe(s);，該方法的實現中呼叫了OnSubscribe_1的call()。

這樣就實現了在指定執行緒執行OnSubscribe的call()函式，無論我們的subscribeOn()放在哪裡,他改變的是subscribe()的過程，而不是onNext()的過程。

那麼如果有多個subscribeOn()，那麼執行緒會怎樣執行呢。如果按照我們的邏輯，有以下程式

Observable.just("ss") 
                .subscribeOn(Schedulers.io())   // ----1---
                .subscribeOn(Schedulers.newThread()) //----2----
                .subscribe(new Action1<String>() {
                    @Override
                    public void call(String s) {

                    }
                });

那麼，我們根據之前的原始碼分析其執行邏輯。

Observable.just(“ss”),建立Observable，OnSubscribe
相關推薦

RxJava之三—— observeOn()與subscribeOn()的詳解

你也可以檢視我的其他同類文章，也會讓你有一定的收貨！為什麼多次呼叫subscribeOn()卻只有第一個起作用？為什麼多次呼叫observeOn()卻可以切換到不同執行緒 observeOn()後能不能再次呼叫subscribeOn()？

RxJava之五—— observeOn()與subscribeOn()的詳解

你也可以檢視我的其他同類文章，也會讓你有一定的收貨！為什麼多次呼叫subscribeOn()卻只有第一個起作用？為什麼多次呼叫observeOn()卻可以切換到不同執行緒 observeOn()後能不能再次呼叫subscribeOn()？如

RxJava之五—— observeOn 與subscribeOn 的詳解

分享一下我老師大神的人工智慧教程！零基礎，通俗易懂！http://blog.csdn.net/jiangjunshow 也歡迎大家轉載本篇文章。分享知識，造福人民，實現我們中華民族偉大復興！

OSPF詳解之三：OSPF LSA詳解

ospf lsa詳解 forwarding address OSPF LSA詳解OSPF V2版本中常用的主要有6類LSA，分別是Router-LSA、Network-LSA、Network-summary-LSA、ASBR-summary-LSA、AS-External-LSA、NSSA-LSA，接

STL之三：deque用法詳解

deque函式： deque容器為一個給定型別的元素進行線性處理，像向量一樣，它能夠快速地隨機訪問任一個元素，並且能夠高效地插入和刪除容器的尾部元素。但它又與vector不同，deque支援高效插入和刪除容器的頭部元素，因此也叫做雙端佇列。deque類常用的函式如下。

java資料結構之：ArrayList與LinkedList詳解

前篇博文講到了Java底層怎麼知道ArrayList和LinkedList哪個是隨機訪問哪個是順序訪問的，本片博文主要介紹各自的資料特點。 ArrayList: 從他的這個建構函式我們可以知道他的底層實現就是一個Object物件陣列。第二個建構函式構造一個空的

MySQL高階之索引失效與優化詳解

案例所用的表結構、索引、與資料如下：索引失效與優化 1、全值匹配我最愛 2、最佳左字首法則（帶頭索引不能死，中間索引不能斷）如果索引了多個列，要遵守最佳左字首法則。指的是查詢從索引的最左前列開始並且不跳過索引中的列。正確的示例參考上圖。錯誤的示例：

高通sensor架構例項分析之三(adsp上報資料詳解、校準流程詳解)

本系列導航：從adsp獲取資料的方法分為同步、非同步兩種方式，但一般在實際使用中使用非同步方式，因為同步獲取資料會因外設匯流排速率低的問題阻塞smgr，降低效率，增加功耗。 Sensor上報資料的方式分為如下幾種 sync 同步資料上報，(每次上報一個數據) async

O(n*logn)級別的演算法之二（快速排序）的三種實現方法詳解及其與歸併排序的對比

快速排序被稱為二十世紀演算法界的一大傑作一，單路快排 1.測試用例： #ifndef INC_06_QUICK_SORT_DEAL_WITH_NEARLY_ORDERED_ARRAY_SORTTESTHELPER_H #define INC_06_QUICK_

[js高手之路]原型對象(prototype)與原型鏈相關屬性與方法詳解

隱式之前 username tar uname create pro getproto .get 一，instanceof： instanceof檢測左側的__proto__原型鏈上，是否存在右側的prototype原型. 我在之前的兩篇文章 [js高手之路]構造函數的基

[數據庫事務與鎖]詳解三: 深入分析事務的隔離級別

不可 ans 提高重要不能標準 insert lec 使用註明: 本文轉載自http://www.hollischuang.com/archives/943 本文詳細介紹四種事務隔離級別，並通過舉例的方式說明不同的級別能解決什麽樣的讀現象。並且介紹了在關系型數據

SpringMVC 框架系列之組件概述與配置詳解

align 概述 handle ont htm 配置文件掃描器 springmvc 解析在上一篇文章 SpringMVC 框架系列之初識與入門實例的實例中，我們已經知道，SpringMVC 框架是一個 web 層的框架，本篇文章就詳細解釋一下 SpringMVC 框架

二層交換機與三層交換機區別詳解

交換機VLAN我們習慣說，在二層網絡環境中相同vlan之間可以通信，不同vlan之間不可以通信，如果想通信必須借助三層設備，所以說三層交換機必須要做的事情是路由轉發，但是二、三層交換機具體有什麽區別呢？二層交換機工作於OSI模型的第2層(數據鏈路層)，故而稱為二層交換機。二層交換技術發展比較成熟，二層交換

一看就懂！【英雄聯盟銳雯】與 Python 詳解設計模式之門面模式

【網路配圖】設計模式（Design Pattern）是一套被反覆使用、多數人知曉的、經過分類的、程式碼設計經驗的總結。使用設計模式的目的：為了程式碼可重用性、讓程式碼更容易被他人理解、保證程式碼可靠性。設計模式使程式碼編寫真正工程化；設計模式是軟體工程的基石脈絡，如同大廈的結構一樣

android 繪圖之Path與Paint詳解[轉]

/** * Paint類介紹 * * Paint即畫筆，在繪圖過程中起到了極其重要的作用，畫筆主要儲存了顏色， * 樣式等繪製資訊，指定了如何繪製文字和圖形，畫筆物件有很多設定方法， * 大體上可以分為兩類，一類與圖形繪製相關，一類與文字繪製相關。 * * * 1.圖形繪製 * setARG

Ocelot簡易教程（三）之主要特性及路由詳解

上篇《Ocelot簡易教程（二）之快速開始2》教大家如何快速跑起來一個ocelot例項專案，也只是簡單的對Ocelot進行了配置，這篇文章會給大家詳細的介紹一下Ocelot的配置資訊。希望能對大家深入使用Ocelot有所幫助。上篇中也提到了，最簡單的Ocelot如下面所示，只有簡單的兩個節點，一個是ReR

Spring Cloud之Zuul(三)：路由配置詳解

主題路由配置詳解前言現實中可能只想讓Zuul代理部分微服務，又或者需要對URL進行更加精確的控制。Zuul的路由配置非常靈活、簡單，本部落格通過幾個例項，詳細講解Zuul的路由配置。內容 1.自定義指定微服務的訪問路徑說明：配置zuul.routes.指定微服務

Gradle使用詳解(三) 之 Android Gradle外掛配置詳解

通上前兩篇博文由淺入深的學習Gradle的基礎和Gradle的Java外掛相關知識後，現在終於到了高潮部分了，讓我們來進一步正式學習Android Gradle外掛。前面提到，Android Gradle外掛是一個基於內建的Java外掛來實現的第三方外掛，它是由google的

Java 高併發第三階段實戰---Java併發包深入解析與使用詳解

汪文君高併發程式設計第三階段01講 AtomicInteger多執行緒下測試講解. 汪文君高併發程式設計第三階段02講 AtomicInteger API詳解，以及CAS演算法詳細介紹. 汪文君高併發程式設計第三階段03講利用CAS構造一個TryLock自定義顯式鎖. 汪文君高併發程式設計第三階段04講利

netty原始碼分析之-SimpleChannelInboundHandler與ChannelInboundHandlerAdapter詳解(6)

每一個Handler都一定會處理出站或者入站（也可能兩者都處理）資料，例如對於入站的Handler可能會繼承SimpleChannelInboundHandler或者ChannelInboundHandlerAdapter，而SimpleChannelIn

RxJava之三—— observeOn()與subscribeOn()的詳解