在分析之前呢，首先需要大家打开源码对照分析。效果更加！！
首先我们看一下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回调，回调后就执行数据发射操作。
整个过程就这样。
最后，我们来看一下执行的顺序：
整个执行顺序
网友评论