该类里有3个重要方法：
- create
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
ObjectHelper.requireNonNull(source, "source is null");
return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}
- 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;
}
}
- subscribeActual(子类ObservableCreate实现)
protected void subscribeActual(Observer<? super T> observer) {
CreateEmitter<T> parent = new CreateEmitter<T>(observer);
observer.onSubscribe(parent);//注释3
try {
source.subscribe(parent);//注释4
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
parent.onError(ex);
}
}
先来个使用示例：
/**
* 基础使用create()
*/
public void demo1(){
Observable.create(new ObservableOnSubscribe<String>() {//1
@Override
public void subscribe(ObservableEmitter<String> e) throws Exception {
e.onNext("发送消息");
}
}).subscribe(new Observer<String>() {//2
@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方法：
- Observable.create：在调用Observable.create的时候，可以看该方法实现，是把create的参数ObservableOnSubscribe保存到了ObservableCreate实例中的一个类属性中，然而现在只是保存，如果不调用subscribe去订阅的话，后续并不会发生任何事情
- .subscribe：注释1会返回ObservableCreate实例，该实例调用subscribe(new Observer)方法的时候，看源码可知，会调用subscribeActual(observer);方法，这个方法是Observable的抽象方法，而ObservableCreate类继承了Observable并实现了该抽象方法，然后我们看ObservableCreate的该方法，会发现该类会把subscribe中的参数Observer包装到CreateEmitter类里，然后调用observer类的onSubscribe的方法传入CreateEmitter参数，在这里我们就知道为什么observer的onSubscribe方法会在ObservableOnSubscribe的subscribe方法前面调用了。而且我们在这里会发现observer的onSubscribe和ObservableOnSubscribe的subscribe方法里的参数是同一个实例（看注释3和注释4）。
注意：也就是在该方法里，注释1中的参数ObservableOnSubscribe和注释2中的参数Observer发生了联系，注释1中参数ObservableOnSubscribe封装到了ObservableCreate里，注释2中的参数Observer封装到了CreateEmitter里，然后执行注释2的方法的时候（调用注释2的subscribe方法的类是上一句代码产生的Observable, 这里产生的Observable是ObservableCreate），在注释2的方法里会执行ObservableCreate类的subscribeActual方法 - 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.create）生成的Observable中的source类属性保存的是ObservableOnSubscribe类型或者就没有source而是直接发送数据（如Observable.just），而后续生成的Observable对象中的source类属性保存的都是上游Observable。
2.订阅： 因为代码的每一步都会生成新的Observable类，所以在执行Observable的subscribe方法的时候，每个Observable类都会执行source.subscribe(parent)方法，其中的source是该Observable保存的上游的Observable，参数parent是该Observable的内部类，这个内部类封装了下游的Observer，如果有线程切换，都是在这个内部类完成的。订阅的执行顺序是由下到上的。
3.执行：除了数据的起始位置，其他的Observable会先执行自己内部的Observer来处理每一步传入的数据，然后再将处理后的数据交给下游的Observer继续处理。 数据的传递和处理顺序是由上到下的。
