...
Single.just(1).map(new Function<Integer, String>() {
@Override
public String apply(Integer integer) throws Exception {
return String.valueOf(integer);
}
}).subscribeOn(Schedulers.io()).observeOn(Schedulers.single())
.subscribe(new SingleObserver<String>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onSuccess(String value) {
}
@Override
public void onError(Throwable e) {
}
});
...
Single.just(1)主要是返回SingleJust然后繼續(xù)點擊查看map源碼其主要放回的是new SingleMap<T, R>(this, mapper)其中this是Single.just(1)返回的，然后繼續(xù)subscribeOn()其返回的是SingleSubscribeOn，同樣實例化的時候都是把上游傳進去，observeOn返回的是new SingleObserveOn<T> 同樣也是把上游傳進去：如圖

Rxjava.png

然后我來說一下為什么subscribeOn為什么影響上游的線程，看源碼知道他是在@Override
protected void subscribeActual(final SingleObserver<? super T> observer) {
final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(observer, source);
observer.onSubscribe(parent);
Disposable f = scheduler.scheduleDirect(parent);
parent.task.replace(f);
}
可以看到她只是做了兩件事情一個是通知下游訂閱開始了onSubscribe
另一個是切換到我們指定的線程中去
public void run() {
source.subscribe(this);
}
可以看到其是在指定的線程中訂閱的。
observeOn為什么只能影響下游了，看源碼可知
他不是在訂閱的時候切換線程而是在返回消息的時候切換線程
public void run() {
Throwable ex = error;
if (ex != null) {
downstream.onError(ex);
} else {
downstream.onSuccess(value);
}
}