注：本文分析的是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的停止是一样的道理
