RxJava 是如何实现线程切换的（下）
通过前一篇的 RxJava2 是如何实现线程切换的 (上) 我们已经知道了在RxJava中，subscribeOn 将上游线程切换到 指定的子线程 是如何实现的。这里就接着来看，observeOn 是如何将下游线程切换到指定线程的。
RxJava - subscribeOn
这里可以通过UML图简单回顾一下subscribeOn的原理。
通过 subscribeOn 我们完成了以下操作：
- 创建了一个 ObservableSubscribeOn 对象，本质上来说他就是一个Observable，他同时实现了 AbstractObservableWithUpstream（HasUpstreamObservableSource ）这样一个接口，是他变了一个 拥有上游 的Observeable。
- 在 ObservableSubscribeOn 的 subscribeActual 方法中
parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
将真正的 subscribe 操作安置在了SubscribeTask这样个一个Runnable当中，这个 Runnable 将由scheduler 这个调度器负责启动，因此就把上游操作放到了 scheduler 所在的线程中。
- Schedulers.newThread()或者Schedulers.io() 都是通过工厂方法的模式创建了某种 指定类型 的线程， 当这个特定的线程执行是，就是执行真实的 subscribe 方法，这样就把上游操作放到了一个特定的线程中去执行。
RxJava - observeOn
简单回顾完 subscribeOn 之后，我们就来看看 observeOn 是如何工作的。
其实，了解 subscribeOn 的原理之后，再来看 observeOn 就简单多了，类的命名及实现思路都有很多相似之处，可以对照着理解。
RxJava的代码写的非常巧妙，可以说是百读不厌，可以学习的地方特别多。为了避免陷入只见树木不见森林的噩梦，我们就带着以下问题去探索 observeOn 的奥秘。
- 在 Android 中线程间传递消息会使用 Handler，这里是否使用？又是如何使用的？
- AndroidSchedulers.mainThread() 做了什么 ？
- 下游任务是如何保证被分配到指定线程的。
示例
private void multiThread() {
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> e) throws Exception {
e.onNext("This msg from work thread :" + Thread.currentThread().getName());
sb.append("/nsubscribe: currentThreadName==" + Thread.currentThread().getName());
}
})
.subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
Log.e(TAG, "accept: s= " + s);
}
});
}
我们还是以这段代码为例，来看看 observeOn 的工作原理。这里通过observeOn(AndroidSchedulers.mainThread())将下游线程切换到了我们非常熟悉的 Android UI 线程。这样就可以确保我们在下游所有的操作都是在 UI 线程中完成。这里和讨论 subscribeOn 一样，我们就从这句代码出发，看看这背后到底发生了什么。
有了上一篇的经验，我们知道 AndroidSchedulers.mainThread() 一定去创建了某种类型的调度器，为了方便后面的叙述，这一次我们先从调度器的创建说起，后面再看 observeOn() 的具体实现。
需要注意的是 AndroidSchedulers 并不是 RxJava 的一部分，是为了在 Android 中方便的使用 RxJava 而专门设计的一个调度器实现，源码 RxAndroid 设计非常巧妙；使用前记得在gradle文件中配置依赖。
AndroidSchedulers.mainThread()
下面就来看看 AndroidSchedulers.mainThread() 这个我们非常熟悉的 Scheduler 是如何创建的。
public final class AndroidSchedulers {
private static final class MainHolder {
static final Scheduler DEFAULT = new HandlerScheduler(new Handler(Looper.getMainLooper()));
}
private static final Scheduler MAIN_THREAD = RxAndroidPlugins.initMainThreadScheduler(
new Callable<Scheduler>() {
@Override public Scheduler call() throws Exception {
return MainHolder.DEFAULT;
}
});
public static Scheduler mainThread() {
return RxAndroidPlugins.onMainThreadScheduler(MAIN_THREAD);
}
}
这里我们可以认为，当调用AndroidSchedulers.mainThread() 时，返回了一个HandlerScheduler 的实例，而这个实例使用到了我们非常熟悉的 Handler。那么重点就来到HandlerScheduler 了。
final class HandlerScheduler extends Scheduler {
private final Handler handler;
HandlerScheduler(Handler handler) {
this.handler = handler;
}
@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, Math.max(0L, unit.toMillis(delay)));
return scheduled;
}
@Override
public Worker createWorker() {
return new HandlerWorker(handler);
}
private static final class HandlerWorker extends Worker {
private final Handler handler;
private volatile boolean disposed;
HandlerWorker(Handler handler) {
this.handler = handler;
}
@Override
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.
handler.sendMessageDelayed(message, Math.max(0L, 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;
}
@Override
public void dispose() {
disposed = true;
handler.removeCallbacksAndMessages(this /* token */);
}
@Override
public boolean isDisposed() {
return disposed;
}
}
private static final class ScheduledRunnable implements Runnable, Disposable {
private final Handler handler;
private final Runnable delegate;
private volatile boolean disposed;
ScheduledRunnable(Handler handler, Runnable delegate) {
this.handler = handler;
this.delegate = delegate;
}
@Override
public void run() {
try {
delegate.run();
} catch (Throwable t) {
IllegalStateException ie =
new IllegalStateException("Fatal Exception thrown on Scheduler.", t);
RxJavaPlugins.onError(ie);
Thread thread = Thread.currentThread();
thread.getUncaughtExceptionHandler().uncaughtException(thread, ie);
}
}
@Override
public void dispose() {
disposed = true;
handler.removeCallbacks(this);
}
@Override
public boolean isDisposed() {
return disposed;
}
}
}
这个类虽然很简单，但是设计非常巧妙。
-
首先 HandlerScheduler 是一个 Scheduler ，通过构造函数他获取到了主线程所在的 Handler实例。而在他的 createWorker() 方法中，他又通过这个 Handler 实例创建了一个HandlerWorker 的实例，这个HandlerWorker 本质上就是一个 Worker。在他的 schedule 方法中，创建了一个 ScheduleRunnable 对象，并会把这个Runnable对象通过 handler 的 sendMessageDelayed 方法发送出去，而我们知道这个 Handler 是主线程，这样在下游中，就把任务从某个子线程转移到了UI线程。
-
ScheduleRunnable 不但实现了 Runnable ，而且实现了我们看到过无数次的 Disposable 。
@Override
public void run() {
try {
delegate.run();
} catch (Throwable t) {
}
}
@Override
public void dispose() {
disposed = true;
handler.removeCallbacks(this);
}
这样，正确情况下 run 方法会正常执行线程中的任务，而一旦 disposable 对象执行了dispose()方法，那么 handler.removeCallbacks(this),就可确保在 handler 的 dispatchMessage 方法中，不会在执行任何操作，从而达到了 dispose 的效果。
observeOn
下面就来看看 Observable 中的 observeOn 方法
Observable.java --- observeOn
public final Observable<T> observeOn(Scheduler scheduler) {
return observeOn(scheduler, false, bufferSize());
}
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));
}
这个方法的实现和 subscribeOn 的实现非常相似，多了两个参数 delayError 和 buffersize 。 buffersize 可以认为是RxJava内部的一个静态变量，默认情况下他的值是128。通过我们之前的经验，这里可以把 observeOn 的过程简化如下：
new ObservableObserveOn<T>(this, scheduler, delayError, bufferSize)
也就是说 observeOn 这个操作符给我们返回了一个 ObservableObserveOn 对象。很容易想到他也是一个 Observeable。那么我们就去看看这个 ObservableObserveOn 到底是什么？我们最关心的 subscribeActual 方法他又是怎样实现的。
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));
}
}
}
和 ObservableSubscribeOn 一样，他也继承了 AbstractObservableWithUpstream ，这样他也是一个拥有上游的 Observeable，他的构造函数很简单，没什么可以说。这里我们重点关注一下 subscribeActual 方法的实现。这里我们的使用的 Scheduler 实例是 AndroidSchedulers.mainThread() ，因此就按 else的逻辑分析。
Scheduler.Worker w = scheduler.createWorker();
source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
通过 scheduler.createWorker() 创建了 Worker 这个对象。这里结合之前对 AndroidSchedulers.mainThread() 的分析，此处的 worker 对象是就是一个持有主线程 handler 引用的 Worker。
接着用这个worker又创建了一个ObserveOnObserver对象。看看这个类的实现。
static final class ObserveOnObserver<T> extends BasicIntQueueDisposable<T>
implements Observer<T>, Runnable { ....}
这个类功能非常强大，首先是一个 Observer ，同时也是一个Runnable，并且还继承了 BasicIntQueueDisposable(保证原子性、拥有操作队列功能和 Disposable功能)。
source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
我们关注一下这行代码，根据之前的说法这里的 source 是其父类(AbstractObservableWithUpstream)中的成员变量，也就是说是上游，那么当前ObservableObserveOn 的上游是谁呢？ 就是我们上一篇所说的 ObservableSubscribeOn 。
因此，当这里开始执行订阅方法 subscribe() 后，将以如下顺序响应：
Observable.subscribe--->Observable.subscribeActual---> ObservableObserveOn.subscribeActual---> ObservableSubscribeOn.subscribeActual--->ObservableCreate.subscribeActual
这些方法的参数均为 observer,通过层层回调，最后的 subscribeActual(Observer<? super T> observer) 执行时，这个 observer 持有之前几个 observer 的引用。
我们再看一下 ObservableCreate.subscribeActual
@Override
protected void subscribeActual(Observer<? super T> observer) {
CreateEmitter<T> parent = new CreateEmitter<T>(observer);
observer.onSubscribe(parent);
try {
source.subscribe(parent);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
parent.onError(ex);
}
}
可以看到，这里首先会触发 observer.onSubscribe ，我们再看一下 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)));
}
好了，这样我们又回到了原点：
source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
回到了最初的 Observer： ObserveOnObserver
这个 ObserveOnObserver 持有我们一开始创建的observer，也就是一个Consumer对象。
下面就来看看这个 ObserveOnObserver
- 构造函数
ObserveOnObserver(Observer<? super T> actual, Scheduler.Worker worker, boolean delayError, int bufferSize) {
this.actual = actual;
this.worker = worker;
this.delayError = delayError;
this.bufferSize = bufferSize;
}
这里指的注意的一点 ，actual 其实就是observer
- onSubscribe
@Override
public void onSubscribe(Disposable s) {
if (DisposableHelper.validate(this.s, s)) {
this.s = s;
// 现阶段，我们用到的Disposable 都是单个的，暂时不讨论其
//为QueueDisposable的情况
queue = new SpscLinkedArrayQueue<T>(bufferSize);
actual.onSubscribe(this);
}
}
在 ObservableCreate.subscribeActual 中我们知道，当执行subscribe 方法后，首先会执行 observer的 onSubscribe 方法。这里的实现非常简单，就是创建了一个queue,并触发了这个 observer 自己的 onSubscribe 方法。
- onNext
@Override
public void onNext(T t) {
if (done) {
return;
}
if (sourceMode != QueueDisposable.ASYNC) {
queue.offer(t);
}
schedule();
}
在 onNext 中会执行 scheule() 方法。
void schedule() {
if (getAndIncrement() == 0) {
worker.schedule(this);
}
}
这个地方就有意思了，前面说过这里的 worker 是一个持有主线程handler 的Worker对象，当他的 schedule 执行时，就会把特定的线程任务通过Handler.postDelay 方法转移到主线中去执行 。
那么这里的this 又是什么呢？前面我们说过，ObserveOnObserver 这个类功能非常强大，他是一个Runnable，那么这里就是执行他自己的run方法喽，我们赶紧看看。
@Override
public void run() {
if (outputFused) {
drainFused();
} else {
drainNormal();
}
}
这里有一个参数 outputFused 他默认是false，至于他什么时候为true，不作为这里讨论的重点。
void drainNormal() {
int missed = 1;
final SimpleQueue<T> q = queue;
final Observer<? super T> a = actual;
for (;;) {
if (checkTerminated(done, q.isEmpty(), a)) {
return;
}
for (;;) {
boolean d = done;
T v;
try {
v = q.poll();
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
s.dispose();
q.clear();
a.onError(ex);
worker.dispose();
return;
}
boolean empty = v == null;
if (checkTerminated(d, empty, a)) {
return;
}
if (empty) {
break;
}
a.onNext(v);
}
missed = addAndGet(-missed);
if (missed == 0) {
break;
}
}
}
这里大概就是通过一个死循环，不断从 onSubscribe 方法中创建的队列中取出事件，执行observer 的 onNext方法。而当为例为空时，就会执行worker.dispose 取消整个事件流，同时从Handler中移除所有消息。
最后在看一眼 onComplete ,onError 和整个类似
@Override
public void onComplete() {
if (done) {
return;
}
done = true;
schedule();
}
可以看到这里的处理也很简单，done 设置为 true .这样最后便完成了下游事件的执行。
最后
好了，由于一些无以诉说的原因，经历了很久终于把 RxJava 线程切换的下篇给完成了。
正文到此结束
热门推荐
相关文章
Loading...
![[HBLOG]公众号](http://www.liuhaihua.cn/img/qrcode_gzh.jpg)
