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)); }
从上面的代码可以看出，observeOn方法最终调用的是含有三个参数的observeOn方法，而这个方法的作用是返回了ObservableObserveOn类的实例并将observeOn方法的参数scheduler注入其中。 现在来看实例化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)); } } //... //省略部分代码 }
可以看到，实例化ObservableObserveOn类的时候，将ObservableSubscribeOn的实例及AndroidSchedulers.mainThread()还有其他的两个默认参数都作为了它的成员变量保存。
由前面的两篇文章可知，下游的subscribe方法最终会调用上游的subscribeActual方法，所以会调用这里的subscribeActual方法，代码如下
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)); } }
现在来一句句的分析上面的代码，首先if条件肯定是不成立的，因为这里scheduler其实是HandlerScheduler,为什么是HandlerScheduler呢？我们来一点点的分析，由
observeOn(AndroidSchedulers.mainThread())
这句代码可以知道，observeOn方法的参数是AndroidSchedulers.mainThread()，那这个AndroidSchedulers.mainThread()又是什么呢？看代码
public static Scheduler mainThread() { return RxAndroidPlugins.onMainThreadScheduler(MAIN_THREAD); } //上面的方法，返回的就是MAIN_THREAD,而MAIN_THREAD最终返回的是 //MainHolder.DEFAULT private static final Scheduler MAIN_THREAD = RxAndroidPlugins.initMainThreadScheduler( new Callable<Scheduler>() { @Override public Scheduler call() throws Exception { return MainHolder.DEFAULT; } }); //上面的MainHolder.DEFAULT就是实例化了HandlerScheduler private static final class MainHolder { static final Scheduler DEFAULT = new HandlerScheduler(new Handler(Looper.getMainLooper()), false); }
上面贴出的代码都是与
AndroidSchedulers.mainThread()
有关的代码，从上面的代码中可以得出结论，AndroidSchedulers.mainThread()最终是实例化了HandlerScheduler，所以，subscribeActual方法中的scheduler是HandlerScheduler，所以，if语句的条件不成立，这里会执行subscribeActual方法中的else语句，即执行下面的代码
Scheduler.Worker w = scheduler.createWorker(); source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
同样，按照代码的执行顺序来分析，看第一句代码，
Scheduler.Worker w = scheduler.createWorker();
从前文中的分析可以知道，这里的scheduler是HandlerScheduler，所以，这里是调用HandlerScheduler类中的createWorker方法，HandlerScheduler类中的createWorker方法的代码如下
public Worker createWorker() { return new HandlerWorker(handler, async); }
从上面的代码可以得出，HandlerScheduler类中的createWorker方法返回了HandlerWorker类的实例，这里传入HandlerWorker构造方法中的两个参数是在上面已经分析过的方法中进行初始化的，如下
private static final class MainHolder { static final Scheduler DEFAULT = new HandlerScheduler(new Handler(Looper.getMainLooper()), false); }
从这句代码中可以得知，HandlerWorker构造方法中的handler是实例化在主线程中的Handler，async的值是false。 好了，到这里我们知道了Scheduler.Worker w = scheduler.createWorker();这句代码的作用是实例化了HandlerWorker，而实例化HandlerWorker的同时，在其构造方法中初始化了两个成员变量。
下面继续看这句代码
source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize))
;,先看这句代码中的这段
new ObserveOnObserver<T>(observer, w, delayError, bufferSize)
代码做了什么，代码如下
ObserveOnObserver(Observer<? super T> actual, Scheduler.Worker worker, boolean delayError, int bufferSize) { this.downstream = actual; this.worker = worker; this.delayError = delayError; this.bufferSize = bufferSize; }
根据上文的分析可以得出这里的几个参数分别代表什么
- this.downstream就是这段代码
new Observer<String>() { @Override public void onSubscribe(Disposable d) { Log.e("wizardev", "onSubscribe: "+Thread.currentThread().getName() ); } @Override public void onNext(String s) { Log.e("wizardev", "接收到上游发射的数据为: " + s); Log.e("wizardev", "下游所在的线程: "+ Thread.currentThread().getName()); } @Override public void onError(Throwable e) { } @Override public void onComplete() { } }
- this.worker就是
new HandlerWorker(handler, async); - this.delayError的值是false
- this.bufferSize就是一个int型的数字
好了，现在继续来看
source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
这里的source就是上游的Observable，这里就是ObservableSubscribeOn类的实例，所以这句代码实际就是调用了ObservableSubscribeOn类中的subscribe方法，而ObservableSubscribeOn没有这个方法，所以是调用其父类的subscribr方法，由之前的文章可知，最终调用的就是ObservableSubscribeOn类中的subscribeActual方法。所以，现在需要把思路切换到ObservableSubscribeOn类中的subscribeActual方法了
ObservableSubscribeOn类中subscribeActual方法分析
还是看代码，subscribeActual方法的如下
public void subscribeActual(final Observer<? super T> observer) { //1 final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(observer); //2 observer.onSubscribe(parent); //3 parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent))); }
根据前面的分析可知，这个方法中的参数就是
new ObserveOnObserver<T>(observer, w, delayError, bufferSize)
这段代码。照旧，按照代码的执行顺序分析，代码中已经标注了1，2，3的执行步骤，
- 现在来分析“1”处代码，看下SubscribeOnObserver类，代码如下
static final class SubscribeOnObserver<T> extends AtomicReference<Disposable> implements Observer<T>, Disposable { private static final long serialVersionUID = 8094547886072529208L; final Observer<? super T> downstream; final AtomicReference<Disposable> upstream; //这里的downstream就是new ObserveOnObserver<T>(observer, w, delayError, bufferSize) SubscribeOnObserver(Observer<? super T> downstream) { this.downstream = downstream; this.upstream = new AtomicReference<Disposable>(); } @Override public void onSubscribe(Disposable d) { DisposableHelper.setOnce(this.upstream, d); } @Override public void onNext(T t) { downstream.onNext(t); } @Override public void onError(Throwable t) { downstream.onError(t); } @Override public void onComplete() { downstream.onComplete(); } @Override public void dispose() { DisposableHelper.dispose(upstream); DisposableHelper.dispose(this); } @Override public boolean isDisposed() { return DisposableHelper.isDisposed(get()); } void setDisposable(Disposable d) { DisposableHelper.setOnce(this, d); } }
- 重要部分已在代码中注释。
- 接着分析“2”处的代码，这里的observe就是ObserveOnObserver的实例，调用的就是ObserveOnObserver类中的
onSubscribe方法,onSubscribe方法的代码如下
public void onSubscribe(Disposable d) { //这里会直接进入if方法中 if (DisposableHelper.validate(this.upstream, d)) { //这句代码的作用就是将new SubscribeOnObserver<T>(observer);赋值给了this.upstream this.upstream = d; //d的值是SubscribeOnObserver的实例，这里if条件不成立 if (d instanceof QueueDisposable) { @SuppressWarnings("unchecked") QueueDisposable<T> qd = (QueueDisposable<T>) d; int m = qd.requestFusion(QueueDisposable.ANY | QueueDisposable.BOUNDARY); if (m == QueueDisposable.SYNC) { sourceMode = m; queue = qd; done = true; downstream.onSubscribe(this); schedule(); return; } if (m == QueueDisposable.ASYNC) { sourceMode = m; queue = qd; downstream.onSubscribe(this); return; } } //实例化大小为bufferSize的队列 queue = new SpscLinkedArrayQueue<T>(bufferSize); //上文已经分析了downstream的值，最下游的onSubscribe与线程调度 //无关，在那个线程调用的subscribe就在哪个线程回调 downstream.onSubscribe(this); } }
- 主要的代码已在文中注释，下面来分析“3”处的代码
- 现在一步步的分析“3”处的代码
new SubscribeTask(parent)
- 代码如下
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，并将
new SubscribeOnObserver<T>(observer)
作为成员变量。
继续看scheduler.scheduleDirect(…)这里的scheduler是这句代码Schedulers.io(),Schedulers.io()代码如下
public static Scheduler io() { return RxJavaPlugins.onIoScheduler(IO); } static { SINGLE = RxJavaPlugins.initSingleScheduler(new SingleTask()); COMPUTATION = RxJavaPlugins.initComputationScheduler(new ComputationTask()); //IO是实例化的IOTask IO = RxJavaPlugins.initIoScheduler(new IOTask()); TRAMPOLINE = TrampolineScheduler.instance(); NEW_THREAD = RxJavaPlugins.initNewThreadScheduler(new NewThreadTask()); } static final class IOTask implements Callable<Scheduler> { @Override public Scheduler call() throws Exception { return IoHolder.DEFAULT; } } //最终会调用这个 static final class IoHolder { static final Scheduler DEFAULT = new IoScheduler(); }
- 上面的代码可以看出，
Schedulers.io()最终返回的是IoScheduler，所以scheduler.scheduleDirect(…)这句代码中的scheduler就是IoScheduler，而scheduleDirect方法是IOTask父类中的方法，代码如下
public Disposable scheduleDirect(@NonNull Runnable run) { return scheduleDirect(run, 0L, TimeUnit.NANOSECONDS); } //最终调用的是这个方法 public Disposable scheduleDirect(@NonNull Runnable run, long delay, @NonNull TimeUnit unit) { //这里的createWorker调用的是IoScheduler中的方法 final Worker w = createWorker(); //仍然是Runnable final Runnable decoratedRun = RxJavaPlugins.onSchedule(run); //实例化DisposeTask并将decoratedRun及w注入 DisposeTask task = new DisposeTask(decoratedRun, w); //调用的w的schedule方法，将上面的三个值作为参数 w.schedule(task, delay, unit); return task; }
- 这里，详细看下
final Worker w = createWorker();
- 这句代码，
createWorker()方法的代码如下
public Worker createWorker() { return new EventLoopWorker(pool.get()); }
- 可以看到这句代码的作用是实例化了EventLoopWorker并返回。 接着看
w.schedule(task, delay, unit);
- 这句代码，w为EventLoopWorker，所以这里是EventLoopWorker类中的
schedule方法，代码如下
public Disposable schedule(@NonNull Runnable action, long delayTime, @NonNull TimeUnit unit) { if (tasks.isDisposed()) { // don't schedule, we are unsubscribed return EmptyDisposable.INSTANCE; } //最终调用的是NewThreadWorker类中的方法 return threadWorker.scheduleActual(action, delayTime, unit, tasks); } //最终会调用这个方法 public ScheduledRunnable scheduleActual(final Runnable run, long delayTime, @NonNull TimeUnit unit, @Nullable DisposableContainer parent) { Runnable decoratedRun = RxJavaPlugins.onSchedule(run); //实例化了ScheduledRunnable ScheduledRunnable sr = new ScheduledRunnable(decoratedRun, parent); if (parent != null) { //不会进入这个方法 if (!parent.add(sr)) { return sr; } } Future<?> f; try { if (delayTime <= 0) { //重点，这里是把ScheduledRunnable放进了线程池中，关于java线程 //池大家可以自行研究 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; }
- 上面中的代码已经有了一些注释，最重要的就是这句代码
if (delayTime <= 0) { f = executor.submit((Callable<Object>)sr); } else { f = executor.schedule((Callable<Object>)sr, delayTime, unit); }
- 这段代码的作用就是将任务放进了线程池中等待执行。 同样，这段代码就是
subscribeOn是怎样将要处理的数据放到到工作线程的？
- 这个问题的答案。
通过分析“3”处的代码，可以发现其实就是将SubscribeTask实例进行了一层层的包装，然后丢到线程队列中等待执行，为了便于理解，我画了一下包装层级图，如下
根据上面的层级关系，会发现最后会调到SubscribeTask的run方法，这个方法又调用了上游的subscribe方法，而这个上游就是ObservableCreate类，由于前面两篇已经分析过了这个类，这里就不再分析。根据前面两篇的分析，知道最后会调用发射数据的方法，而这时这个发射数据其实就已经是在工作线程中了。
