基础

注：本文分析的是RxJava 2.0.0 阅读之前，希望你对RxJava1已经有所了解。
首先了解几个基本接口

public interface Emitter<T> 
    void onNext(T value);
    void onError(Throwable error);
    void onComplete();

public interface ObservableEmitter<T> extends Emitter<T>
    void setDisposable(Disposable d);
    void setCancellable(Cancellable c);
    boolean isDisposed();
    ObservableEmitter<T> serialize();

public interface Disposable {
    void dispose();
    boolean isDisposed();

和第一版的Subscriber挺像的，把Subscriber拆分出了几部分
再看一个接口

public interface Observer<T>
    void onSubscribe(Disposable d);
    void onNext(T value);
    void onError(Throwable e);
    void onComplete();

其中onSubscribe和第一版Subscriber的onStart方法类似
第二版多了个Disposable参数，让你可以取消。
我们可以写这个接口的子类，订阅的时候使用

作为开始，先写一个超级简单的例子：

Observable
       .create(new ObservableOnSubscribe<String>() {
      @Override public void subscribe(ObservableEmitter<String> e) throws Exception {
        e.onNext("1"); } })
        .subscribe(new Observer<String>() {
          @Override public void onSubscribe(Disposable d) {}
          @Override public void onNext(String value) { }
          @Override public void onError(Throwable e) {}
          @Override public void onComplete() {}
        });

（排版有点糟糕，先将就一下～～）
create方法接受一个ObservableOnSubscribe对象
ObservableOnSubscribe是一个接口，只有一个方法

public interface ObservableOnSubscribe<T> 
    void subscribe(ObservableEmitter<T> e) throws Exception;

ObservableEmitter这个接口在开始的时候提到过
看看create的源码：

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

先判空
RxJavaPlugins.onAssembly是一个hook，类似于监听，不设置的话传入什么就返回什么
所以这里返回的是ObservableCreate对象
ObservableCreate继承了Observable，Observable是一个基类，也是一个抽象类
Observable规定了被观察者的基本流程，具体实现由子类完成
ObservableCreate的构造，仅仅是接口保存起来

现在来看subscribe方法
里面经过一些检查后，会调用subscribeActual方法
这个方法是abstract的
前面我们得到的实际上是ObservableCreate对象，所以这个具体实现在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包装了observer，这个observer就是订阅时传入的observer
然后，调用onSubscribe方法
之后，source就是create的时候传入的接口，终于回到我们的自定义方法了
然而，这个时候，我们和observer还隔着一个CreateEmitter
先看看CreateEmitter部分代码：

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

        @Override
        public void onError(Throwable t) {
            if (t == null) {
                t = new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
            }
            if (!isDisposed()) {
                try {
                    observer.onError(t);
                } finally {
                    dispose();
                }
            } else {
                RxJavaPlugins.onError(t);
            }
        }

        @Override
        public void onComplete() {
            if (!isDisposed()) {
                try {
                    observer.onComplete();
                } finally {
                    dispose();
                }
            }
        }

onNext: 检查发射的值是否为null, observer是否已取消观察， 然后才交给observer
注意，异常需要自己处理
onComplete:检查observer是否已取消观察， 交给observer， 最后调用dispose
onError: 检查异常是否为null, observer是否已取消观察， 然后才交给observer， 最后dispose
注意，要保证onComplete和onError只能调用一次

再来看看取消观察
CreateEmitter继承了AtomicReference
也就是说这个类有一个变量，这个变量用来标志是否已经取消。
调用dispose最终会调用这个方法

public static boolean dispose(AtomicReference<Disposable> field) {
        Disposable current = field.get();
        Disposable d = DISPOSED;
        if (current != d) {
            current = field.getAndSet(d);
            if (current != d) {
                if (current != null) {
                    current.dispose();
                }
                return true;
            }
        }
        return false;
    }

简单来说，就是把这个标志设置为DISPOSED，DISPOSED就是一个单例。

isDispose方法：

public static boolean isDisposed(Disposable d) {
        return d == DISPOSED;
    }

很简单，看这个标志是否等于DISPOSE

回顾一下，在create的时候返回ObservableCreate对象，这个对象保存了传入的接口
subscribe的时候，传入的observer会到达ObservableCreate，进行包装，然后交给create时我们的自定义处理
取消订阅就是设置标志位，本身并不会做些什么，这和Thread的停止是一样的道理