对于为什么RxJava中的线程切换方法subscribeOn为什么多次调用只有第一次有效的问题，经过本人对源码的研究，有所了解了。
首先说一个小细节，RxJava中包括操作符在内很多方法返回的大多都是中间变量Observable被观察者，这样的好处就是便于时间的传递。
接着放上一个简单的例子（为了使大家看的明白，我将一段代码分为几段来书写）：
Observable<String> observable0 = Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("RxJava do call");
}
});
Observable<String> observable1 = observable0.subscribeOn(Schedulers.io());
Observable<String> observable2 = observable1.subscribeOn(Schedulers.newThread());
observable2.observeOn(AndroidSchedulers.mainThread());
observable2.subscribe(new Action1<String>() {
@Override
public void call(String s) {
Log.e("myTag", s);
}
});
observable2.subscribe();
大家可以看到，这里我分别调用了两次subscribeOn线程切换的方法，最终打印的结果会是一条还是两条呢？
结果可以说明一切：
image.png
这里，我通过对源码的理解为大家回答为什么只会调用一次呢？
首先看下subscribeOn线程切换方法：
public final Observable<T> subscribeOn(Scheduler scheduler) {
return this instanceof ScalarSynchronousObservable
?((ScalarSynchronousObservable)this).scalarScheduleOn(scheduler)
:create((Observable.OnSubscribe)(new OperatorSubscribeOn(this, scheduler)));
}
接着会调用三目运算符的create方法，该create方法实际上只是生成了下一个Observable，故此略过，我们看下里面的参数OperatorSubscribeOn，该参数实际上是一个OnSubscribe类型，当RxJava最终执行订阅的时候会执行该对象的call方法，不懂的可以看下源码，这里不做详解。我们看下OperatorSubscribeOn的call方法实现：
public void call(final Subscriber<? super T> subscriber) {
final Worker inner = this.scheduler.createWorker();
subscriber.add(inner);
inner.schedule(new Action0() {
public void call() {
final Thread t = Thread.currentThread();
Subscriber<T> s = new Subscriber<T>(subscriber) {
......
};
OperatorSubscribeOn.this.source.unsafeSubscribe(s);
}
});
}
其中inner.schedule这个方法就是subscribeOn线程切换的关键方法，里面的具体执行逻辑放在线程池当中，具体实现逻辑这里也不做讲解。
然后就是OperatorSubscribeOn.this.source.unsafeSubscribe(s)这个方法，这个方法很重要，就是通过该方法实现的RxJava方法订阅，其中source就是在生成observable1的时间传递过来的observable0。
作为额外插曲，还是放上源码吧，unsafeSubscribe的实现逻辑如下：
public final Subscription unsafeSubscribe(Subscriber<? super T> subscriber) {
try {
subscriber.onStart();
hook.onSubscribeStart(this, this.onSubscribe).call(subscriber);
return hook.onSubscribeReturn(subscriber);
} catch (Throwable var6) {
......
}
}
其中call方法的执行，就是observable0中的OnSubscribe的call方法调用，参见最上面的例子。
好了，重点来了，这里我又通过subscribeOn方法再次生成了observable2，在最终线程切换里又会调用了call方法，该方法具体是谁的调用呢，就是observable1。至此大家也明白了，这就是一个链式的调用，最终调用的就是最开始的observable0的OnSubscribe的call方法，所以，无论你做了多少次线程切换，最终都会递归切换到第一次的切换方法，所以，也就只有第一次线程切换才有效了。
线程切换：
observable0.subscribeOn(...) ---> observable1
observable1.subscribeOn(...) ---> observable2
...
observable(n-1).subscribeOn(...) ---> observablen
RxJava事件订阅：
observablen.subscribe(...) ---> observable(n-1).call(...) ---> ...... ---> observable1.call(...) ---> observable0.call(...)
网友评论