再忆RxJava---线程切换
RxJava已经陆陆续续看了将近两年多了，觉得很有必要重新认识一下。而且最新github上都出了3.0了。重新看一下，一来作为回顾，二来也算是学习一下3.0
在了解线程切换前，先回顾一下简单的同步操作
1.同步
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) {
// onNext
}
}).subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(String s) {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
Observable.create会生成一个ObservableCreate，subscribe最终会调用ObservableCreate的subscribeActual方法
很明显，同步的原理很简单。Observable.create生成ObservableCreate，ObservableCreate会new一个emitter出来，这个emitter就包裹了Observer，通过emitter来完成Observer函数的调用
2.异步
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) {
}
}).subscribeOn(new IoScheduler())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(String s) {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
Observable.create还是一样的，生成ObservableCreate
subscribeOn会生成ObservableSubscribeOn
observeOn会生成ObservableObserveOn
最后subscribe就会调用各自的subscribeActual方法
2.1 ObservableObserveOn
@Override
protected void subscribeActual(Observer<? super T> observer) {
if (scheduler instanceof TrampolineScheduler) {
source.subscribe(observer);
} else {
Scheduler.Worker w = scheduler.createWorker();
source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
}
}
这个observer就是我们自己写的Observer
scheduler就是HandlerScheduler（里面有Handler）
然后触发下一个subscribeActual
说白了就是我自己写的Observer被ObserveOnObserver包裹了一层
2.2 ObservableSubscribeOn的subscribeActual方法
@Override
public void subscribeActual(final Observer<? super T> observer) {
final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(observer);
observer.onSubscribe(parent);
parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
}
会用SubscribeOnObserver来包裹上一层的Observer（ObserveOnObserver），scheduler很明显就是IoScheduler
IoScheduler其实就是生成线程池处理SubscribeTask
SubscribeTask代码里面（ source.subscribe(parent) ）source就是ObservableCreate，所以就执行我们自己写的subscribe方法
总结：生成一个线程池，启动一个线程处理我们自己写的subscribe方法。
2.3 ObservableCreate
这个其实就和同步的一样
3 具体的流程图如下
异步线程切换流程图.png
简单来说，就是把所有的操作从右到左包装成一对观察者与被观察者的关系，然后通过发射器使所有的操作连续执行
4 流程举例
4.1 下载并显示图片
- 创建CreateEmitter，下载，等到结束，发射onNext。这是出于流程图中的（1），onNext在子线程中发射（网络请求一般会自己new Thread出来执行的）
（注意：此时已经有子线程处理了，所以subscribeOn其实已经没有意义了，可以不写。直接看observeOn）
- 接下来走到ObserveOnObserver的onNext方法
ObserveOnObserver本身也是Runnable，所以会执行自身的run方法。通过源码可以很清楚的知道，其实就是把上一个步骤中的结果收集起来，放到队列里，然后poll出来。poll结束就直接onNext。（有两点需要说明）
- 这里还区分同步异步。同步不进入队列，q.poll=null，直接onNext。异步进队列，直到q.poll != null 才会onNext
- 传进来onNext的时候，是处于下载线程中，传出去onNext已经经过Handler处理
- poll结束就走到我们自己写的Observer的onNext方法
4.2 批量处理图片并显示
这个时候是需要subscribeOn的，毕竟批量处理图片的时候，subscribe函数中我们不会自己去new Thread出来
（简单来说，如果subscribe函数中的耗时操作没有new Thread，那么就需要subscribeOn）
是否需要subscribeOn还需要注意创建retrofit的时候RxJava2CallAdapterFactory的创建方式，如果用了createWithScheduler，那么就不用写subscribeOn了，里面默认有了，具体可参见源码
5 subscribeOn和ObserveOn
ObserveOn每次都有效是因为每次都有scheduler.createWorker操作，举例来说，ComputationScheduler以及其他的IoScheduler都会生成线程池来操作，那么必然会跑到对应的线程池中的线程中操作
下面我们来谈谈subscribeOn只有首次有效的问题
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(final ObservableEmitter<String> emitter) {
emitter.onNext("111");
emitter.onComplete();
}
}).subscribeOn(new IoScheduler())
.map(new Function<String, String>() {
@Override
public String apply(String s) {
Log.e("IoScheduler", Thread.currentThread().getName());
return "555";
}
})
.subscribeOn(new ComputationScheduler())
.map(new Function<String, String>() {
@Override
public String apply(String s) {
Log.e("ComputationScheduler", Thread.currentThread().getName());
return "222";
}
})
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(String s) {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
两个log打印出来的都是相同的值，很明显，只有首次“有效”。其实真的是有没有效的问题么？我在ObservableSubscribeOn的subscribeActual方法里面debug，发现ComputationScheduler在主线程，IoScheduler在Computation线程中，subscribe方法在IoScheduler线程中，很明显，其实是有效的，确实切换了。那为什么会有无效的说法呢？其实也很好理解，我们的操作在A线程中执行，而A在线程B中执行，请问，我们的操作在哪个线程中执行？肯定是A啊（说B其实也没错，但是从学术角度来讲不准确）。
