对于为什么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线程切换的方法,最终打印的结果会是一条还是两条呢?
结果可以说明一切:
这里,我通过对源码的理解为大家回答为什么只会调用一次呢?
首先看下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(...)