标签：做了 作用 map操作 任务 nbsp code 设置 run 原理
一、概述
本节将分析RxJava2的线程切换模型。通过对线程切换源代码的分析到达对RxJava2线程切换彻底理解的目的。通过对本节的学习你会发现，RxJava2线程切换是如此的简单，仅仅是通过两个操作符就能完成从子线程到主线程，或者主线程到子线程，再或者从子线程到子线程的切换。对应的操作符为：observerOn：指定观察者运行的线程。subscribeOn：执行被观察者运行的线程。
二、简单例子入手
private void threadSwitchTest() {
Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
emitter.onNext("《深入Java虚拟机》");
MyLog.log("Thread:" + Thread.currentThread().getName());
}
});
observable
.observeOn(AndroidSchedulers.mainThread())//被观察者执行线程
.subscribeOn(Schedulers.io())//观察者执行线程
.subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
MyLog.log("Thread:" + Thread.currentThread().getName());
}
});
}
以上例子中我们使用observeOn(AndroidSchedulers.mainThread())来指定观察者运行在主线程，使用subscribeOn(Schedulers.io())来指定被观察运行在子线程
三、源码分析
本节针对RxJava2的源代码我们需要弄明白三件事情：
1.子线程如何切换到主线程原理分析
2.主线程如何切换到子线程原理分析
3.子线程如何切换到子线程原理分析
通过上一节的分析我们知道RxJava2通过创建一个被观察者(ObservableCreate)和一个观察者(LambdaObserver)，并实现观察者和被观察者的绑定。通过ObservableEmitter.onNext发送消息，Consumer.accept中接收消息。而操作符map仅仅是对被观察者ObservableCreate做了一层包装（装饰模式），变成了ObservableMap。而观察者装饰后则变成了MapObserver。
很显然，observeOn和subscribeOn都属于操作符（他们都是用来做线程切换的操作符而已），所以这两个操作符也符合上面Map操作符的包装规则。
subscribeOn源码分析：
从上述源码可以看出subscribeOn确实如上面所说，会被包装成为一个ObservableSubscribeOn。其构造方法会传入两个参数，一个是this：代表当前被观察者，也就是操作符上面修饰的那个被观察者，本例中指的是ObservableObserveOn,ObservableObserveOn又装饰了ObservableCreate。scheduler指的是Schedulers.io(), 指被观察者运行在io线程，也就是子线程中。
下面看下Schedulers类是个什么东西。
Schedulers内部封装了各种Scheduler。每一个Scheduler中都封装的有线程池，用于执行后台任务。
到此处ObservableObserveOn对象也就创建完成了。
下面看下ObserveOn操作符都干了什么事情：
ObserveOn方法内部包装了一个ObservableObserveOn对象，其有两个参数，this：代表当前Observable对象，此处指的是ObservableCreate这个对象，scheduler代表的是AndroidSchedulers.mainThread()。
我们看一下AndroidSchedulers的源代码，看它都干了写什么事情
AndroidSchedulers的内部类MainHolder的作用是在主线程中创建一个Handler。由new Handler(Looper.getMainLooper())来完成。因为Looper所在的线程为Handler所在的线程，又因为Looper.getMainLooper()获取到的是主线程的looper，所以当前Handler运行在主线程，顺带着这块的逻辑也是在主线程中完成的。字段MAIN_THREAD仅仅是把HandlerScheduler返回而已，而HandlerScheduler持有主线程handler。那么manThread()方法就好理解了 ，就是返回了一个持有主线程Handler的Scheduler而已。
所以ObservableSubscribeOn包装了ObservableCreate并持有了主线程Handler。到此被观察者就已经创建完成了。
下面说下观察者Consumer.accept方法在这个链式调用中是如何被执行的：
1.经过上面的分析被观察者已经变为：ObservableObserveOn，ObservableObserveOn持有ObservableSubscribeOn对象的引用，ObservableSubscribeOn又持有ObservableCreate的引用。所以Observable对象经过三层包装最终成为了ObservableObserveOn。
2.Observable.subscribe(Consumer consumer)方法执行订阅，会把原始的观察者对象LambdaObserver对象包装成为SubscribeObserveOn对象。用以在ObservableSubscribeOn对象执行subscribeActual方法的时候正式执行绑定操作。至此，观察者和被观察者建立了绑定关系。
3.被观察者在执行ObservableOnSubscribe实例的subscribe方法的ObservableEmitter参数的onNext方法的时候,会首先调用SubscribeOnObserver的onNext方法，又由于SubscribeOnObserver持有ObserveOnObserver的引用，因此在SubscribeOnObserver的onNext方法中又会调用ObserveOnObserver对象的onNext方法，在此Next方法中又会调用LambdaObserver的onNext方法，然后在LambdaObserver的onNext方法中又会调用Consumer.accept方法。最后完成数据的从发送到接收的流转。
了解了以上操作符的整体流转流程后，我们接下来回过头来看开头我们提出的三个问题：
1.主线程切换到子线程
我们先来看ObservableSubscribeOn这个类，在上面的小例子中，直接将被观察者运行在IO线程中了。我们直接看ObservableSubscribeOn的subscribeActual方法的源代码
此处 SubscribeOnObserver是观察者。scheduler就是上面参数中的Schedulers.io()。scheduler.schedulDirect接收一个Runnable类型的SubscribeTask。在scheduleDirect方法内部会开启一个线程池来执行这个异步任务，如下代码所示
我们重点分析上面一个方法，schedulerDirect是Scheduler的一个方法。而Scheduler是一个抽象类，createWorker()方法是个抽象方法，其具体的实现在IoScheduler中。而worker.schedule方法的具体实现其实是调用NewThreadWorker的schedulerDirect方法，其内部调用ScheduleExecutorService将任务运行到线程池中。
以上代码是ioScheduler的createWorker的具体实现。
以上是NewThreadWorker.scheduleDirect的具体执行过程。成功把一个任务放入了线程池中执行（子线程中执行），执行后会运行SubscribeTask的run方法。在run方法中执行了source.subscribe方法。到此：ObservableOnSubscribe的subscribe方法在io线程中被执行了
2.主线程如何切换到主线程
回过头看ObservableObserveOn的subscribeActual方法
这里的scheduler指的是HandlerScheduler。HandlerScheduler内部维护了一个运行在主线程的Handler和一个内部类HandlerWorker。其调用source.subscribe执行观察者和被观察者的订阅。当ObservableEmitter.onNext方法执行后，会调用ObserveOnObserver内部的onNext方法。
schedule方法又会调用worker.scheduler方法
此处的worker为HandlerScheduler中的Worker，源码如下
通过Handler把ScheduleRunnable发送到主线程中执行。因为HandlerScheduler是主线程handler所以在Handler中执行的逻辑也会被切换到主线程中去执行。其实这里的run方法最终运行的是ObserveOnObserver中的run方法。在其run方法中会调用其上级包装类SubscribeOnObserver的onNext方法。之后又会调用LambdaObserver的onNext方法。在其onNext方法中会调用Consumer.accept方法，最终让其运行在主线程中。
3.子线程如何切换到子线程
这里分析下把Consumer.accept方法运行在子线程的流程
同样只需要设置observeOn(Schedulers.io())就OK了。同样会创建一个ObserveOnObserver，其接受两个重要的参数this：当前Observer，scheduler：ioScheduler。
其绑定过程会执行ObservableObserveOn的subscribeActual方法
只是此处的scheduler不再是HandlerScheduler，而是IoScheduler。当ObservableEmitter.onNext方法被执行的时候，会调用ObserveOnObserver的onNext方法。而在onNext方法中又会调用IoScheduler中worker.schedule。最终会执行NewThreadWorker的scheduleActual方法
当上述方法被执行后就会调用ObserveOnObserver中的run方法。其run方法又会逐个解包装调用其OnNext方法。知道LambdaObserver的onNext被调用。onNext又会调用Consumer.accept。经过以上步骤就完成了最终的调用。因为run是在线程池中执行的，所以跟着把业务逻辑代码也切换到了线程池中执行，即子线程中执行。
总结：
经过上面的分析，RxJava切换线程已经分析完了，相信大家了解后对RxJava的线程切换会有一定的感悟。在这里再用白花总结一下。
1.子线程切换主线程：给主线程所在的Handler发消息，然后就把逻辑切换过去了。
2.主线程切换子线程：把任务放到线程池中执行就能把执行逻辑切换到子线程
3.子线程切换子线程：把任务分别扔进两个线程就行了。
标签：做了 作用 map操作 任务 nbsp code 设置 run 原理
原文地址：https://www.cnblogs.com/tony-yang-flutter/p/12331753.html
