上一篇：RxJava：基本订阅流程
我们在Rxjava中最常用的两个方法：
- subscribeOn(Scheduler scheduler)
- observeOn(Scheduler scheduler)
通常，我们使用subscribeOn(Schedulers.io())方法指定在IO线程中订阅-----进行数据处理，observeOn(AndroidSchedulers.mainThread())方法指定在主线程中观察-----进行UI操作。
so，Rxjava是如何进行线程切换的？
Observable.subscribeOn(Scheduler scheduler)
public final Observable<T> subscribeOn(Scheduler scheduler) {
ObjectHelper.requireNonNull(scheduler, "scheduler is null");
return RxJavaPlugins.onAssembly(new ObservableSubscribeOn<T>(this, scheduler));//1
}
可以看到，subscribeOn()方法与Observable.create()类似，都返回了一个Observable<T>对象，其内部主要调用了ObservableSubscribeOn(ObservableSource<T> source, Scheduler scheduler)方法，构造了一个ObservableSubscribeOn对象。ObservableSubscribeOn类不长，我们贴出它的全部代码，如下：
public final class ObservableSubscribeOn<T> extends AbstractObservableWithUpstream<T, T> {
final Scheduler scheduler;
public ObservableSubscribeOn(ObservableSource<T> source, Scheduler scheduler) {//2
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);
//Disposable用来做资源处理，在这里我们不用关心
parent.setDisposable(scheduler.scheduleDirect(new Runnable() {//3
@Override
public void run() {
source.subscribe(parent);
}
}));
}
static final class SubscribeOnObserver<T> extends AtomicReference<Disposable> implements Observer<T>, Disposable {
private static final long serialVersionUID = 8094547886072529208L;
final Observer<? super T> actual;
final AtomicReference<Disposable> s;
SubscribeOnObserver(Observer<? super T> actual) {
this.actual = actual;
this.s = new AtomicReference<Disposable>();
}
@Override
public void onSubscribe(Disposable s) {
DisposableHelper.setOnce(this.s, s);
}
@Override
public void onNext(T t) {
actual.onNext(t);
}
@Override
public void onError(Throwable t) {
actual.onError(t);
}
@Override
public void onComplete() {
actual.onComplete();
}
@Override
public void dispose() {
DisposableHelper.dispose(s);
DisposableHelper.dispose(this);
}
@Override
public boolean isDisposed() {
return DisposableHelper.isDisposed(get());
}
void setDisposable(Disposable d) {
DisposableHelper.setOnce(this, d);
}
}
}
ObservableSubscribeOn类继承关系如下：
ObservableSubscribeOn<T> extends AbstractObservableWithUpstream
bstractObservableWithUpstream<T, U> extends Observable<U>
ObservableSubscribeOn类也是一个Observable，从名字上也可以看出，这是一个进行了SubscribeOn的Observable。那么，他和我们上一篇提到的ObservableCreate类最大的区别在哪里？
再贴出Observable实现类中的关键方法：ObservableCreate.subscribeActual(Observer<? super T> observer)，做一个对比：
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);
}
}
ObservableCreate内部定义了发射器类-----CreateEmitter，observer.onSubscribe()方法传入的是CreateEmitter发射器对象。
-
ObservableSubscribeOn内部定义了一个SubscribeOnObserver对象，subscribeActual方法中，s.onSubscribe()传入的是这个SubscribeOnObserver对象。即对Observer做一次重新包装，放入一个Runnable中执行。
SubscribeOnObserver也是一个Observer，将我们传入的Observeable进行了包装
ObservableCreate.subscribeActual中，执行了source.subscribe(parent)即使用发射器执行我们定义的subscribe(ObservableEmitter<Bundle> e)方法；-
ObservableSubscribeOn.subscribeActual中，调用的是scheduler.scheduleDirect(Ruunable run)，传入一个Runnable，来执行source.subscribe(parent)。我们使用发射器发射的数据，是通过subscribe()调用的，而subscribe()真正被调用的，是在Observable实现类ObservableSubscribeOn.subscribeActual()。
我们主要看一下关键逻辑步骤：scheduler.scheduleDirect()方法，了解Runnable是如何在我们指定线程中执行的。
public Disposable scheduleDirect(Runnable run, long delay, TimeUnit unit) {
final Worker w = createWorker();
final Runnable decoratedRun = RxJavaPlugins.onSchedule(run);
w.schedule(new Runnable() {
@Override
public void run() {
try {
decoratedRun.run();
} finally {
w.dispose();
}
}
}, delay, unit);
return w;
}
我们可以看到，其内部调用的了Worker.schedule()方法，来实际执行我们传入的Runnable。而createWorker()是一个抽象方法，其具体实现由我们传入的Scheduler的实现类决定，即我们调用subscribeOn(Scheduler scheduler)中传入的Scheduler。在上文代码中，我用注释//1 //2 //3标明了这条引用链。
我们就以Schedulers.io()为例：我们具体看下它的实现。我们先来看一下Schedulers.io()所代表的Scheduler：
public static Scheduler io() {
return RxJavaPlugins.onIoScheduler(IO);
}
static {
......
IO = RxJavaPlugins.initIoScheduler(new Callable<Scheduler>() {
@Override
public Scheduler call() throws Exception {
return IoHolder.DEFAULT;
}
});
......
}
static final class IoHolder {
static final Scheduler DEFAULT = new IoScheduler();
}
Schedulers.io()返回的是一个Scheduler对象。在IoScheduler类中，createWorker()返回了一个EventLoopWorker对象，EventLoopWorker.schedule方法如下：
public Disposable schedule(Runnable action, long delayTime, TimeUnit unit) {
if (tasks.isDisposed()) {
// don't schedule, we are unsubscribed
return EmptyDisposable.INSTANCE;
}
return threadWorker.scheduleActual(action, delayTime, unit, tasks);
}
scheduleActual方法中，对Runnable进行了一系列封装，最终使用的是一个Android的一个ExecutorService实现类----ScheduledExecutorService来执行。
public ScheduledRunnable scheduleActual(final Runnable run, long delayTime, TimeUnit unit, 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) {
parent.remove(sr);
RxJavaPlugins.onError(ex);
}
return sr;
}
这个executor构造如下:
public NewThreadWorker(ThreadFactory threadFactory) {
executor = SchedulerPoolFactory.create(threadFactory);
}
public static ScheduledExecutorService create(ThreadFactory factory) {
final ScheduledExecutorService exec = Executors.newScheduledThreadPool(1, factory);
if (exec instanceof ScheduledThreadPoolExecutor) {
ScheduledThreadPoolExecutor e = (ScheduledThreadPoolExecutor) exec;
POOLS.put(e, exec);
}
return exec;
}
我们从线程池中获取一个ExecutorService，使用该ExecutorService执行我们传入的Runnable。总结以上：Observable.observeOn(Schedulers.io()) 切换子线程的原理在于：ObservableOnSubscribe中自定义的subscribe()，实际被传入的Scheduler实现类的Worker.schedule()方法指定线程中执行。
Observable. observeOn(Scheduler scheduler)
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()是一个final方法，所以所有的Observeable都是调用的该方法，它实际上返回的是一个ObservableObserveOn对象实例。在切换完线程后使用subscribe()方法定义订阅观察者进行ui操作，在上一篇已经讲到，Observeable.subscribe实际执行的是Observable实现类的subscribeActual方法，所以，我们需要关注的是ObservbleObserveOn.subscribeActual()。
ObservableObserveOn.subscribeActual()代码如下：
protected void subscribeActual(Observer<? super T> observer) {
//如果传入的scheduler指代当前线程，不进行线程切换，直接调用原observer的subscribe()方法，此处的 //source是上游的ObservableSource(Observable)
if (scheduler instanceof TrampolineScheduler) {
source.subscribe(observer);
} else {
Scheduler.Worker w = scheduler.createWorker();
source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
}
}
这里重点关注的是else中的线程切换流程，和subscribeOn()方法进行线程切换类似，也是由传入的Scheduler决定Worker。这里引入ObserveOnObserver类，传入的observer进行封装，即决定observer的运行线程。
ObserveOnObserver类实现了Runnable接口，run方法如下：
//ObserveOnObserver实现的run方法，drainFused和drainNormal中执行了Observer的回调方法。
public void run() {
if (outputFused) {
drainFused();
} else {
drainNormal();
}
}
ObserveOnObserver的worker主要作用于schedule()方法，而schedule()如下，即调用worker.schedule()方法，执行当前ObserveOnObserver。
void schedule() {
if (getAndIncrement() == 0) {
worker.schedule(this);
}
}
So，observeOn方法也是通过Scheduler来实现线程切换的。而AndroidSchedulers.mainThread()所代表的Scheduler如下：
static final Scheduler DEFAULT = new HandlerScheduler(new Handler(Looper.getMainLooper()));
在schedle()方法中，我们看到是通过handler，将传入的Runnable发送到主线程执行。
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);
//将传入的Runnable和handler封装成一个ScheduledRunnable，作为Message的Callback
ScheduledRunnable scheduled = new ScheduledRunnable(handler, run);
//使用Handler将消息发送MainLooper
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;
}
