先上代码：

ObservableOnSubscribe<Integer> oos = new ObservableOnSubscribe<Integer>() {
    @Override
    public void subscribe(ObservableEmitter<Integer> e) throws Exception {
         final int max = 100;
         for (int i = 1; i <= max; i++) {
             e.onNext(max);
         }
         e.onComplete();
     }
 };
 Observer<Integer> observer = new Observer<Integer>() {
     @Override
     public void onSubscribe(Disposable d) {
     }
     @Override
     public void onNext(Integer integer) {
     }
     @Override
     public void onError(Throwable e) {
     }
     @Override
     public void onComplete() {
     }
 };

 Observable.create(oos)
           .observeOn(AndroidSchedulers.mainThread())
           .subscribeOn(Schedulers.computation())
           .subscribe(observer);

上面是Rxjava最简单的实现模型。
从链式调用的返回值来看：

  Observable.create()------》ObservableCreate extends Observable
  ObservableCreate.observerOn()------->ObservableObserveOn extends  AbstractObservableWithUpstream  extends  Observable
  ObservableObserveOn.subscribeOn()------->ObservableSubscribeOn extends  AbstractObservableWithUpstream  extends  Observable

所以最后的调用对象是

  ObservableSubscribeOn.subscribe(observer)

从上面的返回值可以看出中间任一一个的返回值返回的都是observable的子对象。

为什么要强调中间几个的返回值都是observable的返回值，这里要先明确一下，待会会大量用到subscribe()方法，在Observable（子类）中的subscribe()方法：

public final void subscribe(Observer<? super T> observer) {
    ObjectHelper.requireNonNull(observer, "observer is null");
    try {
        //钩子，如果未设置的话，返回值还是observer
        observer = RxJavaPlugins.onSubscribe(this, observer);
        //空检验
        ObjectHelper.requireNonNull(observer, "Plugin returned null Observer");
        //核心
        subscribeActual(observer);
    } catch (NullPointerException e) { // NOPMD
        ...... //handle exception 
    }
}

因此下面的分析代码中，如果是调用上面4个对象的subscribe()方法的时候，直接看subscribeActual()方法即可。

那就从最后一层 ObservableSubscribeOn 的 subscribeActual() 方法开始分析。

@Override
public void subscribeActual(final Observer<? super T> s) {
    final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(s);

    s.onSubscribe(parent);

    parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
}

这里的这个s就是我们缩写的observer
第一行：先把我们的observer封装成了SubscribeOnObserver
第二行：调用了observer.onSubscribe()方法
              也就是observer订阅Observable时候的方法，一般这个时候可以做一些操作
第三行：
parent.setDisposable() 以及scheduler调度器先不论，待会再分析，这里先看SubscribeTask这个类：

final class SubscribeTask implements Runnable {
    private final SubscribeOnObserver<T> parent;

    SubscribeTask(SubscribeOnObserver<T> parent) {
        this.parent = parent;
    }

    @Override
    public void run() {
        source.subscribe(parent);
    }
}

看到runnable,熟悉线程的同学已经可以猜到source.subscribe(parent) 这句代码很可能在子线程中执行，这里先mark一下，待会回到这个地方再具体看。

这里要先插入一下source和Observer的问题：

public ObservableCreate(ObservableOnSubscribe<T> source) {
    this.source = source;
}

public ObservableSubscribeOn(ObservableSource<T> source, Scheduler scheduler) {
    super(source);
    this.scheduler = scheduler;
}

public ObservableObserveOn(ObservableSource<T> source, Scheduler scheduler, boolean delayError, int bufferSize) {
    super(source);
    this.scheduler = scheduler;
    this.delayError = delayError;
    this.bufferSize = bufferSize;
}

//后面两个类继承抽象类，调用super(source)方法。
AbstractObservableWithUpstream(ObservableSource<T> source) {
    this.source = source;
}

通过上面的代码可以看到所有的这三个关键类，source都是通过构造传入进来的，
而后两个类都还有schedule参数，这个涉及线程调度，待会也会说，也mark一下。
通过以上代码可以分析：

ObservableCreate 的 source 是 oos
ObservableObserveOn 的 source 是 ObservableCreate
ObservableSubscribeOn 的 source 是 ObservableObserveOn

至于Observer，通过代码可以分析：

 (ObservableObserveOn)source
    .subscribe(parent(SubscribeOnObserver));

 (ObservableCreate)source
    .subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));


 CreateEmitter<T> parent = new CreateEmitter<T>(observer);
 (oos)source.subscribe(parent);

也即是：

ObservableObserveOn 的 observer 是 SubscribeOnObserver
ObservableCreate 的 observer 是 ObserveOnObserver
oos 的 observer 是 CreateEmitter

这里有点一级一级调用的意味了，而这个意味就是Rxjava的一个很重要的点。

插入结束，继续回到刚才的 SubscribeTask
结合上面的分析：

 source.subscribe(parent)

也就意味着

ObservableObserveOn.subscribeActual()

这里转了两个弯，各位可以稍微思考一下
而在ObservableObserveOn中：

@Override
protected void subscribeActual(Observer<? super T> observer) {
    if (scheduler instanceof TrampolineScheduler) {
        //这个暂时忽略，未设置的时候不走这里
        source.subscribe(observer);
    } else {
        //线程调度，待会再分析
        Scheduler.Worker w = scheduler.createWorker();
        //最终会调用这个，又是很熟悉的source subscribe()方法
        source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
    }
}

继续往上走，走到 ObservableCreate 中，这里省略了重复流程。

@Override
protected void subscribeActual(Observer<? super T> observer) {
    CreateEmitter<T> parent = new CreateEmitter<T>(observer);
    observer.onSubscribe(parent);
    try {
        source.subscribe(parent);
    } catch (Throwable ex) {
        Exceptions.throwIfFatal(ex);
        parent.onError(ex);
    }
}

走到最上层，这个最上层的source就是我们前面写的oos。
这个observer是在 ObservableObserverOn 中的 ObserveOnObserver。这个名字有点像，汗

第一行：先把 ObserveOnObserver 封装成 CreateEmitter
第二行：调用 ObserveOnObserver.onSubscribe()方法。

@Override
public void onSubscribe(Disposable s) {
    if (DisposableHelper.validate(this.s, s)) {
        this.s = s;
        if (s instanceof QueueDisposable) {
            @SuppressWarnings("unchecked")
            QueueDisposable<T> qd = (QueueDisposable<T>) s;

            int m = qd.requestFusion(QueueDisposable.ANY | QueueDisposable.BOUNDARY);

            if (m == QueueDisposable.SYNC) {
                sourceMode = m;
                queue = qd;
                done = true;
                actual.onSubscribe(this);
                schedule();
                return;
            }
            if (m == QueueDisposable.ASYNC) {
                sourceMode = m;
                queue = qd;
                actual.onSubscribe(this);
                return;
            }
        }

        queue = new SpscLinkedArrayQueue<T>(bufferSize);

        actual.onSubscribe(this);
    }
}

这个方法比较长，但是对我们的流程分析关键的代码其实就一句

actual.onSubscribe(this);

根据前面的observer的分析，这个observer其实就是 ObservableSubscribeOn 的 SubscribeOnObserver
最后找到源码，调用了 SubscribeOnObserver 的 onSubscribe()方法。

    @Override
    public void onSubscribe(Disposable s) {
        DisposableHelper.setOnce(this.s, s);
    }

    public static boolean setOnce(AtomicReference<Disposable> field, Disposable d) {
        ObjectHelper.requireNonNull(d, "d is null");
        if (!field.compareAndSet(null, d)) {
            d.dispose();
            if (field.get() != DISPOSED) {
                reportDisposableSet();
            }
            return false;
        }
        return true;
    }

涉及到CAS的操作，感兴趣的同学可以研究一下，这里对我们的流程没有太大影响。

第三行：至此，整个的流程终于回到了我们的oos。

从ObservableSubscribeOn的subscribe()方法历尽千辛万苦终于调用了oos的subscribe()方法。

    ObservableOnSubscribe<Integer> oos = new ObservableOnSubscribe<Integer>() {
        @Override
        public void subscribe(ObservableEmitter<Integer> e) throws Exception {
            final int max = 100;
            for (int i = 1; i <= max; i++) {
                e.onNext(max);
            }
            e.onComplete();
        }
    };

首先创建了 ObservableEmitter ，然后调用emmiter.onNext()方法。

    @Override
    public void onNext(T t) {
        if (t == null) {
            onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
            return;
        }
        if (!isDisposed()) {
            observer.onNext(t);
        }
    }

异常先不考虑，最终是调用 observer.onNext()方法。
根据上面的分析，这个Observer是ObserveOnObserver
第一行：先把 ObserveOnObserver 封装成 CreateEmitter，而CreateEmmiter的构造：

    CreateEmitter(Observer<? super T> observer) {
        this.observer = observer;
    }

所以，可以知道这个observer就是最前面的 ObserveOnObserver

也就是e.Next(n)------>最终会调用ObserveOnObserver.onNext(n)

    @Override
    public void onNext(T t) {
        if (done) {
            return;
        }
        if (sourceMode != QueueDisposable.ASYNC) {
            queue.offer(t);
        }
        schedule();
    }

最终调用了 schedule() 方法。

    void schedule() {
        if (getAndIncrement() == 0) {
            worker.schedule(this);
        }
    }

将当前对象添加到worker中，这个是线程调度的问题了，待会分析。

再看一下ObserveOnObserver 类的声明：

static final class ObserveOnObserver<T> extends BasicIntQueueDisposable<T> implements Observer<T>, Runnable{}

实现了Runnable接口，所以关键代码就在run()方法之中。

   @Override
    public void run() {
        if (outputFused) {
            drainFused();
        } else {
            drainNormal();
        }
    }

    //我们看drainNormal()的方法
    void drainNormal() {
        int missed = 1;

        final SimpleQueue<T> q = queue;
        final Observer<? super T> a = actual;

        for (;;) {
            if (checkTerminated(done, q.isEmpty(), a)) {
                return;
            }

            for (;;) {
                boolean d = done;
                T v;

                try {
                    v = q.poll();
                } catch (Throwable ex) {
                    Exceptions.throwIfFatal(ex);
                    s.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;
            }
        }
    }

终于看到了a.onNext()方法，也就是actual.onNext()方法。
通过 ObserveOnObserver 的构造：

//构造方法
ObserveOnObserver(Observer<? super T> actual, Scheduler.Worker worker, boolean delayError, int bufferSize) {
    this.actual = actual;
    this.worker = worker;
    this.delayError = delayError;
    this.bufferSize = bufferSize;
}

//创建对象的时候
@Override
protected void subscribeActual(Observer<? super T> observer) {
    if (scheduler instanceof TrampolineScheduler) {
        source.subscribe(observer);
    } else {
        Scheduler.Worker w = scheduler.createWorker();
        //这里new了ObserveOnObserver对象。
        source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
    }
}

可以看出，这个actual对象，其实就是传入进来的observer。
而这个observer结合SubscribeTask代码，可以知道：
这个observer其实就是讲我们的observer封装起来的SubscribeOnObserver对象。

而SubscribeOnObserver的onNext()方法：

    @Override
    public void onNext(T t) {
        actual.onNext(t);
    }

其实就是我们的o.next()方法。

七转八弯，经历这个这么多,也是本文最核心的：

subscribe()方法，先一层一层往上回调，调用了我们的oos的onNext()方法，
而onNext()里面又一层一层往下回调，调用了我们的obsrever的onNext()方法，实现了数据的传递。

然后是线程切换问题：

还记得我们之前说ObservableSubscribeOn, ObservableObserveOn这两个对象的构造都会传入一个 schedule 的调度器吗？

先看 ObservableSubscribeOn

@Override
public void subscribeActual(final Observer<? super T> s) {
    final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(s);
    s.onSubscribe(parent);
    parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
}

再结合前面的代码，我们知道这个 scheduler 是Schedulers.computation()

最后走到了：

public Disposable scheduleDirect(final Runnable run, long delayTime, TimeUnit unit) {
    ScheduledDirectTask task = new ScheduledDirectTask(RxJavaPlugins.onSchedule(run));
    try {
        Future<?> f;
        if (delayTime <= 0L) {
            f = executor.submit(task);
        } else {
            f = executor.schedule(task, delayTime, unit);
        }
        task.setFuture(f);
        return task;
    } catch (RejectedExecutionException ex) {
        RxJavaPlugins.onError(ex);
        return EmptyDisposable.INSTANCE;
    }
}

executor,我们非常熟悉的线程池。看到这，也就大概明白了我们的 source.subscribe(parent)
以及其对应的一层层往上回调都是在subscribeOn(线程) 所调用的线程之中。

然后线程什么时候会再度切换呢？
是在ObservableObserveOn中的 schedule() 方法中：

    void schedule() {
        if (getAndIncrement() == 0) {
            worker.schedule(this);
        }
    }

这个worker一层层追踪溯源，找到了其初始化的地方，是在ObservableObserveOn的subscribeActual()方法之中：

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));
    }
}

这个schedule就是observerOn所对应的线程。
AndroidSchedulers.mainThread() 的实现是 HandlerScheduler

    public Disposable schedule(Runnable run, long delay, TimeUnit unit) {
        if (run == null) throw new NullPointerException("run == null");
        if (unit == null) throw new NullPointerException("unit == null");

        if (disposed) {
            return Disposables.disposed();
        }

        run = RxJavaPlugins.onSchedule(run);

        ScheduledRunnable scheduled = new ScheduledRunnable(handler, run);

        Message message = Message.obtain(handler, scheduled);
        message.obj = this; // Used as token for batch disposal of this worker's runnables.

        if (async) {
            message.setAsynchronous(true);
        }

        handler.sendMessageDelayed(message, unit.toMillis(delay));

        // Re-check disposed state for removing in case we were racing a call to dispose().
        if (disposed) {
            handler.removeCallbacks(scheduled);
            return Disposables.disposed();
        }

        return scheduled;
    }

最终通过hanler进行了线程的切换。
也就是最后我们的observer.onNext()方法执行的线程是由observeOn()所对应的线程