我所理解的Rxjava2
最近又在看网易公开课里面的美国名校的毕业演讲，虽然以前看过，但现在重新看的话，依然能学到不少东西，能把某个话题讲的很细，很深，我现在特别喜欢Michelle，作为第一夫人却没有第一夫人的架子，听她的演讲，能感受到那种能量密度特别大，她演讲到人要想成为什么样的人时，我特别震撼，她认为如果想要成为什么样的人的话，必须要有适应力和责任心。而国内的演讲都是讲怎么成功，怎么成才，怎么爱国，且都谈的特别的大，泛泛而谈，我认为，这也是现在的年轻人特别浮躁的原因之一吧。对于我目前而言要做的是，脚踏实地的做事情，好好孝敬父母。
本篇文章主要是明白Rxjava的流程 ，本文基于Rxjava2.2.8 的源码
1.分析案例
分析下经常用的这个操作符到底做了哪些事情
Observable.create(new ObservableOnSubscribe<String>(){
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
emitter.onNext("hellp");
}
}).map(new Function<String, Object>() {
@Override
public Object apply(String s) throws Exception {
return s + " world";
}
}).subscribe(new Observer<Object>() {
@Override
public void onSubscribe(Disposable d) {
Log.d("verse", "onSubscribe");
}
@Override
public void onNext(Object o) {
String str = (String) o;
Log.d("verse", str);
}
@Override
public void onError(Throwable e) {
Log.d("verse", "onError");
}
@Override
public void onComplete() {
Log.d("wangyang", "onComplete");
}
});
上述代码很简单，在此不再多说
2.源码分析create()
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
ObjectHelper.requireNonNull(source, "source is null");
return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}
当调用Observable.create（ObservableOnSubscribe）时，初始化了ObservableCreate对象，并引用了ObservableOnSubscribe对象
3.源码分析map()
public final <R> Observable<R> map(Function<? super T, ? extends R> mapper) {
ObjectHelper.requireNonNull(mapper, "mapper is null");
return RxJavaPlugins.onAssembly(new ObservableMap<T, R>(this, mapper));
}
实际上是ObservableCreate.map(Function)，初始化了ObservableMap对象，并引用了ObservableCreate对象和Function对象
4.源码分析subscribe()
其实调用的是ObservableMap.subscribe()
public final void subscribe(Observer<? super T> observer) {
ObjectHelper.requireNonNull(observer, "observer is null");
try {
observer = RxJavaPlugins.onSubscribe(this, observer);
ObjectHelper.requireNonNull(observer, "The RxJavaPlugins.onSubscribe hook returned a null Observer. Please change the handler provided to RxJavaPlugins.setOnObservableSubscribe for invalid null returns. Further reading: https://github.com/ReactiveX/RxJava/wiki/Plugins");
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(observer);
查看该代码发现是抽象方法，该实现方法是ObservableMap.subscribeActual(observer)
public void subscribeActual(Observer<? super U> t) {
source.subscribe(new MapObserver<T, U>(t, function));
}
将实现的Observer和Function封装在MapObserver中，而source是引用上游的对象
即ObservableCreate.subscribe(Observer)，还是这段代码
public final void subscribe(Observer<? super T> observer) {
ObjectHelper.requireNonNull(observer, "observer is null");
try {
observer = RxJavaPlugins.onSubscribe(this, observer);
ObjectHelper.requireNonNull(observer, "The RxJavaPlugins.onSubscribe hook returned a null Observer. Please change the handler provided to RxJavaPlugins.setOnObservableSubscribe for invalid null returns. Further reading: https://github.com/ReactiveX/RxJava/wiki/Plugins");
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;
}
}
又继续调用ObservableCreate.subscribeActual(observer)
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);
}
}
又将observer封装在CreateEmitter中，并调用observer.onSubscribe(parent)，而这个observer就是MapObserver，查看MapObserver.onSubscribe()
public final void onSubscribe(Disposable d) {
if (DisposableHelper.validate(this.upstream, d)) {
this.upstream = d;
if (d instanceof QueueDisposable) {
this.qd = (QueueDisposable<T>)d;
}
if (beforeDownstream()) {
downstream.onSubscribe(this);
afterDownstream();
}
}
}
而这里的downstream就是我们在subscribe中实现的内部内，看源码知道了原来onSubscribe()方法啥事情也没干，只是调用了该方法，一般在这个方法里做一些初始化的操作，继续往下看，会调用
source.subscribe(parent);
source就是在 create()所实现的类，并传递CreateEmitter，当调用onNext()时，就是调用的CreateEmitter.onNext()，查看该方法
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);
}
}
剔除非关键性代码后，只有一行代码
observer.onNext(t);
而这个Observer是MapObserver，查看MapObserver.onNext()
public void onNext(T t) {
if (done) {
return;
}
if (sourceMode != NONE) {
downstream.onNext(null);
return;
}
U v;
try {
v = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper function returned a null value.");
} catch (Throwable ex) {
fail(ex);
return;
}
downstream.onNext(v);
}
剔除非关键性代码后，只有两行代码
v = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper function returned a null value.");
downstream.onNext(v);
mapper是实现的Function，转化为相应的类型后，会调用
downstream.onNext(v);
而downstrean是在subscribe中所实现的内部内，最后会调用我们实现的匿名内部内Observer.onNext()
自此，该案例分析完毕
5.总结
当调用操作符时，会创建对应的对象，并引用上游的对象，如调用操作符map()时，会创建ObservableMap对象，并持有上游的ObservableCreate对象，这样一环扣着一环，直至调用subscribe时，会将subscribe(Observer)里面的参数和操作符的内部内封装在Observable类里面的静态内部内里面，由于引用了上游的Observable，再调用上游的subscribe的方法，依次类推，直至ObservableCreate.subscribe(Observer)方法里，在这里，会先执行一个onSubscribe()方法，一般是在这里面做一个执行前的操作，再会执行ObservableOnSubscribe.subscribe(ObservableEmitter)方法，在这个方法里可以执行ObservableEmitter.onNext()，而当前的实例是在ObservableCreate的内部内的onNext()方法里实现的，在这个onNext里，又会调用observer.onNext()方法，这里的observer是下游的Observable的内部内，在下游的内部内的onNext()执行需要的操作后又继续传递给下下游的Observable.onNext()，依次类推，直至最后一个onNext()，也就是subscribe(Observer)方法的参数的Observer.onNext()
一图胜千言，我在网上找了Rxjava2的流程图
rxjava2事件流程图
