本篇文章目的：分析subscribeOn()多次调用，为什么只有第一次起作用。
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()方法是在主线程被调用的
.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)));
}
先来张图
RxJava2InvokeSubscribeOnManyTimes.png
- 使用传入的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()方法。我们传入的schedulerSchedulers.computation(), Schedulers.computation()是一个SingleScheduler对象。这个调度的详细细节就不去深究了，大致就是SingleScheduler内部维护了一个线程池，scheduleDirect()方法会从中取出一个线程来执行传入的SubscribeTask对象，内部会对传入的SubscribeTask对象进行一系列的包装，但是最终总是会调用SubscribeTask对象的run()方法的。
public void run() {
//当被调度的时候，会执行run()方法
source.subscribe(parent);
}
这里的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(), Schedulers.io()是一个IoScheduler对象。这个调度的详细细节就不去深究了，大致就是IoScheduler内部维护了一个线程池，scheduleDirect()方法会从中取出一个线程来执行传入的SubscribeTask对象，内部会对传入的SubscribeTask对象进行一系列的包装，但是最终总是会调用SubscribeTask对象的run()方法的。
这是轮到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()方法，在这里最终发出了数据
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()方法的时候，会分别调用下游观察者的对应方法。
总结：到这里我们可以看到，调用多次subscribeOn()方法，最终原始被观察者会在第一次调用subscribeOn()方法所指定的Scheduler对象所创建的线程中发出数据。并且在这个过程中，构建的一系列的原始被观察者的包装对象内部并没有做切换线程的操作，所以原始被观察者是和原始被观察者在同一个线程中工作的。
网友评论