RxJava线程切换之subscribeOn()与observeOn

我们都知道android在日常开发时都是在IO线程执行耗时操作,然后在UI线程进行更新UI,那么RxJava怎么进行线程切换的呢?这就用到了subscribeOn和observeOn这两个操作符。首先我们来看下官方文档对着两个操作符的解释:

By default, an Observable and the chain of operators that you apply to it will do its work, and will notify its observers, on the same thread on which its Subscribe method is called. 

大致翻译下,默认情况下,一个被观察者和你应用到这条链上的操作符将会执行而且执行完成之后会通知观察者,这些操作都是发生在同一线程中,该线程就是订阅操作被调用的线程,我们来举个例子:

    Flowable.just(1)
                .map(new Function<Integer, String>() {

                    @Override
                    public String apply(Integer integer) throws Exception {
                        System.out.println(Thread.currentThread().getName());
                        return String.valueOf(integer);
                    }
                })
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        System.out.println(Thread.currentThread().getName());
                    }
                });

可以看到打印的都是在主线程,因为我们订阅的时候是在主线程。
我们在指定下线程加入subscribeOn

        Flowable.just(1)
                .map(new Function<Integer, String>() {

                    @Override
                    public String apply(Integer integer) throws Exception {
                        System.out.println(Thread.currentThread().getName());
                        return String.valueOf(integer);
                    }
                })
                .subscribeOn(Schedulers.newThread())
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        System.out.println(Thread.currentThread().getName());
                    }
                });
image.png

可以看到都是在子线程中执行,这就印证上面的结论。

The SubscribeOn operator changes this behavior by specifying a different Scheduler on which the Observable should operate. The ObserveOn operator specifies a different Scheduler that the Observable will use to send notifications to its observers.

SubscribeOn操作符指定被观察者发送数据的线程,ObserveOn指定被观察者发送通知给观察者的线程,也就是观察者执行操作的线程。

        Flowable.just(1)
                .map(new Function<Integer, String>() {

                    @Override
                    public String apply(Integer integer) throws Exception {
                        System.out.println(Thread.currentThread().getName());
                        return String.valueOf(integer);
                    }
                })
                .subscribeOn(Schedulers.newThread())
                .observeOn(Schedulers.io())
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        System.out.println(Thread.currentThread().getName());
                    }
                });
image.png

通过subscribeOn指定发送数据的线程,通过ObserveOn指定观察者接收通知的线程
再来个复杂一点的

        Flowable.just(1)
                .map(new Function<Integer, String>() {

                    @Override
                    public String apply(Integer integer) throws Exception {
                        System.out.println(Thread.currentThread().getName());
                        return String.valueOf(integer);
                    }
                })
                .subscribeOn(Schedulers.newThread())
                .subscribeOn(Schedulers.io())
                .subscribeOn(Schedulers.io())
                .observeOn(Schedulers.newThread())
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        System.out.println(Thread.currentThread().getName());
                    }
                });
image.png

说明调用多次subscribeOn,只有第一次生效。那么多次调用observeOn呢

        Flowable.just(1)
                .map(new Function<Integer, String>() {

                    @Override
                    public String apply(Integer integer) throws Exception {
                        System.out.println("第一个map执行线程: " + Thread.currentThread().getName());
                        return String.valueOf(integer);
                    }
                })
                .observeOn(Schedulers.io())
                .map(new Function<String, String>() {
                    @Override
                    public String apply(String s) throws Exception {
                        System.out.println("第二个map执行线程: " + Thread.currentThread().getName());
                        return s;
                    }
                })
                .subscribeOn(Schedulers.newThread())
                .observeOn(Schedulers.io())
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        System.out.println("Consumer执行线程:  " + Thread.currentThread().getName());
                    }
                });
image.png

observeOn指定了其后代码执行的线程,多次调用则多次切换。第一个map执行的线程是subscribeOn执行的线程,subscribeOn指定的是发射线程,而第一个map前并未通过observeOn在重新指定线程,所以第一个map执行在subscribeOn指定的默认线程RxNewThreadScheduler-1,第二个map上执行了observeOn进行线程切换,所以第二个map执行在RxCachedThreadScheduler-2线程,又因为subscribe之前又调用了observeOn,所以subscribe执行在其上面observeOn指定的线程。

由此可得知,observeOn可达到切换线程的作用,而subscribeOn只是执行了发射数据操作执行的线程。
结尾附上官方文档的一段话和一个图

As shown in this illustration, the SubscribeOn operator designates which thread the Observable will begin operating on, no matter at what point in the chain of operators that operator is called. ObserveOn, on the other hand, affects the thread that the Observable will use below where that operator appears. For this reason, you may call ObserveOn multiple times at various points during the chain of Observable operators in order to change on which threads certain of those operators operate.
最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
平台声明:文章内容(如有图片或视频亦包括在内)由作者上传并发布,文章内容仅代表作者本人观点,简书系信息发布平台,仅提供信息存储服务。

推荐阅读更多精彩内容

  • 作者寄语 很久之前就想写一个专题,专写Android开发框架,专题的名字叫 XXX 从入门到放弃 ,沉淀了这么久,...
    戴定康阅读 7,671评论 13 85
  • 转一篇文章 原地址:http://gank.io/post/560e15be2dca930e00da1083 前言...
    jack_hong阅读 950评论 0 2
  • 前言我从去年开始使用 RxJava ,到现在一年多了。今年加入了 Flipboard 后,看到 Flipboard...
    占导zqq阅读 9,205评论 6 151
  • 在正文开始之前的最后,放上GitHub链接和引入依赖的gradle代码: Github: https://gith...
    苏苏说zz阅读 701评论 0 2
  • 2014.09.26入职旺旺,参加新人训之后开始做旺旺精神操,走路要走黄线内,上班需要带工号牌。从开始的莫名其妙...
    swallow林燕阅读 719评论 0 0