RxJava 2 源码分析(一)

在分析之前呢，首先需要大家打开源码对照分析。效果更加！！

首先我们看一下RxJava 2 三步曲的一个基本实现：
1 创建被观察者（也叫数据发射者）
2 创建观察者（也叫数据消费者）
3 建立订阅关系

代码如下：

    //第一步 创建被观察者
    Observable<String> observable = Observable.create(
            new ObservableOnSubscribe<String>() {
                @Override
                public void subscribe(ObservableEmitter<String> e) throws Exception {

                    e.onNext("发射数据");
                    e.onComplete();
                }
            }
    );

    //第二步 创建观察者
    Observer<String> observer = new Observer<String>() {

        private Disposable disposable;

        @Override
        public void onSubscribe(Disposable d) {
            disposable = d;
        }

        @Override
        public void onNext(String value) {

        }

        @Override
        public void onError(Throwable e) {

        }

        @Override
        public void onComplete() {

        }
    };

    //第三步 建立订阅关系
    observable.subscribe(observer);

新记：
我们来看一下创建Observable的这个静态方法create.实际上这个是RxJava大量的操作符中的一个，create方法会返回一个Observable实例。
create方法的参数是一个实现了ObservableOnSubscribe接口的对象实例，该接口提供了发射数据的回调subscribe()方法,回调回来的ObservableEmitter实例就可以看成是数据发射器，用来发射数据。

我们来看看Observable.create()的内部实现：

public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
    ObjectHelper.requireNonNull(source, "source is null");
    return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}

首先是创建了一个ObservableCreate实例，是Observable的子类，把之前参数中创建的ObservableOnSubscribe实例直接传了进去，作了件什么事呢？实际上实现了一个代理的作用，代理的是谁？是Observer，后面我们再详细分析，我们可以确定，ObservableCreate就是一个Observable。
我们再看RxJavaPlugins.onAssembly().这个方法拿了ObservableCreate(Observable)实例去做了什么:

public static <T> Observable<T> onAssembly(Observable<T> source) {
    Function<Observable, Observable> f = onObservableAssembly;
    if (f != null) {
        return apply(f, source);
    }
    return source;
}

RxJavaPlugins这个类是一个钩子函数集合类，为RxJava中大量的操作符提供钩子函数的注入。这里的钩子函数会对Observable实例按照钩子函数实际提供的功能进行加工处理，然后返回一个处理过的Observable。实际上在我们上面所写的这个代码示例中，这里的钩子函数为null，并没有对Observable做任何处理就直接返回了。

实际上我们示例代码中整个被观察者Observable的创建，实际创建的是一个ObservableCreate实例，该实例提供了回调方法subscribe()，当发生订阅行为时会回调，也就是示例代码中执行observable.subscribe(),订阅后就可以发射数据了。通过数据发射器ObservableEmitter来进行数据发射。

像这种订阅后才开始发射数据的，我们称为Cold Observable； 另外一种称作Hot Observable,这种是不管有没有观察者来订阅都会不断地发射数据。

我们再来看看observable.subscribe()订阅的内部代码:

  public final void subscribe(Observer<? super T> observer) {
    ObjectHelper.requireNonNull(observer, "observer is null");
    try {
        observer = RxJavaPlugins.onSubscribe(this, observer);

        ObjectHelper.requireNonNull(observer, "Plugin returned null Observer");

        subscribeActual(observer);
    ...//这里把不关心的代码省略掉了
}

首先是代码中的RxJavaPlugins.onSubscribe()的调用与上面讲的钩子方法是一样的，这里是通过钩子方法对observer作了某种处理。示例中实际上也并未调用实际的钩子方法。

最后就执行到了subscribeActual()方法，我们前面讲过我们的Observable是一个ObservableCreate实例，subscribeActual方法在Observable中是一个虚方法，真正的实现是在ObservableCreate中，代码如下：

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

代码中的CreateEmitter是ObservableCreate的内部类，实现了对observer的代理，CreateEmitter同时也实现了Disposable接口，该接口提供了dispose方法，可以用来停止对数据的接收。
我们接下来再看后面的几行代码就很明了了，首先是进行了observer的onSubscribe回调，然后是调用了observable的subscribe回调，回调后就执行数据发射操作。
整个过程就这样。

最后，我们来看一下执行的顺序：

整个执行顺序