RxJava2线程切换原理分析
Posted 飘杨......
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了RxJava2线程切换原理分析相关的知识,希望对你有一定的参考价值。
一、概述
本节将分析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源码分析:
@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)); }
从上述源码可以看出subscribeOn确实如上面所说,会被包装成为一个ObservableSubscribeOn。其构造方法会传入两个参数,一个是this:代表当前被观察者,也就是操作符上面修饰的那个被观察者,本例中指的是ObservableObserveOn,ObservableObserverOn又装饰了ObservableCreate。scheduler指的是Schedulers.io(), 指被观察者运行在io线程,也就是子线程中。
下面看下Schedulers类是个什么东西。
public final class Schedulers { @NonNull static final Scheduler SINGLE; @NonNull static final Scheduler COMPUTATION; @NonNull static final Scheduler IO; @NonNull static final Scheduler TRAMPOLINE; @NonNull static final Scheduler NEW_THREAD; static final class SingleHolder { static final Scheduler DEFAULT = new SingleScheduler(); } static final class ComputationHolder { static final Scheduler DEFAULT = new ComputationScheduler(); } static final class IoHolder { static final Scheduler DEFAULT = new ioscheduler(); } static final class NewThreadHolder { static final Scheduler DEFAULT = new NewThreadScheduler(); } static { SINGLE = RxJavaPlugins.initSingleScheduler(new SingleTask()); COMPUTATION = RxJavaPlugins.initComputationScheduler(new ComputationTask()); IO = RxJavaPlugins.initIoScheduler(new IOTask()); TRAMPOLINE = TrampolineScheduler.instance(); NEW_THREAD = RxJavaPlugins.initNewThreadScheduler(new NewThreadTask()); }
Schedulers内部封装了各种Scheduler。每一个Scheduler中都封装的有线程池,用于执行后台任务。
到此处ObservableSubscribeOn对象也就创建完成了。
下面看下ObserverOn操作符都干了什么事情:
@CheckReturnValue @SchedulerSupport(SchedulerSupport.CUSTOM) public final Observable<T> observeOn(Scheduler scheduler, boolean delayError, int bufferSize) { ObjectHelper.requireNonNull(scheduler, "scheduler is null"); ObjectHelper.verifyPositive(bufferSize, "bufferSize"); return RxJavaPlugins.onAssembly(new ObservableObserveOn<T>(this, scheduler, delayError, bufferSize)); }
ObserveOn方法内部包装了一个ObservableObserveOn对象,其有两个参数,this:代表当前Observable对象,此处指的是ObservableCreate这个对象,scheduler代表的是AndroidSchedulers.mainThread()。
我们看一下AndroidSchedulers的源代码,看它都干了写什么事
public final class AndroidSchedulers { private static final class MainHolder { static final Scheduler DEFAULT = new HandlerScheduler(new Handler(Looper.getMainLooper()), false); } private static final Scheduler MAIN_THREAD = RxAndroidPlugins.initMainThreadScheduler( new Callable<Scheduler>() { @Override public Scheduler call() throws Exception { return MainHolder.DEFAULT; } }); /** A {@link Scheduler} which executes actions on the Android main thread. */ public static Scheduler mainThread() { return RxAndroidPlugins.onMainThreadScheduler(MAIN_THREAD); } /** A {@link Scheduler} which executes actions on {@code looper}. */ public static Scheduler from(Looper looper) { return from(looper, false); }
AndroidSchedulers的内部类MainHolder的作用是在主线程中创建一个Handler。由new Handler(Looper.getMainLooper())来完成。因为Looper所在的线程为Handler所在的线程,又因为Looper.getMainLooper()获取到的是主线程的looper,所以当前Handler运行在主线程,顺带着这块的逻辑也是在主线程中完成的。字段MAIN_THREAD仅仅是把HandlerScheduler返回而已,而HandlerScheduler持有主线程handler。那么manThread()方法就好理解了 ,就是返回了一个持有主线程Handler的Scheduler而已。
所以ObservableObserverOn包装了ObservableCreate并持有了主线程Handler。到此被观察者就已经创建完成了。
下面说下观察者Consumer.accept方法在这个链式调用中是如何被执行的:
1.经过上面的分析被观察者已经变为:ObservableObserverOn,ObservableObserverOn持有ObservableSubscribeOn对象的引用,ObservableSubscribeOn又持有ObservableCreate的引用。所以Observable对象经过三层包装最终成为了ObservableObserverOn。
2.Observable.subscribe(Consumer consumer)方法执行订阅,会把原始的观察者对象LambdaObserver对象包装成为ObserverOnObserver对象,ObserverOnObserver又会被包装成SubscribeOnObserver对象。用以在ObservableSubscribeOn对象执行subscribeActual方法的时候正式执行绑定操作。至此,观察者和被观察者建立了绑定关系。
public final class ObservableSubscribeOn<T> extends AbstractObservableWithUpstream<T, T> { final Scheduler scheduler; 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 SubscribeTask(parent))); }
从上面的代码中我们基本无法判断是在哪里绑定的。从上面的分析我们知道scheduler要是一个HandlerScheduler.那么我们可以断定的是scheduler.scheduleDirect一定是用来执行任务的,那么SubscribeTask肯定是一个任务没错。事实也如我们所料一样,确实是这样的。
看下HandlerScheduler的scheduleDirect都干了什么
@Override public Disposable scheduleDirect(Runnable run, long delay, TimeUnit unit) { if (run == null) throw new NullPointerException("run == null"); if (unit == null) throw new NullPointerException("unit == null"); run = RxJavaPlugins.onSchedule(run); ScheduledRunnable scheduled = new ScheduledRunnable(handler, run); handler.postDelayed(scheduled, unit.toMillis(delay)); return scheduled; }
非常的简单,构建一个ScheduleRunnable,并把handler和runnable传入进去,然后执行handler.postDeayed向handler发送消息就行了。postDeayed方法最终会调用Runnable的run方法。
private static final class ScheduledRunnable implements Runnable, Disposable { private final Handler handler; private final Runnable delegate; private volatile boolean disposed; // Tracked solely for isDisposed(). ScheduledRunnable(Handler handler, Runnable delegate) { this.handler = handler; this.delegate = delegate; } @Override public void run() { try { delegate.run(); } catch (Throwable t) { RxJavaPlugins.onError(t); } }
ScheduleRunnable在run方法中又会调用SubscribeTask的run方法。
SubscribeTask.java
final class SubscribeTask implements Runnable { private final SubscribeOnObserver<T> parent; SubscribeTask(SubscribeOnObserver<T> parent) { this.parent = parent; } @Override public void run() { source.subscribe(parent); } }
在subscribeTask的run方法中最终完成了绑定,source指ObservableOnSubscribe
3.被观察者在执行ObservableOnSubscribe实例的subscribe方法的ObservableEmitter参数的onNext方法的时候,会首先调用SubscribeOnObserver的onNext方法,又由于SubscribeOnObserver持有ObserverOnObserver的引用,因此在SubscribeOnObserver的onNext方法中又会调用ObserveOnObserver对象的onNext方法,在此Next方法中又会调用CreateObserver的onNext方法,在其内部又会调用LambdaObserver.onNext,然后在LambdaObserver的onNext方法中又会调用Consumer.accept方法。最后完成数据的从发送到接收的流转。
了解了以上操作符的整体流转流程后,我们接下来回过头来看开头我们提出的三个问题:
1.主线程切换到子线程
我们先来看ObservableSubscribeOn这个类,在上面的小例子中,直接将被观察者运行在IO线程中了。我们直接看ObservableSubscribeOn的subscribeActual方法的源代码
@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 SubscribeTask(parent))); }
在subscribeActual方法内部先创建一个SubscribeOnObserver对象,并执行setDisposable执行任务。这里的scheduler指的是HandlerScheduler。SubscribeTask是一个实现了Runnable的对象在其内部完成了绑定操作。
先来看下HandlerScheduler的scheduleDirect方法
@Override public Disposable scheduleDirect(Runnable run, long delay, TimeUnit unit) { if (run == null) throw new NullPointerException("run == null"); if (unit == null) throw new NullPointerException("unit == null"); run = RxJavaPlugins.onSchedule(run); ScheduledRunnable scheduled = new ScheduledRunnable(handler, run); handler.postDelayed(scheduled, unit.toMillis(delay)); return scheduled; }
scheduleDirect方法逻辑上很简单,1.把subscribeTask和handler封装成ScheduleRunnable。然后利用Handler.postDelayed执行这个Runnable对象。postDelayed执行的最后会调用msg.callback.run()其实就是调用ScheduleRunnable的run方法。在在run方法内又会调用SubscribeTask的run方法。
private static final class ScheduledRunnable implements Runnable, Disposable { private final Handler handler; private final Runnable delegate; private volatile boolean disposed; // Tracked solely for isDisposed(). ScheduledRunnable(Handler handler, Runnable delegate) { this.handler = handler; this.delegate = delegate; } @Override public void run() { try {
//SubscribeTask的run方法 delegate.run(); } catch (Throwable t) { RxJavaPlugins.onError(t); } }
我们看下SubscribeTask的run方法都干了啥事。
final class SubscribeTask implements Runnable { private final SubscribeOnObserver<T> parent; SubscribeTask(SubscribeOnObserver<T> parent) { this.parent = parent; } @Override public void run() { source.subscribe(parent); } }
在SubscribeTask类的run方法中完成最终的绑定。此处的source指的是ObservableOnSubscribe
在主线程中执行其实也就这么多最终会把方法放到Handler中执行
2.在子线程中执行任务
直接看ObservableObserveOn类的subscribeActual
@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)); } }
首先根据TrampolineScheduler判断任务是否是在当前线程执行,如果是就直接绑定。如果不是就创建一个ObserverOnObserver对象,并把Observer和Worker对象传递进去。即可完成绑定。
我们接下来主要看下其是如何在子线程中执行的
ObserverOnObserver继承了Runnable对象。在执行onNext方法的时候会调用worker的schedule(this)方法。
@Override public void onNext(T t) { if (done) { return; } if (sourceMode != QueueDisposable.ASYNC) { queue.offer(t); } schedule(); }
void schedule() { if (getAndIncrement() == 0) { worker.schedule(this); } }
其实到这我们大致可以判断出来worker.schedule(this)必定会运行run方法。不着急,我们先看IoSchedule类中的worker以及worker.schedule干了什么
@NonNull @Override public Worker createWorker() { return new EventLoopWorker(pool.get()); }
创建一个Worker对象
@NonNull @Override public Disposable schedule(@NonNull Runnable action, long delayTime, @NonNull TimeUnit unit) { if (tasks.isDisposed()) { // don\'t schedule, we are unsubscribed return EmptyDisposable.INSTANCE; } return threadWorker.scheduleActual(action, delayTime, unit, tasks); }
执行threadWorker.scheduleActual(action),这里的action指的就是ObserverObserveOn对象,因其继承了Runnable对象。
看看ThreadWorker.scheduleActual干了啥
@NonNull public ScheduledRunnable scheduleActual(final Runnable run, long delayTime, @NonNull TimeUnit unit, @Nullable DisposableContainer parent) { Runnable decoratedRun = RxJavaPlugins.onSchedule(run); ScheduledRunnable sr = new ScheduledRunnable(decoratedRun, parent); if (parent != null) { if (!parent.add(sr)) { return sr; } } Future<?> f; try { if (delayTime <= 0) { f = executor.submit((Callable<Object>)sr); } else { f = executor.schedule((Callable<Object>)sr, delayTime, unit); } sr.setFuture(f); } catch (RejectedExecutionException ex) { if (parent != null) { parent.remove(sr); } RxJavaPlugins.onError(ex); } return sr; }
逻辑很清晰,把传入的Runnable(ObservableObserveOn)封装成为一个ScheduleRunnable对象。并把这个对象放入线程池中去执行。
executor都代表线程池。执行的时候会运行ScheduleRunnable的run方法。在其run方法内部又会调用ObserverObserveOn的run方法。
下面回过头来再看看ObserverObserveOn的run方法
@Override public void run() { if (outputFused) { drainFused(); } else { drainNormal(); } }
void drainNormal() { .... a.onNext(v); .... } }
其会调用a.onNext方法,让onNext方法运行在线程池中。a值的就是一个CreateObserver或者其包装类。通过一层层的调用Consume.accept方法最终会运行到子线程中。
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.子线程切换子线程:把任务分别扔进两个线程就行了。
以上是关于RxJava2线程切换原理分析的主要内容,如果未能解决你的问题,请参考以下文章