RxJava 源码分析系列(五) -线程调度原理
对于RxJava来说，简洁的线程切换操作是它优秀的地方之一。所以了解它的线程调度原理是完全有必要，这个既能帮助我们理解其中的奥妙，同时如果自己在开发当中需要做类似的需求，可以作为一个参考。同时，RxJava的线程调度原理也是RxJava当中比较难以理解的一部分。
1.概述
说线程调度原理比较难以理解，其实我觉得就是涉及到的类比较多，然后他们之间的关系又难以梳理清楚，所以显得这部分的代码高深莫测。在这里，我先对线程调度涉及到的比较重要的类做一个小小的说明。
| 类名 | 含义 |
|---|---|
| Schedulers | 此类里面提供了我们需要使用的调度器对象 |
| Scheduler | 线程调度器，Observer或者Observable通过此类进行线程调度，每一个线程对应一个调度器 |
| Worker | 真正执行任务的类，每个调度器都需要获取一个Worker对象来执行任务 |
| Runnable | 在整个调度的过程中，一个任务会包装成一个Runnable来执行，包括线程池和Handler |
整个线程调度的核心在Scheduler和Worker类中。接下来，我们重点分析这两个类，看这两个类怎么担任线程调度的重任的。
2. Scheduler的原理分析
Scheduler是一个抽象类，我们来看看它的源码：
public abstract class Scheduler {
@NonNull
public Disposable scheduleDirect(@NonNull Runnable run, long delay, @NonNull TimeUnit unit) {
final Worker w = createWorker();
final Runnable decoratedRun = RxJavaPlugins.onSchedule(run);
DisposeTask task = new DisposeTask(decoratedRun, w);
w.schedule(task, delay, unit);
return task;
}
public abstract Worker createWorker();
}
createWorker方法是一个非常重要的方法，后续的Scheduler都必须在这个方法里面返回一个Worker。所以每一个Scheduler都对应着一个Worker。
其次scheduleDirect方法也是非常重要，当一个观察者被subscribe时，会调用到此方法来。当然这时候可能还不清楚是怎么调用到这里来的，我们也不用太害怕，这里我们只需要知道会调用这个方法来的就行，后续我会详细的追踪源码来解释。
如果此时的Scheduler是一个子线程的Scheduler,例如接下来要讲的NewThreadScheduler,那么此时提交到Worker去执行，肯定在NewThreadScheduler所在线程执行。同理，如果是Android主线程的Scheduler,肯定就在主线程里面执行了。
我们发现在这个方法里面，先是调用了createWorker创建一个Worker对象,然后后续经过一系列的包装，将任务包装成为一个Runnable，然后调用Worker的schedule方法来执行任务。
接下来，我们看几个Scheduler的子类。
(1).NewThreadScheduler
public final class NewThreadScheduler extends Scheduler {
final ThreadFactory threadFactory;
public NewThreadScheduler() {
this(THREAD_FACTORY);
}
public NewThreadScheduler(ThreadFactory threadFactory) {
this.threadFactory = threadFactory;
}
@NonNull
@Override
public Worker createWorker() {
return new NewThreadWorker(threadFactory);
}
}
我简略了一下NewThreadScheduler的代码，让它看起来更加的清晰。整个过程，我们发现在createWorker方法里面返回了一个 NewThreadWorker对象，其他感觉也没什么东西，是不是感觉咱们分析错了。当然不是，核心在Worker里面，当然此时不是分析Worker的时机。我们只需要记住整个NewThreadScheduler工作流程。
(2).HandlerScheduler
我们再来看看HandlerScheduler的源码：
final class HandlerScheduler extends Scheduler {
private final Handler handler;
private final boolean async;
HandlerScheduler(Handler handler, boolean async) {
this.handler = handler;
this.async = async;
}
@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;
}
@Override
public Worker createWorker() {
return new HandlerWorker(handler, async);
}
}
在Android当中，如果想要在主线程中工作，Handler是必不可少的。所以，我们在HandlerScheduler中看到Handler也是理所当然的。
不过，这里我们发现，HandlerScheduler的scheduleDirect方法并没有去调用createWorker方法来获取一个Worker对象,而是直接将交给了handler来执行。这是为什么？
在这里，我们得区分scheduleDirect方法的调用时机，scheduleDirect不是每个Scheduler里面都会被调用，而是这有上游的Scheduler才会被调用，也就是subscribeOn里面的Scheduler，在observeOn里面的Scheduler走的是另一个流程。
所以，当HandlerScheduler的scheduleDirect方法被调用时，此时肯定是上游在工作，这样直接提交给Handler来执行了就是了，不必要去绕圈子。
3. Worker的工作原理
现在我们来看看Worker的相关源码:
public abstract static class Worker implements Disposable {
@NonNull
public Disposable schedule(@NonNull Runnable run) {
return schedule(run, 0L, TimeUnit.NANOSECONDS);
}
@NonNull
public abstract Disposable schedule(@NonNull Runnable run, long delay, @NonNull TimeUnit unit);
}
我将很多繁琐的代码省去了，Worker的代码看上就是这样的，其中，我们发现就有一个 schedule方法。这个方法之前在Scheduler的scheduleDirect方法里面我们看到过，在scheduleDirect方法里面传递一个Runnable来执行。
我们来看一个Worker的子类，看看是怎么执行的。
(1).NewThreadWorker
先来看看NewThreadWorker的源码：
public class NewThreadWorker extends Scheduler.Worker implements Disposable {
private final ScheduledExecutorService executor;
volatile boolean disposed;
public NewThreadWorker(ThreadFactory threadFactory) {
executor = SchedulerPoolFactory.create(threadFactory);
}
@NonNull
@Override
public Disposable schedule(@NonNull final Runnable run) {
return schedule(run, 0, null);
}
@NonNull
@Override
public Disposable schedule(@NonNull final Runnable action, long delayTime, @NonNull TimeUnit unit) {
if (disposed) {
return EmptyDisposable.INSTANCE;
}
return scheduleActual(action, delayTime, unit, null);
}
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;
}
}
好像代码有点长😅，但是不急，我们一一的来分析。
首先在构造方法里面，创建了一个线程池的对象，关于线程池相关的知识，这个属于Java基础知识，这里就不进行讲解，我默认大家都懂哈😂。
然后就是一系列的schedule调用,最终会调用到scheduleActual方法里面去。scheduleActual方法里面看是做了很多的操作，其实根本目的就是调用ScheduledExecutorService的submit方法提交一个任务而已。
对的，NewThreadWorker就是这么简单，是不是之前将它想的太难了？哈哈哈😂
整个Scheduler和Worker的调用差不多就是这样的，但是是不是觉得，现在还是一脸懵逼，看到这里，我们还是不知道RxJava究竟是怎么进行线程调度的。
不急，接下来，我们将一个例子入手，由浅入深的带领大家熟悉线程调度的奥妙。我们熟悉了Scheduler和Worker，对其他的才会很快了解。
4. subscribeOn的工作流程
先来举一个栗子：
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
}
}).subscribeOn(Schedulers.newThread()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Integer integer) {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
是不是觉得这个栗子非常的熟悉，像极了前世的恋人🤓。熟悉就好，就怕大家淡忘了自己心中那个人。
这个使用方法，其他的大佬不知道讲解了多少，在这里我就不重复了，直接开门见山的分析subscribeOn方法。
subscribeOn方法主要是跟ObservableSubscribeOn类有关。我们来看看ObservableSubscribeOn类：
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)));
}
}
首先，ObservableSubscribeOn的构造方法里面多了一个Scheduler参数，这个参数到底是谁的对象得取决于在subscribeOn时。上面的栗子，我们传递的是schedulers.newThread()，所以这里的scheduler就是NewThreadScheduler的对象。
在subscribeActual方法里面，我们发现：
parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
先来看看SubscribeTask是什么东西：
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从根本来说就是一个Runnable，当这个Runnable被执行的时候，也就是run方法被调用时，我们发现此时调用了subscribe方法，此时任务就正式启动。假如，我们在向后台获取数据，此时就已经开始获取了。
然后，我们发现调用了Scheduler的scheduleDirect,还记得这个方法在做什么吗？这个方法我们在前面讲解Scheduler时已经讲解，在这个方法里面显示调用onCreateWorker方法来获取一个Worker对象,然后将我们的任务提交给Worker来执行。
这就是整个subscribeOn的流程，是不是很简单？
5. observeOn的工作流程
observeOn方法主要是表示下游的方法执行所在的线程，跟subscribeOn方法差不多，observeOn主要跟ObservableObserveOn,我们来看看ObservableObserveOn的相关代码：
public final class ObservableObserveOn<T> extends AbstractObservableWithUpstream<T, T> {
final Scheduler scheduler;
final boolean delayError;
final int bufferSize;
public ObservableObserveOn(ObservableSource<T> source, Scheduler scheduler, boolean delayError, int bufferSize) {
super(source);
this.scheduler = scheduler;
this.delayError = delayError;
this.bufferSize = bufferSize;
}
@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));
}
}
}
还是先来看看构造方法，同样的，scheduler也是一个调度器，这里由于Android的主线程，所以是HandlerScheduler；delayError表示是否延迟抛出异常；bufferSize表示缓冲区的大小，如果上游的发射速度比较快的话，这里有可能出现背压问题。
然后看看subscribeActual方法里面，我们发现，在这里，调用了onCreateWorker方法来获取一个Worker对象，然后subscribe了一个ObserveOnObserver对象，传进去了一些相关参数。看来真正的核心在ObserveOnObserver这里面。
不过在这里，我们发现ObservableObserveOn跟ObservableSubscribeOn的subscribeActual方法还是有很大的区别，在ObservableObserveOn的subscribeActual方法没有调用Scheduler的相关方法。之前，我们说了，scheduleDirect方法只在上游执行，从这里就可以看出来。
我们来看看ObserveOnObserver相关代码，ObserveOnObserver比较重要的是onNext方法，看看在方法里面是怎么将数据分发到主线程去执行的。
@Override
public void onNext(T t) {
if (done) {
return;
}
if (sourceMode != QueueDisposable.ASYNC) {
queue.offer(t);
}
schedule();
}
onNext方法还是比较简单，先将数据放在队列里面去，然后在调用schedule方法。我们再来看看schedule方法：
void schedule() {
if (getAndIncrement() == 0) {
worker.schedule(this);
}
}
这里getAndIncrement() == 0的意思比较好理解，表示当前队列中只有一个数据，就调用Worker的schedule方法。到这里，我们根本没有看到调用Observer的onNext方法，是不是分析错了？肯定不是的，我们来看看Worker的schedule方法。这里的Worker是HandlerWorker，所以我们看一下HandlerWorker的schedule方法:
public Disposable schedule(Runnable run, long delay, TimeUnit unit) {
if (run == null) throw new NullPointerException("run == null");
if (unit == null) throw new NullPointerException("unit == null");
if (disposed) {
return Disposables.disposed();
}
run = RxJavaPlugins.onSchedule(run);
ScheduledRunnable scheduled = new ScheduledRunnable(handler, run);
Message message = Message.obtain(handler, scheduled);
message.obj = this; // Used as token for batch disposal of this worker's runnables.
if (async) {
message.setAsynchronous(true);
}
handler.sendMessageDelayed(message, unit.toMillis(delay));
// Re-check disposed state for removing in case we were racing a call to dispose().
if (disposed) {
handler.removeCallbacks(scheduled);
return Disposables.disposed();
}
return scheduled;
}
看到写了这么多，其实就一句话，将传递进来的Runnable交给Handler来执行。那个这个Runnable是谁？就是ObservableObserveOn自己，Handler会执行Runnable的run方法。
哈哈，现在得回来ObservableObserveOn的run方法：
@Override
public void run() {
if (outputFused) {
drainFused();
} else {
drainNormal();
}
}
这里的两个drain方法着实将我吓着了。其实，认真想一想，跟BufferAsyncEmitter里面的drain方法差不多，这里就不深入的讲解，反正这里就是将数据发送到下游的真正操作，而且是在observeOn方法里面传递的那个线程执行。
6.总结
RxJava的线程调度这部分的知识着实是不好理解，所以尽管我在这里说的天花乱坠，还是得需要我们亲自来看看这部分的代码。在这里，我对这部分的知识做一个简单的总结。
1.每种类型线程对一个Scheduler，而每个Scheduler对应一个Worker,真正操作都是Worker来完成的。
2.子线程执行的根本就是将我们的subscribe方法调用放在一个Runnable里面去执行，类似于代理机制，又像装饰者模式。
