2. 程式人生 > >RxJava2.0(四)執行緒之間切換的內部原理

RxJava2.0(四)執行緒之間切換的內部原理

阿新 • • 發佈：2018-12-04

基本程式碼

來看一下基本程式碼：

 Observable.create((ObservableOnSubscribe<Integer>) e -> {
            e.onNext(1);
            e.onNext(2);
            e.onComplete();
        }).subscribeOn(Schedulers.io())
          .observeOn(AndroidSchedulers.mainThread())
          .subscribe(i -> System.out.println("onNext : i= "
 
 + i));

複製程式碼

很簡單，即訂閱時將task交給子執行緒去做，而資料的回撥則在Android主執行緒中執行。

一、subscribeOn()

點選檢視原始碼:

public final Observable<T> subscribeOn(Scheduler scheduler) {
        //非空判斷和hook
        ObjectHelper.requireNonNull(scheduler, "scheduler is null");
        return RxJavaPlugins.onAssembly(new ObservableSubscribeOn<T>(this
 
, scheduler));
    }

複製程式碼

實際上這個方法返回了一個ObservableSubscribeOn物件。我們有理由猜測這個ObservableSubscribeOn應該和上文的ObservableMap及ObservableDoOnEach相似，都是Observable的一個包裝類（裝飾器）：

//1.ObservableSubscribeOn也是Observable的一個裝飾器
public final class ObservableSubscribeOn<T> extends AbstractObservableWithUpstream<T, T> 
 
{
    final Scheduler scheduler;

    public ObservableSubscribeOn(ObservableSource<T> source, Scheduler scheduler) {
       //2.儲存上游的ObservableSource和排程器
        super(source);
        this.scheduler = scheduler;
    }

    @Override
    public void subscribeActual(final Observer<? super T> s) {
        //3.new 一個SubscribeOnObserver
        final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(s);

        //4.回撥方法，這說明下游的onSubscribe回撥方法所線上程和執行緒排程無關
        //  是訂閱時所在的執行緒
        s.onSubscribe(parent);

        //5.立即執行執行緒排程
        parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
    }
}

複製程式碼

前兩步我們不需要 再多解釋，直接看第三點，我們看看SubscribeOnObserver這個類：

SubscribeOnObserver

static final class SubscribeOnObserver<T> extends AtomicReference<Disposable> implements Observer<T>, Disposable {

        private static final long serialVersionUID = 8094547886072529208L;
        //下游的Observer
        final Observer<? super T> actual;
        //儲存上游的Disposable，自身dispose時，連同上游一起dispose
        final AtomicReference<Disposable> s;

        SubscribeOnObserver(Observer<? super T> actual) {
            this.actual = actual;
            this.s = new AtomicReference<Disposable>();
        }

        @Override
        public void onSubscribe(Disposable s) {
            DisposableHelper.setOnce(this.s, s);
        }

        @Override
        public void onNext(T t) {
            actual.onNext(t);
        }

        @Override
        public void onError(Throwable t) {
            actual.onError(t);
        }

        @Override
        public void onComplete() {
            actual.onComplete();
        }

        @Override
        public void dispose() {
            DisposableHelper.dispose(s);
            DisposableHelper.dispose(this);
        }


複製程式碼

類似Observable和ObservableMap，SubscribeOnObserver同樣是Disposable和Observer的一個裝飾器，提供了對下游資料的傳遞，以及將task dispose的介面。

第4步我們之前就講過了，直接看第5步：

    //5.立即執行執行緒排程
        parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
複製程式碼

我們看看SubscribeTask這個類：

SubscribeTask

final class SubscribeTask implements Runnable {
        private final SubscribeOnObserver<T> parent;

        SubscribeTask(SubscribeOnObserver<T> parent) {
            this.parent = parent;
        }

        @Override
        public void run() {
            source.subscribe(parent);
        }
    }

複製程式碼

難以置信的簡單，SubscribeTask 僅僅是一個Runnable 介面的實現類而已，通過將SubscribeOnObserver作為引數存起來，在run()方法中添加了上游Observable的被訂閱事件，就沒有了別的操作，

接下來我們看一下scheduler.scheduleDirect(SubscribeTask)中的程式碼：

public abstract class Scheduler {
    //...
    public Disposable scheduleDirect(@NonNull Runnable run) {
        return scheduleDirect(run, 0L, TimeUnit.NANOSECONDS);
    }

    public Disposable scheduleDirect(@NonNull Runnable run, long delay, @NonNull TimeUnit unit) {
        // Worker 本身就是Disposable 的實現類
        // 請注意， createWorker()所建立的worker，
        // 實際就是Schdulers.io()所提供的IoScheduler所建立的worker
        final Worker w = createWorker();

        //hook相關
        final Runnable decoratedRun = RxJavaPlugins.onSchedule(run);

        DisposeTask task = new DisposeTask(decoratedRun, w);

        //即 worker.schedule(task, 0, TimeUnit.NANOSECONDS): 立即執行task
        w.schedule(task, delay, unit);

        return task;
    }
    //...
}

複製程式碼

我們不要追究過深，我們看一下這個createWorker方法的註釋說明：

/**
     * Retrieves or creates a new {@link Scheduler.Worker} that represents serial execution of actions.
     * 檢索或建立一個新的{@link Scheduler.Worker}表示一系列的action
     * 
     * When work is completed it should be unsubscribed using {@link Scheduler.Worker#dispose()}.
     * 當work完成後，應使用{@link Scheduler.Worker＃dispose（）}取消訂閱。
     * 
     * Work on a {@link Scheduler.Worker} is guaranteed to be sequential.
     *  {@link Scheduler.Worker} 上面的work保證是順序執行的
     */

複製程式碼

現在我們知道了：我們通過呼叫subscribeOn()傳入Scheduler，當下遊ObservableSource被訂閱時（請注意，訂閱順序是由下到上的），距離最近的執行緒排程subscribeOn()方法中，儲存的Scheduler會建立一個worker(對應相應的執行緒，本文中為IoScheduler)，在其對應的執行緒中，立即執行task

多次subscribeOn()

現在考慮一個問題，假如在我們的程式碼中，多次使用了subscribeOn()程式碼，到執行緒會怎麼處理呢？

上文已經講到了，不管我們怎麼通過subscribeOn()方法切換執行緒，由於訂閱執行順序是由下到上，因此當最上游的ObservableSource被訂閱時，所線上程當然是距離上游最近的subscribeOn()所提供的執行緒，即最終Observable總是在第一個subscribeOn()所在的執行緒中執行。

二、observeOn()

先看observeOn()內部，果然是hook+Observable的包裝類：

public final Observable<T> observeOn(Scheduler scheduler) {
        return observeOn(scheduler, false, bufferSize());
    }

    public final Observable<T> observeOn(Scheduler scheduler, boolean delayError, int bufferSize) {
        ObjectHelper.requireNonNull(scheduler, "scheduler is null");
        ObjectHelper.verifyPositive(bufferSize, "bufferSize");

        //例項化ObservableObserveOn物件並返回
        return RxJavaPlugins.onAssembly(new ObservableObserveOn<T>(this, scheduler, delayError, bufferSize));
    }

複製程式碼

再看ObservableObserveOn：

public final class ObservableObserveOn<T> extends AbstractObservableWithUpstream<T, T> {
    final Scheduler scheduler;
    final boolean delayError;
    final int bufferSize;
    public ObservableObserveOn(ObservableSource<T> source, Scheduler scheduler, boolean delayError, int bufferSize) {
        super(source);
        //1.相關依賴注入
        this.scheduler = scheduler;
        this.delayError = delayError;
        this.bufferSize = bufferSize;
    }

    @Override
    protected void subscribeActual(Observer<? super T> observer) {
        if (scheduler instanceof TrampolineScheduler) {
            source.subscribe(observer);
        } else {
            //2.建立主執行緒的worker
            Scheduler.Worker w = scheduler.createWorker();
            //3.上游資料來源被訂閱
            source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
        }
    }
}

複製程式碼

和subscribeOn()不同的是，我們並不是立即在對應的執行緒執行task，而是將對應的執行緒(實際上是worker)作為引數，例項化ObserveOnObserver並存儲起來。

當上遊的資料傳遞過來時，ObserveOnObserver執行對應的方法，比如onNext(T)，再切換到對應執行緒中，並交由下游的Observer去接收：

ObserveOnObserver

ObserveOnObserver中程式碼極多，我們簡單瞭解原理後，以onNext（T）為例：

static final class ObserveOnObserver<T> extends BasicIntQueueDisposable<T>
    implements Observer<T>, Runnable {

        //...省略其他程式碼
        ObserveOnObserver(Observer<? super T> actual, Scheduler.Worker worker, boolean delayError, int bufferSize) {
            this.actual = actual;
            this.worker = worker;
            this.delayError = delayError;
            this.bufferSize = bufferSize;
        }
        //佇列
        SimpleQueue<T> queue;

       @Override
        public void onNext(T t) {
            if (done) {
                return;
            }
            //將資料存入佇列
            if (sourceMode != QueueDisposable.ASYNC) {
                queue.offer(t);
            } 
            //對應執行緒取出資料並交由下游的Observer
            schedule();
        }

        void schedule() {
            if (getAndIncrement() == 0) {
                worker.schedule(this);
            }
        }
         //...省略其他程式碼
}

複製程式碼

多次observerOn()

由上文得知，與subscribeOn()相反，observerOn()操作會將切換到對應的執行緒，然後交由下游的Observer處理，因此observerOn()僅對下游的Observer生效，並且，如果多次呼叫，observerOn()的執行緒排程會持續到下一個observerOn()操作之前。

總結

subscribeOn()

  • 訂閱順序當從下到上，上游的ObservableSource被訂閱時，先切換執行緒，然後立即執行task;

  • 當存在多個subscribeOn()方法時，僅第一個subscribeOn()有效。

observerOn()

  • 訂閱順序當從下到上，上游的ObservableSource被訂閱時，會將對應的worker建立並作為構造引數儲存在Observer的裝飾器中，並不會立即切換執行緒；

  • 當資料由上游傳送過來時，先將資料儲存到佇列中，然後切換執行緒，然後在新的執行緒中將資料傳送給下游的Observer；

  • 當存在多個observerOn()方法時，僅對距下游下一個observerOn()之前的observer有效

有興趣可以關注我的小專欄，學習更多知識：小專欄

相關推薦

RxJava2.0(四)執行緒之間切換的內部原理

基本程式碼 來看一下基本程式碼： Observable.create((ObservableOnSubscribe<Integer>) e -> { e.onNext(1); e.onNext(2); e.onCo

android RXJava2.0(二)--執行緒控制

正題 還是以之前的例子, 兩根水管: 正常情況下, 上游和下游是工作在同一個執行緒中的, 也就是說上游在哪個執行緒發事件, 下游就在哪個執行緒接收事件. 怎麼去理解呢, 以Android為例, 一個Activity的所有動作預設都是在主執行緒中執行的,

通俗易懂地給女朋友講：執行緒池的內部原理

餐廳的約會 餐盤在燈光的照耀下格外晶瑩潔白，女朋友拿起紅酒杯輕輕地抿了一小口，對我說：“經常聽你說執行緒池，到底執行緒池到底是個什麼原理？”我楞了一下，心裡想女朋友今天是怎麼了，怎麼突然問出這麼專業的問題，但做為一個專業人士在女朋友面前也不能露怯啊，想了一下便說：“我先給你講講我前同事老王的故事吧！”歡迎

執行緒之間的協作

當多個執行緒可以一起工作去解決某個問題時，如果某些部分必須在其它部分之前完成，那麼就需要對執行緒進行協調。 join() 線上程中呼叫另一個執行緒的 join() 方法，會將當前執行緒掛起，而不是忙等待，直到目標執行緒結束。 對於以下程式碼，雖然 b 執行緒先啟動，但是因為在 b 執行緒中呼叫了 a 執

執行緒之間的通訊方式：wait/notify

1.什麼是執行緒間的通訊 通訊，顧名思義就是一種通知交通的方式，在多執行緒的環境下，如果各個執行緒之間可以互相通訊的話，可以很好地提高工作效率，提高CPU的利用率。 2.執行緒間常用的通訊方式 多執行緒間的通訊一般採取等待/通知機制進行實現，即Object類中的wait()和notify

多執行緒學習-----------執行緒之間的通訊(三)

執行緒安全問題(demo:銀行轉賬):多個執行緒操作相同一個數據的時候就會出現執行緒安全問題。 程式舉例: public class SynchronizedStudy { class Output{ public void printName(String name){

java基礎學習總結（二十）：多執行緒上下文切換

什麼是上下文切換        即使是單核CPU也支援多執行緒執行程式碼，CPU通過給每個執行緒分配CPU時間片來實現這個機制。時間片是CPU分配給各個執行緒的時間，因為時間片非常短，所以CPU通過不停地切換執行緒執行，讓我們感覺多個執行緒時同時執行的，時

java併發程式設計一一多執行緒之間通訊(一)

1.多執行緒之間如何實現通訊 多執行緒之間通訊，其實就是多個執行緒在操作同一個資源，但是操作的動作不同。 1.1什麼是多執行緒之間通訊？ 需求：第一個執行緒寫入（input）使用者，另一個執行緒讀取（out）使用者。實現讀一個，寫一個操作。 1.2多執行緒之間通訊需求？

java多執行緒之間的通訊

一、使用wait()和notify()方法結合實現通訊 注意：wait()和notify()必須在synchronized中使用 wait():是當前執行緒休眠並且釋放鎖。 notify():喚醒另一個執行緒 例項：input執行緒向共享資料中寫入一條資料，out執行緒重共享資料中讀

【Linux】多程序與多執行緒之間的區別

http://blog.csdn.net/byrsongqq/article/details/6339240 網路程式設計中設計併發伺服器，使用多程序與多執行緒 ，請問有什麼區別？  答案一: 1，程序：子程序是父程序的複製品。子程序獲得父程序資料空間、堆和棧的複製品。 2，執行緒：相

Jsoup簡單例子2.0——多執行緒爬取網頁內的郵箱

上一篇文章講了利用Jsoup爬取貼吧帖子裡的郵箱，雖然爬取成功了，但我對效率有所追求。10頁的帖子爬取了兩百多個郵箱，最快用時8秒，一般需要9秒。在思考了一下怎麼提升效率後，決定採用多執行緒的方式爬取網頁內的郵箱。廢話不多說，直接上程式碼。 引入Jsoup的jar包此處省略，沒有的可以檢視上篇文

執行緒之間的資料共享？

1.同一個程序中的不同執行緒之間資料是否共享。 一個執行緒的資料在另一個執行緒中修改其值，然後再在該執行緒中列印值發生了變化。所以說同一個程序中的執行緒之間的資料是共享的。 2.同一個程序中的不同執行緒之間是否共享檔案描述符？ 對於同一個檔案描述符，一個執行緒寫，一個執行緒

死鎖|執行緒之間的互動，wait和notify

A執行緒佔用物件蓋倫 B執行緒佔用物件提莫 A試圖佔用提莫，等待B釋放cpu資源 B同理 相互等待，GG ————————————————————————————————————————————————————————— 同樣加血減血 當hurt hp==1時就要停止，當

多執行緒之間實現同步

為什麼有執行緒安全問題 　　當多個執行緒同時共享同一個全域性變數或靜態變數，做寫的操作時，可能會發生資料衝突問題，也就是執行緒安全問題。但是做讀操作是不會發生資料衝突問題。 什麼是多執行緒之間同步 　　當多個執行緒共享同一個資源,不會受到其他執行緒的干擾。 需求說明：2個視窗同時賣100張火車票問題

瀏覽器核心中各個執行緒之間的關係

  到了這裡，已經對瀏覽器的執行有了一個整體的概念，接下來，先簡單梳理一些概念 :   GUI渲染執行緒與JS引擎執行緒互斥   由於JavaScript是可操縱DOM的，如果在修改這些元素屬性同時渲染介面（即JS執行緒和UI執行緒同時執行），那麼渲染執行緒前後獲得

併發程式設計（一）：執行緒基礎、執行緒之間的共享與協作

一、基礎概念 1.1 CPU核心數、執行緒數 **兩者的關係：**cpu的核心數與執行緒數是1:1的關係，例如一個8核的cpu，支援8個執行緒同時執行。但在intel引入超執行緒技術以後，cpu與執行緒數的關係就變成了1:2。此外在開發過程中並沒感覺到執行緒的限制，那是因為cpu

sysbench-1.0 多執行緒插入資料

在做MySQL oltp相關的測試需要進行資料準備，sysbench提供了多執行緒插入資料的方法，如下可以參考如下指令碼 sysbench ./lua/oltp_insert.lua \ --mysql-db=ashe \ --mysql-host=100.110.0.9 \ --m

任務、程序、執行緒之間的區別

          任務（task）是最抽象的,是一個一般性的術語,指由軟體完成的一個活動。一個任務既可以是一個程序,也可以是一個執行緒。簡而

程序與執行緒之間的關係

程序和執行緒之間的關係 轉載來自：https://www.cnblogs.com/losing-1216/p/5083097.html 程序概念 　　程序是表示資源分配的基本單位，又是排程執行的基本單位。例如，使用者執行自己的程式，系統就建立一個程序，併為它分配資源，包括各種表格、

Java——多執行緒基本使用（三） 餓漢式和懶漢式的單例設計模式，多執行緒之間的通訊

這一則部落格主要寫的是單例設計模式，與實現多執行緒之間的通訊等等~ 1.單例設計模式：保證類在記憶體中只有一個物件 2.保證類在記憶體中只有一個物件            &