拆轮子系列--RxJava理解（二）--subscribeOn

本系列文章如下：

上一篇文章主要介绍了RxJava中基本的调用流程以及常用的操作符--map。本文主要介绍RxJava中线程调度的核心操作符之一subscribeOn，本文源码分析基于RxJava2。
本文的大纲如下：

  • 一个具体的例子
  • subscribeOn源码分析
  • 总结

1 . 一个具体的例子

Observable.create(new ObservableOnSubscribe<String>() {

            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                e.onNext("1");
                e.onNext("2");
                e.onNext("3");
                e.onComplete();
            }
        }).map(new Function<String, Integer>() {
            @Override
            public Integer apply(String str) throws Exception {
               Log.e("TAG", "-map-thread-" + Thread.currentThread().getName() + "-string-" + s);
                return Integer.valueOf(s);
            }
        }).subscribeOn(Schedulers.io())
             .subscribe(new Consumer<Integer>() {
              @Override
              public void accept(Integer integer) throws Exception {
                       Log.e("TAG", "-sub-" + Thread.currentThread().getName() + "-integer-" + integer);
                    }
                });
    }

运行下上面的代码，我们先看看结果会是怎么样的：

E/TAG: -map-thread-RxCachedThreadScheduler-1-string-1
E/TAG: -sub-RxCachedThreadScheduler-1-integer-1
E/TAG: -map-thread-RxCachedThreadScheduler-1-string-2
E/TAG: -sub-RxCachedThreadScheduler-1-integer-2
E/TAG: -map-thread-RxCachedThreadScheduler-1-string-3
E/TAG: -sub-RxCachedThreadScheduler-1-integer-3

如果我们调用多次的subscribeOn()会是怎么样子的呢？我们改下代码看看：

Observable.create(new ObservableOnSubscribe<String>() {

            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                e.onNext("1");
                e.onNext("2");
                e.onNext("3");
                e.onComplete();
            }
        })
                .subscribeOn(AndroidSchedulers.mainThread())
                .map(new Function<String, Integer>() {
                    @Override
                    public Integer apply(String s) throws Exception {
                        Log.e("TAG", "-map-thread-" + Thread.currentThread().getName() + "-string-" + s);
                        return Integer.valueOf(s);
                    }
                })
                .subscribeOn(Schedulers.io())
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.e("TAG", "-sub-" + Thread.currentThread().getName() + "-integer-" + integer);
                    }
                });
    }

我们再来看看打印的信息：

E/TAG: -map-thread-main-string-1
E/TAG: -sub-main-integer-1
E/TAG: -map-thread-main-string-2
E/TAG: -sub-main-integer-2
E/TAG: -map-thread-main-string-3
E/TAG: -sub-main-integer-3

有没有发现，当我们调用两次的subscribeOn()这个操作符，只有最上面的那个起了作用，后面的subscribeOn()操作符无效。这个是为什么呢？接下来进行源码分析。

2. subscribeOn源码分析

首先，我们来看看subscribeOn()这个操作符到底干了些什么事情，然后再来解释为什么多次调用subscribeOn()这个操作符，只有第一个操作符有效。

@CheckReturnValue
@SchedulerSupport(SchedulerSupport.CUSTOM)
public final Observable<T> subscribeOn(Scheduler scheduler) {
   ObjectHelper.requireNonNull(scheduler, "scheduler is null");
   return RxJavaPlugins.onAssembly(new ObservableSubscribeOn<T>(this, scheduler));
    }

我们首先看看onAssembly()这个里面到底干了些什么事情：

public static <T> Observable<T> onAssembly(Observable<T> source) {
        Function<Observable, Observable> f = onObservableAssembly;
        if (f != null) {
            return apply(f, source);
        }
        return source;
    }

从源码中我们可以看出，它并没有干什么事情，就是返回了一个Observable的对象，接下来我们看看ObservableSubscribeOn()这部分的源码：

public ObservableSubscribeOn(ObservableSource<T> source, Scheduler scheduler) {
        super(source);
        this.scheduler = scheduler;
    }

    @Override
    public void subscribeActual(final Observer<? super T> s) {
    final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(s);
    s.onSubscribe(parent);
    parent.setDisposable(scheduler.scheduleDirect(new Runnable() {
         @Override
         public void run() {
              source.subscribe(parent);
         }
     }));
  }

在上面的代码中，真正的执行操作是在subscribeActual()方法中，该方法中首先执行的操作就是通过代理的方式使用SubscribeOnObserver将Observer进行封装，设置Disposable来将subscribe进行线程切换到scheduler的线程中。
简单点来说，subscribeOn这个操作符就是将上一层的ObservableSource(就是上一层的Observable)放到一个新的线程去发射元素。
ok，这里也就解释了为什么调用多个subscribeOn()时，只有最上面的subscribeOn()起作用，那么根据上面的分析，当我们在操作的时候加入多个map()操作符的时候，情况会是怎么样子的？如下示例：

Observable.create(new ObservableOnSubscribe<String>() {

            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                e.onNext("1");
                e.onNext("2");
                e.onComplete();
            }
        })
                .map(new Function<String, Integer>() {
                    @Override
                    public Integer apply(String s) throws Exception {
                        Log.e("TAG", "-map-thread-" + Thread.currentThread().getName() + "-string-" + s);
                        return Integer.valueOf(s);
                    }
                })
                .subscribeOn(Schedulers.io())
                .map(new Function<Integer, Long>() {
                @Override
                public Long apply(Integer integer) throws Exception {
                        Log.e("TAG", "-map-thread-" + Thread.currentThread().getName() + "-Integer-" + integer);
                        return Long.valueOf(integer);
                    }
                })
                .subscribe(new Consumer<Long>() {
                @Override
                public void accept(Long longValue) throws Exception {
                        Log.e("TAG", "-sub-" + Thread.currentThread().getName() + "-longValue-" + longValue);
                    }
                });

结果如下:

E/TAG: -map-thread-RxCachedThreadScheduler-1-string-1
E/TAG: -map-thread-RxCachedThreadScheduler-1-Integer-1
E/TAG: -sub-RxCachedThreadScheduler-1-longValue-1
E/TAG: -map-thread-RxCachedThreadScheduler-1-string-2
E/TAG: -map-thread-RxCachedThreadScheduler-1-Integer-2
E/TAG: -sub-RxCachedThreadScheduler-1-longValue-2

从结果中我们发现，不管执行了多少次的map()操作，每个Observable只有第一个subscribeOn()方法起作用，而且多次的map()操作不会影响线程的变换。

3. 总结

首先，用一张图来展示subscribeOn()这个操作符基本运行的流程：

subscribeOn基本流程.png

subscribeOn这个操作符主要的执行逻辑现在总结如下：

    1. 代理Observable做的事情就是将Observable1发射到scheduler指定的线程中进行subscribe()，并且调用下一层传上来的observer1来接收事件；
    1. 当我们多次调用subscribeOn()方法后，其实此时的事件并没有发射出去，当执行到第一个subscribeOn()方法时，后面subscribeOn()切换线程就会被第一个subscribeOn()切换线程的方法截断，因此，多次调用subscribeOn()对整个流程是不产生影响的，从而这里可以解释前面的问题，当使用多个subscribeOn()的时候，也只有第一个subscribeOn()起作用。

再举个浅显的栗子来说明下多次调用subscribeOn的情况：

new Thread(new Runnable() {
            @Override
            public void run() {
                new Thread(new Runnable() {
                    @Override
                    public void run() {
                        new Thread(new Runnable() {
                            @Override
                            public void run() {
                                Log.e("TAG", "which thread?=" + Thread.currentThread().getName());
                            }
                        }, "线程3").start();
                    }
                }, "线程2").start();
            }
        }, "线程1").start();

上面代码执行的结果：

E/TAG: which thread?=线程3
    1. 当我们使用subscribeOn()且使用多个map()操作符的时候，subscribeOn会将上一层Observable切换到一个指定的线程进行事件的发射，而上一层Observable拿到的observer已经是由下层observer一路封装上来的，也就是该observer经历了所有的转换。因此，不管执行了多少次的map()操作，
      不管该map()操作是执行在subscribeOn()之前还是之后，其最后的发射事件会在subscribeOn()指定的线程执行。

RxJava中线程调度方法subscribeOn()方法这里就介绍完毕，下一篇继续分析RxJava中另一个线程调度方法observableOn()。

如果文章中有什么疏漏或者错误的地方，还望各位指正，你们的监督是我最大的动力，谢谢！

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 214,776评论 6赞 496
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 91,527评论 3赞 389
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 160,361评论 0赞 350
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 57,430评论 1赞 288
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 66,511评论 6赞 386
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 50,544评论 1赞 293
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 39,561评论 3赞 414
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 38,315评论 0赞 270
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 44,763评论 1赞 307
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 37,070评论 2赞 330
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 39,235评论 1赞 343
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 34,911评论 5赞 338
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 40,554评论 3赞 322
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 31,173评论 0赞 21
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 32,424评论 1赞 268
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 47,106评论 2赞 365
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 44,103评论 2赞 352