再忆RxJava---线程切换

RxJava已经陆陆续续看了将近两年多了,觉得很有必要重新认识一下。而且最新github上都出了3.0了。重新看一下,一来作为回顾,二来也算是学习一下3.0
在了解线程切换前,先回顾一下简单的同步操作

1.同步

        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) {
                // onNext
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(String s) {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });

Observable.create会生成一个ObservableCreate,subscribe最终会调用ObservableCreate的subscribeActual方法
很明显,同步的原理很简单。Observable.create生成ObservableCreate,ObservableCreate会new一个emitter出来,这个emitter就包裹了Observer,通过emitter来完成Observer函数的调用

2.异步

        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) {

            }
        }).subscribeOn(new IoScheduler())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(String s) {

                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });

Observable.create还是一样的,生成ObservableCreate
subscribeOn会生成ObservableSubscribeOn
observeOn会生成ObservableObserveOn
最后subscribe就会调用各自的subscribeActual方法

2.1 ObservableObserveOn

    @Override
    protected void subscribeActual(Observer<? super T> observer) {
        if (scheduler instanceof TrampolineScheduler) {
            source.subscribe(observer);
        } else {
            Scheduler.Worker w = scheduler.createWorker();

            source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
        }
    }

这个observer就是我们自己写的Observer
scheduler就是HandlerScheduler(里面有Handler)
然后触发下一个subscribeActual
说白了就是我自己写的Observer被ObserveOnObserver包裹了一层

2.2 ObservableSubscribeOn的subscribeActual方法

    @Override
    public void subscribeActual(final Observer<? super T> observer) {
        final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(observer);

        observer.onSubscribe(parent);

        parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
    }

会用SubscribeOnObserver来包裹上一层的Observer(ObserveOnObserver),scheduler很明显就是IoScheduler
IoScheduler其实就是生成线程池处理SubscribeTask
SubscribeTask代码里面( source.subscribe(parent) )source就是ObservableCreate,所以就执行我们自己写的subscribe方法
总结:生成一个线程池,启动一个线程处理我们自己写的subscribe方法。

2.3 ObservableCreate

这个其实就和同步的一样

3 具体的流程图如下

异步线程切换流程图.png

简单来说,就是把所有的操作从右到左包装成一对观察者与被观察者的关系,然后通过发射器使所有的操作连续执行

4 流程举例

4.1 下载并显示图片

  1. 创建CreateEmitter,下载,等到结束,发射onNext。这是出于流程图中的(1),onNext在子线程中发射(网络请求一般会自己new Thread出来执行的)

(注意:此时已经有子线程处理了,所以subscribeOn其实已经没有意义了,可以不写。直接看observeOn)

  1. 接下来走到ObserveOnObserver的onNext方法
    ObserveOnObserver本身也是Runnable,所以会执行自身的run方法。通过源码可以很清楚的知道,其实就是把上一个步骤中的结果收集起来,放到队列里,然后poll出来。poll结束就直接onNext。(有两点需要说明)
  • 这里还区分同步异步。同步不进入队列,q.poll=null,直接onNext。异步进队列,直到q.poll != null 才会onNext
  • 传进来onNext的时候,是处于下载线程中,传出去onNext已经经过Handler处理
  1. poll结束就走到我们自己写的Observer的onNext方法

4.2 批量处理图片并显示

这个时候是需要subscribeOn的,毕竟批量处理图片的时候,subscribe函数中我们不会自己去new Thread出来
(简单来说,如果subscribe函数中的耗时操作没有new Thread,那么就需要subscribeOn)
是否需要subscribeOn还需要注意创建retrofit的时候RxJava2CallAdapterFactory的创建方式,如果用了createWithScheduler,那么就不用写subscribeOn了,里面默认有了,具体可参见源码

5 subscribeOn和ObserveOn

ObserveOn每次都有效是因为每次都有scheduler.createWorker操作,举例来说,ComputationScheduler以及其他的IoScheduler都会生成线程池来操作,那么必然会跑到对应的线程池中的线程中操作
下面我们来谈谈subscribeOn只有首次有效的问题

Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(final ObservableEmitter<String> emitter) {
                emitter.onNext("111");
                emitter.onComplete();
            }
        }).subscribeOn(new IoScheduler())
                .map(new Function<String, String>() {
                    @Override
                    public String apply(String s) {
                        Log.e("IoScheduler", Thread.currentThread().getName());
                        return "555";
                    }
                })
                .subscribeOn(new ComputationScheduler())
                .map(new Function<String, String>() {
                    @Override
                    public String apply(String s) {
                        Log.e("ComputationScheduler", Thread.currentThread().getName());
                        return "222";
                    }
                })
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(String s) {
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });

两个log打印出来的都是相同的值,很明显,只有首次“有效”。其实真的是有没有效的问题么?我在ObservableSubscribeOn的subscribeActual方法里面debug,发现ComputationScheduler在主线程,IoScheduler在Computation线程中,subscribe方法在IoScheduler线程中,很明显,其实是有效的,确实切换了。那为什么会有无效的说法呢?其实也很好理解,我们的操作在A线程中执行,而A在线程B中执行,请问,我们的操作在哪个线程中执行?肯定是A啊(说B其实也没错,但是从学术角度来讲不准确)。

最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 216,544评论 6 501
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 92,430评论 3 392
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 162,764评论 0 353
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 58,193评论 1 292
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 67,216评论 6 388
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 51,182评论 1 299
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 40,063评论 3 418
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 38,917评论 0 274
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 45,329评论 1 310
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 37,543评论 2 332
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 39,722评论 1 348
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 35,425评论 5 343
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 41,019评论 3 326
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 31,671评论 0 22
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 32,825评论 1 269
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 47,729评论 2 368
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 44,614评论 2 353

推荐阅读更多精彩内容

  • 很久不写命题作文了,今日想着根据命题写一篇吧,于是就有了这个题目,我的可爱朋友。 说实话,到了不惑之年,朋友越来越...
    Jessy自由行走的猫阅读 166评论 0 1
  • 金融消费者自行决策,自助选择资产,自我承担决策风险,这真是一条行得通的路吗? 向个人推送金融资产,如果推得动,你确...
    怪物办公室阅读 322评论 0 0