RxJava2源码分析-Observable

该类里有3个重要方法：

1. create

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

2. 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);
        } catch (NullPointerException e) { // NOPMD
            throw e;
        } catch (Throwable e) {
            Exceptions.throwIfFatal(e);
            // can't call onError because no way to know if a Disposable has been set or not
            // can't call onSubscribe because the call might have set a Subscription already
            RxJavaPlugins.onError(e);

            NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");
            npe.initCause(e);
            throw npe;
        }
    }

3. subscribeActual(子类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);
        }
    }

先来个使用示例：

    /**
     * 基础使用create()
     */
    public void demo1(){
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                e.onNext("发送消息");
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                System.out.println("观察者接收到：onSubscribe");
            }

            @Override
            public void onNext(String s) {
                System.out.println("观察者接收到："+s);
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {
                System.out.println("观察者接收到：onComplete");
            }
        });
    }

开始讲解demo1方法：

1. Observable.create：在调用Observable.create的时候，可以看该方法实现，是把create的参数ObservableOnSubscribe又作为参数保存到了ObservableCreate的实例，然而现在只是保存，如果不调用subscribe去订阅的话，后续并不会发生任何事情
2. .subscribe：Observable.create方法会返回ObservableCreate实例，该实例调用subscribe(new Observer)方法的时候，看源码可知，会调用subscribeActual(observer);方法，这个方法是Observable的抽象方法，而ObservableCreate类继承了Observable并实现了该抽象方法，然后我们看ObservableCreate的该方法，会发现会把subscribe中的参数Observer包装到CreateEmitter类里，然后调用observer类的onSubscribe的方法传入CreateEmitter参数，在这里我们就知道为什么observer的onSubscribe方法会在ObservableOnSubscribe的subscribe方法前面调用了。而且我们在这里会发现observer的onSubscribe和ObservableOnSubscribe的subscribe方法里的参数是同一个实例。
   注意：也就是在该方法里，Observable.create的参数ObservableOnSubscribe和.subscribe的参数Observer发生了联系，Observable.create的参数ObservableOnSubscribe封装到了ObservableCreate里，.subscribe的参数Observer封装到了CreateEmitter里，然后执行.subscribe方法的时候（调用.subscribe方法的类是ObservableCreate），在ObservableCreate的.subscribe方法里会执行ObservableCreate类的subscribeActual方法
3. e.onNext("发送消息");：因为e这个实例是ObservableEmitter，我们看下这个类的代码：

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

我们会发现，实际调用的就是CreateEmitter包装的Observer类，而这个Observer是我们在上面第2步骤传入的参数，所以会执行到Observer的onNext方法。

好了，流程就是这样，更复杂的功能我也在学习，欢迎一起学习探讨。我们搞技术不仅要知其然，还要知其所以然。

总结如下：

1.创建：订阅前，每一步都生成对应的Observable对象，中间的每一步都将上游的Observable存储；
2.订阅： 每一步都会生成对应的Observer对上一步生成并存储的Observable进行订阅。订阅的执行顺序是由下到上的。
3.执行：先执行每一步传入的函数操作，然后将操作后的数据交给下游的Observer继续处理。 数据的传递和处理顺序是由上到下的。