ObservableOn()
直接查看实现，会发现onSubscribe()中做了一些判断，比如82 104等几行都是做了一些同步 异步 等的判断，然后初始化Disposable，onSubscribe()是上游Observable完成了整条订阅链之后调用的，所以这些操作是在开始订阅之后才初始化操作，然后106行可以看出把一个包装处理过的Disposable传递给下游
image.png
和之前的一样，subscribeActual方法里会将observer进行包装，然后传递给source也就是上游进行订阅
* `40`行进行了判断所传进来的`scheduler`是否跟原本的线程一致，如果是一样的就直接传递不用进行处理
* `43`行创建了一个对应`scheduler`的`worker`，`worker`在后续负责把数据在对应的线程进行发射操作
image.png
发射数据onNext处理
@Override
public void onNext(T t) {
...
// 前面的都先忽略掉，会发现最后会调用这个方法
schedule();
}
void schedule() {
if (getAndIncrement() == 0) {
// 在这个可以看到，上面根据schedule的worker执行了schedule(),并且把自身传进去，this其实实现了runnable，所以可以理解为传了一个runnable进去
worker.schedule(this);
}
}
接着上面的以AndroidSchedulers.mainThread()这个scheduler为例,这里实际上是将主线程的looper传进去了
image.png
查看一下这个scheduler的worker，会发现worker的基类schedule()方法是相同的互相调用的，所以可以直接看多个参数的schedule(),可以看到73行创建了一个ScheduledRunnable对象，并且把主线程的handler以及外面的Observer传递过去，接着82行用主线程的handler发送消息，119行ScheduledRunnable里的run被调用，接着Observer也就是runnable也调用run方法
image.png
image.png
到这里可以看出，实际上当切换线程的时候，observer(也实现了Runnable)的onNext往scheduler里发送自身，让scheduler来决定自身应该在什么线程执行run方法,接下来看回observer的run方法，就是判断了一下要执行哪个方法
image.png
可以看到最后是调用了onNext方法，到这里就完成了指定线程发射数据的功能
void drainNormal() {
int missed = 1;
final SimpleQueue<T> q = queue;
final Observer<? super T> a = downstream;
for (;;) {
if (checkTerminated(done, q.isEmpty(), a)) {
return;
}
for (;;) {
boolean d = done;
T v;
try {
v = q.poll();
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
disposed = true;
upstream.dispose();
q.clear();
a.onError(ex);
worker.dispose();
return;
}
boolean empty = v == null;
if (checkTerminated(d, empty, a)) {
return;
}
if (empty) {
break;
}
a.onNext(v);
}
missed = addAndGet(-missed);
if (missed == 0) {
break;
}
}
}
值得注意的是可以看到的是v也就是我们要发射的数据，是通过poll方法获取的，查看代码可以发现
queue实际上就是一个Disposable也就是说是上游Observable,通过上游的poll方法去获取要onNext的数据
image.png
查看Observable其中一个实现ObservableMap的poll方法，可以看到这里实际上也是调用上游的poll方法，并且对数据的格式也就是不允许为null做了一层判断
public U poll() throws Exception {
T t = qd.poll();
return t != null ? ObjectHelper.<U>requireNonNull(mapper.apply(t), "The mapper function returned a null value.") : null;
}
poll方法操作的对象实际上是下图104行的时候new出来的，具体查看其实就是缓存数据，类似一个容量池的作用
[图片上传失败...(image-a0f00c-1551301878533)]
image.png
用一段伪代码来展示切换线程之后的observer，其实相当于onNext等方法都被放在指定的线程里去发射数据
public class Observer {
Observer oldObserver;
public Observer(Observer observer) {
oldObserver = observer;
}
public void onNext(T t) {
// 一些其他操作
new Thread("Android mainThread") {
@Override
public void run() {
oldObserver.onNext(t);
}
} .start();
}
public void onError(Throwable e) {
// 一些其他操作
new Thread("Android mainThread") {
@Override
public void run() {
oldObserver.onError(e);
}
} .start();
}
public void onComplete() {
// 一些其他操作
new Thread("Android mainThread") {
@Override
public void run() {
oldObserver.onComplete();
}
} .start();
}
}
