RxJava的subscribeOn多次切换只有第一次有效

对于为什么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(...)