本篇文章目的：分析subscribeOn()多次调用，为什么只有第一次起作用。
我们在Android主线程调用下面的方法，以这个例子，进行分析。
private void test() {
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
emitter.onComplete();
}
})
.subscribeOn(Schedulers.io())
.subscribeOn(Schedulers.computation())
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.e(TAG, "onSubscribe: ");
}
@Override
public void onNext(Integer integer) {
Log.e(TAG, "onNext: " + integer);
}
@Override
public void onError(Throwable e) {
Log.e(TAG, "onError: " + e.getMessage());
}
@Override
public void onComplete() {
Log.e(TAG, "onComplete: ");
}
});
}
先来张图
Observable的create()方法简化版
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
return new ObservableCreate<T>(source);
}
create()方法返回的返回Observable是一个ObservableCreate对象。
Observable的subscribeOn()方法简化版
public final Observable<T> subscribeOn(Scheduler scheduler) {
return new ObservableSubscribeOn<T>(this, scheduler);
}
subscribeOn()方法返回的返回Observable是一个ObservableSubscribeOn对象。
Observable的subscribe()方法简化版
public final void subscribe(Observer<? super T> observer) {
subscribeActual(observer);
}
当调用subscribe()方法以后，观察者和被观察者才真正的关联起来。方法内部调用Observable的subscribeActual()方法。
ObservableSubscribeOn的subscribeActual()方法
public void subscribeActual(final Observer<? super T> observer) {
//1. 使用传入的observer构建一个SubscribeOnObserver对象
final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(observer);
//2. 调用Observer的onSubscribe()方法
observer.onSubscribe(parent);
//3. 使用指定的scheduler立即调度
parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
}
- 使用传入的observer构建一个SubscribeOnObserver对象
传入的observer对象。
new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.e(TAG, "onSubscribe: ");
}
@Override
public void onNext(Integer integer) {
Log.e(TAG, "onNext: " + integer);
}
@Override
public void onError(Throwable e) {
Log.e(TAG, "onError: " + e.getMessage());
}
@Override
public void onComplete() {
Log.e(TAG, "onComplete: ");
}
}
SubscribeOnObserver类是ObservableSubscribeOn类的静态内部类
static final class SubscribeOnObserver<T> extends AtomicReference<Disposable>
implements Observer<T>, Disposable {
private static final long serialVersionUID = 8094547886072529208L;
final Observer<? super T> downstream;
final AtomicReference<Disposable> upstream;
SubscribeOnObserver(Observer<? super T> downstream) {
this.downstream = downstream;
this.upstream = new AtomicReference<Disposable>();
}
@Override
public void onSubscribe(Disposable d) {
DisposableHelper.setOnce(this.upstream, d);
}
@Override
public void onNext(T t) {
downstream.onNext(t);
}
@Override
public void onError(Throwable t) {
downstream.onError(t);
}
@Override
public void onComplete() {
downstream.onComplete();
}
@Override
public void dispose() {
DisposableHelper.dispose(upstream);
DisposableHelper.dispose(this);
}
@Override
public boolean isDisposed() {
return DisposableHelper.isDisposed(get());
}
void setDisposable(Disposable d) {
DisposableHelper.setOnce(this, d);
}
}
SubscribeOnObserver类对传入的下游观察者做了一层包装，增加了dispose相关功能。
调用Observer的onSubscribe()方法，这里可以看到Observer的onSubscribe()方法最先调用。 Observer的onSubscribe()方法和subscribe()方法在同一个线程执行。在上面的例子中，我们是在主线程进行的操作，所以此时onSubscribe()方法是运行在主线程的。
使用指定的scheduler立即调度
3.1 首先构建了一个对象SubscribeTask。SubscribeTask是ObservableSubscribeOn一个内部类。
final class SubscribeTask implements Runnable {
private final SubscribeOnObserver<T> parent;
SubscribeTask(SubscribeOnObserver<T> parent) {
this.parent = parent;
}
@Override
public void run() {
//当被调度的时候，会执行run()方法
source.subscribe(parent);
}
}
3.2 对于subscribeOn(Schedulers.computation())方法返回的ObservableSubscribeOn对象来说，调用scheduler的scheduleDirect()方法。 Schedulers.computation()是一个ComputationScheduler对象,内部保存固定线程数的工作线程池，scheduleDirect()方法会从中取出一个线程来执行传入的SubscribeTask对象，最终会调用SubscribeTask对象的run()方法的。
SubscribeTask的run()方法
public void run() {
//当被调度的时候，会执行run()方法
source.subscribe(parent);
}
注意!!!，这时候，SubscribeTask的run()方法是运行在ComputationScheduler的线程池中的。
这里的source就是subscribeOn(Schedulers.io())返回的ObservableSubscribeOn对象。
ObservableSubscribeOn的subscribeActual()方法会再次执行。
ObservableSubscribeOn的subscribeActual()方法
public void subscribeActual(final Observer<? super T> observer) {
//1. 使用传入的observer构建一个SubscribeOnObserver对象
final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(observer);
//2. 调用Observer的onSubscribe()方法
observer.onSubscribe(parent);
//3. 使用指定的scheduler立即调度
parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
}
3.2 对于subscribeOn(Schedulers.io())方法返回的ObservableSubscribeOn对象来说，调用scheduler的scheduleDirect()方法。我们传入的scheduler是Schedulers.io()，是一个IoScheduler对象。IoScheduler内部维护了一个线程池ScheduledThreadPoolExecutor，核心线程数量是1，最大线程数量是Integer.MAX_VALUE。scheduleDirect()方法会从中取出一个线程来执行传入的SubscribeTask对象，最终会调用SubscribeTask对象的run()方法的。
SubscribeTask的run()方法
public void run() {
//当被调度的时候，会执行run()方法，这个source就是ObservableCreate对象
source.subscribe(parent);
}
!!!，注意这时候，SubscribeTask的run()方法是运行在IoScheduler的线程池中的。
这是轮到ObservableCreate对象的subscribeActual()方法了。
@Override
protected void subscribeActual(Observer<? super T> observer) {
//发出数据的对象，包装下游的观察者
CreateEmitter<T> parent = new CreateEmitter<T>(observer);
observer.onSubscribe(parent);
try {
//1. 调用source的subscribe()方法
source.subscribe(parent);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
parent.onError(ex);
}
}
- 调用source的subscribe()方法，在这里最终发出了数据，这里的source就是我们传入的匿名的ObservableOnSubscribe对象。
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
emitter.onComplete();
}
})
最终发出数据的对象是CreateEmitter对象
static final class CreateEmitter<T>
extends AtomicReference<Disposable>
implements ObservableEmitter<T>, Disposable {
private static final long serialVersionUID = -3434801548987643227L;
final Observer<? super T> observer;
CreateEmitter(Observer<? super T> observer) {
this.observer = observer;
}
@Override
public void onNext(T t) {
if (!isDisposed()) {
observer.onNext(t);
}
}
@Override
public void onError(Throwable t) {
if (!tryOnError(t)) {
RxJavaPlugins.onError(t);
}
}
@Override
public boolean tryOnError(Throwable t) {
if (!isDisposed()) {
try {
observer.onError(t);
} finally {
dispose();
}
return true;
}
return false;
}
@Override
public void onComplete() {
if (!isDisposed()) {
try {
observer.onComplete();
} finally {
dispose();
}
}
}
}
CreateEmitter对象调用onNext()方法，onError()方法，onComplete()方法的时候，会分别调用下游观察者（就是我们手写的observer）的对应的 onNext()方法，onError()方法，onComplete()方法。
总结：到这里我们可以看到，调用多次subscribeOn()方法，最终原始被观察者会在第一次调用subscribeOn()方法所指定的Scheduler对象所创建的线程中发出数据。
