Rxjava 过程分析四之 observeOn
说明
- 本文只分析 observeOn 主要思想和流程， 如果想看 subscribeOn 请看上一篇文章
基本使用
Flowable.create(new FlowableOnSubscribe<String>() {
@Override
public void subscribe(FlowableEmitter<String> emitter) throws Exception {
// emitter.onNext("");
// emitter.onError();
// emitter.onComplete();
}
}, BackpressureStrategy.LATEST)
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new FlowableSubscriber<String>() {
@Override
public void onSubscribe(Subscription s) {
}
@Override
public void onNext(String s) {
}
@Override
public void onError(Throwable t) {
}
@Override
public void onComplete() {
}
});
我们经常使用 subscribeOn 去切换到主线程去更新 ui 之类的。 为什么一行代码就可以把运行的线程给切换了呢？ 内部咋实现的呢？ handler？ 我们一点点看源码去揭开。
引发的思考
- 这货是怎么把线程给切换过来的呢？ 怎么实现的呢？
- 这货可以使用多次吗？
源码分析
还是老样子， 前一堆和后一堆， 我们不再分析， 如果不明白或者需要回顾的可以看以前的几篇文章。 那么我们从 observeOn 开始吧。
public final Flowable<T> observeOn(Scheduler scheduler, boolean delayError) {
return observeOn(scheduler, delayError, bufferSize());
}
然后调用到的方法直接点
public final Flowable<T> observeOn(Scheduler scheduler, boolean delayError, int bufferSize) {
return new FlowableObserveOn<T>(this, scheduler, delayError, bufferSize);
}
你看看， Rxjava 的一模一样， 都是只是新建了一个 Flowable, 然后把当前的 Flowable 作为参数传入， 进行赋值。 通过我们前三篇的分析， 别浪费时间了， 直接看 FlowableObserveOn 中的 subscribeActual 吧。
public void subscribeActual(Subscriber<? super T> s) {
Worker worker = scheduler.createWorker();
source.subscribe(new ObserveOnSubscriber<T>(s, worker, delayError, prefetch));
}
看吧， 猜就猜到了， 肯定是连锁反应的订阅了。 应为我们还不知道当前的 worker 内部是怎么实现的呢， 所以我们还是一行一行的看吧。
第一行
我们传入的 AndroidSchedulers.mainThread()， 所以我们看看这个 Schedulers 是怎么实现的。
public static Scheduler mainThread() {
return MAIN_THREAD;
}
MAIN_THREAD 是什么呀， 抬头一看， 是成员变量。 我们整体先看看这个类吧。
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;
}
});
/** 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) {
if (looper == null) throw new NullPointerException("looper == null");
return new HandlerScheduler(new Handler(looper));
}
private AndroidSchedulers() {
throw new AssertionError("No instances.");
}
}
从源码中我们发现， 其实我们可以使用这个切换到任何线程呢， 不一定要走主线程，当然使用mainThead() 默认使用了 MainLooper 而已， 因为可以传入 Looper 去在任何线程中运行。 以我们的开发使用情况来看， 现在我们重点是看看 HandlerScheduler 是怎么操作的呢。
final class HandlerScheduler extends Scheduler {
private final Handler handler;
HandlerScheduler(Handler handler) {
this.handler = handler;
}
@Override
public Worker createWorker() {
return new HandlerWorker(handler);
}
}
看到这里你是否明白了， 为什么可以切换主线程了吗？ 不管是怎么实现的。 其实应该在上一步就应该知道了， 因为使用的是 Handler(MainLooper)。 走到这里了我还想多说一句， 还记得我们介绍 subscribeOn 的这里时， 我们也说了好多， 就是操作方法一样， 只是实现不一样而已。 我记得没错的话， 在 subscribeOn 里应该时一个线程池， 而在这里换成了一个 Handler 而已。
第二行
其实和原来的一模一样， 在这里把上层的 Flowable 进行了连锁形式的订阅。 然后是上层的 Flowable 上游发射流出数据， 会直接到 ObserveOnSubscriber 中相应的方法， 那么好， 让我们直接看 ObserveOnSubscriber 中 onNext 时怎么操作才能到主线程的把， 反正跟 Handler 没跑了。
public final void onNext(T t) {
if (done) {
return;
}
if (sourceMode == ASYNC) {
trySchedule();
return;
}
if (!queue.offer(t)) {
upstream.cancel();
error = new MissingBackpressureException("Queue is full?!");
done = true;
}
trySchedule();
}
我们主要分析主流程， 我们看到不管哪种方式都会调用到 trySchedule(), 我们就进去看下
final void trySchedule() {
worker.schedule(this);
}
看到了吗。 把当前 Runable 放到 worker 里了， 我们看下 HandlerScheduler 是怎么操作的。
public Disposable schedule(Runnable run, long delay, TimeUnit unit) {
run = 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)));
return scheduled;
}
核心方法如下， 我们看到又 new 出来一个中间的 ScheduledRunnable， 把我们要执行的传了进去， 然后使用 Handler 不管是否延迟把， 会调用到 ScheduledRunnable 的 run 方法， 让我们看看里面怎么实现的。
public void run() {
try {
delegate.run();
} catch (Throwable t) {
}
}
这也太简单了把， 直接调用到了我们传进来的那个 Runnable。 那么好饶了这么一大圈就是为了， 我们的 Run 方法可以运行在指定的线程中。 好了， 我们返回到最开始去看看我们的 ObserveOnSubscriber 中 run 方法执行了什么。
public final void run() {
if (outputFused) {
runBackfused();
} else if (sourceMode == SYNC) {
runSync();
} else {
runAsync();
}
}
呀呵， 根据不同条件做不同的操作呢， 反正都是在我们指定的线程中运行呢。
我们随便看一个把 runSync 把
void runSync() {
final Subscriber<? super T> a = downstream;
final SimpleQueue<T> q = queue;
v = q.poll();
a.onNext(v);
}
把其他我们分析主流程不太关注的流程去掉，看到这几行代码是不是觉得特简单呢。
前面的疑惑问题
- 这货是怎么把线程给切换过来的呢？ 怎么实现的呢？
很明显， 是通过 handler 去处理的， 当然具体在哪个线程中， 就要看 looper 了。
- 这货可以使用多次吗？
为啥不可以， 没使用一次， 就相当于， 通过 Handler 把运行结果切换到了自己指定的线程中。
