专栏首页刘晓杰再忆RxJava---线程切换

再忆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其实也没错,但是从学术角度来讲不准确)。

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • AIDL(续)

    http://blog.csdn.net/lxj1137800599/article/details/50998757 这篇文章讲的是在不同的工程文件中实现...

    提莫队长
  • 桥接模式详解

    桥接(Bridge)模式的定义如下:将抽象与实现分离,使它们可以独立变化。它是用组合关系代替继承关系来实现,从而降低了抽象和实现这两个可变维度的耦合度。 具体...

    提莫队长
  • okhttp拦截器的使用------修改baseurl

    有的时候做项目的时候会遇到调用接口的url和之前接口定义的url不一样,很可能连retrofit已经设置好的baseurl都要换,但是retrofit并没有提供...

    提莫队长
  • Android利用Andserver搭建服务器的详细教程

      在手机上有一个功能,但是需要使用电脑控制什么时候开始这个功能,然后上网搜索,找到了AndServer可以满足要求,我只是实现了简单的功能,如果感兴趣的话,可...

    砸漏
  • Android全能开源项目xUtils3开发教程、简单封装

    码农笔录
  • Android全能开源项目xUtils3开发教程、简单封装

    用户2235302
  • tomcat源码解读三(1) tomcat的jmx管理

        JMX即Java 管理扩展(Java Management Extensions,JMX)用来管理检测 Java 程序(同时 JMX 也在 J2EE 1...

    cfs
  • P1-Android基于MVP实现号码归属地查询

    张风捷特烈
  • Android中网络框架简单封装的实例方法

    Android作为一款主要应用在移动终端的操作系统,访问网络是必不可少的功能。访问网络,最基本的接口有:HttpUrlConnection,HttpClient...

    砸漏
  • 设计模式之设计原则

    单一职责原则:类的职责单一,不能将太多的职责放在一个类中,该原则是实现高内聚、低耦合的指导方针

    路行的亚洲

扫码关注云+社区

领取腾讯云代金券