先上被分析的代码：
val observer = object : Observer<String> {
override fun onSubscribe(d: Disposable) {
Log.d(TAG, "onSubscribe")
Log.d(TAG, "Observer work thread is" + Thread.currentThread().name)
disposable = d
}
override fun onNext(s: String) {
Log.d(TAG, "Observer onNext data is :$s")
Log.d(TAG, "Observer work thread is" + Thread.currentThread().name)
if (s == "hello") {
//执行了hello之后终止
disposable.dispose()
}
val compositeDisposable = CompositeDisposable()
compositeDisposable.dispose()
}
override fun onError(e: Throwable) {
Log.d(TAG, "Observer onError data is :" + e.toString())
}
override fun onComplete() {
Log.d(TAG, "Observer onComplete")
}
}
val observable = Observable.create(object : ObservableOnSubscribe<String> {
@Throws(Exception::class)
override fun subscribe(e: ObservableEmitter<String>) {
Log.i(TAG, "Observable work thread is" + Thread.currentThread().name)
e.onNext("hello")
Log.i(TAG, "Observable 发送 hello")
e.onNext("world")
Log.i(TAG, "Observable 发送 world")
e.onComplete()
Log.i(TAG, "Observable 调用 onComplete")
}
})
observable.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(observer)
这次我们倒着分析，在调用.subscribe(observer)方法的时候，最终会调用调用方的subscribeActual方法，调用方是observeOn方法返回的Observable，查看observeOn方法内部，发现返回的是ObservableObserveOn类，然后我们看ObservableObserveOn类的subscribeActual方法：
@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));
}
}
if (scheduler instanceof TrampolineScheduler)先跳过，我们的scheduler也不是TrampolineScheduler，所以执行的是else下面的代码，Scheduler.Worker w = scheduler.createWorker();获取到的就是observeOn方法中传入的Scheduler，我们这里传入的是AndroidSchedulers.mainThread()，而source是上游的Observable（也就是调用subscribeOn返回的Observable），source.subscribe传入的参数是包装后的Observer，其中包含了当前的scheduler，然后source最终也会执行它自己的subscribeActual方法，而当前source是ObservableSubscribeOn，所以我们看ObservableSubscribeOn类的该方法：
@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)));
}
然后就和上一篇文章一样了
注意：如果有多个observeOn方法，通过ObservableObserveOn类的subscribeActual方法可知，会从下到上依次订阅observer
未完待续
