阅读本文之前建议先阅读RxJava2.+创建流程源码分析
Demo
Observable observable = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
Log.d(TAG, "subscribe work thread is " + Thread.currentThread().getName());
emitter.onNext("1");
Log.d(TAG, "发送1");
emitter.onNext("2");
Log.d(TAG, "发送2");
emitter.onNext("3");
Log.d(TAG, "发送3");
emitter.onNext("4");
Log.d(TAG, "发送4");
emitter.onComplete();
Log.d(TAG, "发送onComplete");
}
});
Observer observer = new Observer<String>() {
Disposable disposable;
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "onSubscribe work thread is " + Thread.currentThread().getName());
disposable = d;
}
@Override
public void onNext(String value) {
Log.d(TAG, "onNext data is :" + value);
Log.d(TAG, "onNext work thread is " + Thread.currentThread().getName());
if (value.equals("2")) {
//执行了hello之后终止
disposable.dispose();
}
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "onError data is :" + e.toString());
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete");
}
};
observable.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(observer);
输出：
onSubscribe work thread is main
subscribe work thread is RxCachedThreadScheduler-1
发送1
发送2
发送3
onNext data is :1
发送4
发送onComplete
onNext work thread is main
onNext data is :2
onNext work thread is main
如果你有兴趣多测试几次，你会发现每次日志输出得到顺序是有可能不不一样的。为什么呢？接下来进入源码分析。（Obserable的创建和subscribe调用过程请参考的另一篇博文 ）
Observable#subscribeOn
@SchedulerSupport(SchedulerSupport.CUSTOM)
public final Observable<T> subscribeOn(Scheduler scheduler) {
ObjectHelper.requireNonNull(scheduler, "scheduler is null");
return RxJavaPlugins.onAssembly(new ObservableSubscribeOn<T>(this, scheduler));
}
通过上一篇文章我们已经知道当订阅发生时，真正调用的地方是subscribeActual，所以我们直接进入ObservableSubscribeOn的subscribeActual方法。如下：
image.png
很明显我们通过线程池去执行一个线程，我们去看一下SubscribeTask的run方法：
image.png
source就是我们的上级Observable，所以我们创建的Observable的subscribe方法中说执行的操作自然在我们指定的线程中执行。
Observable#observeOn
同理我们直接进入ObservableObserveOn的subscribeActual方法：
image.png
可以看到这里并没有直接调用线程执行，所以调用真正发生的地方肯定在ObserveOnObserver的onNext方法中，通过查找，我们发现以下代码：
image.png
这是真正的线程执行发生的地方。我们仔细看这个参数是this，所以说ObserveOnObserver实行了Runnable接口，我们去看下他的run方法：
image.png
有2个，我们还是看normal的吧。这个方法有点复杂，2个死循环，什么鬼？先不管它，仔细看代码里面有2句话：
final Observer<? super T> a = actual;
a.onNext(v);
在指定的线程中执行了actual.onNext(),所以这就达到了为下级操作符切换线程的效果。
总结
好了，关于Rxjava的线程切换到这就分析完了。
