RxJava 触发流基本原理源码解析
itbird01 人气:0正文
本节，我们从Rxjava使用代码入手，去结合自己已有的知识体系，加查阅部分源码验证的方式，来一起探索一下Rxjava实现的基本原理。
为了本文原理分析环节，可以被更多的人理解、学习，所以小编从初学者的角度，从使用入手，一点点的分析了其中的源码细节、思想，建议大家随着本文的章节步骤，一步一步的来阅读，才能更快、更好的理解Rxjava的真正的思想精髓，也为我们之后的实践课程留一个好的底子。
触发流
到目前为止，我们讲了构建流、订阅流，但是依然没有触发真正的observer中的事件，例如：
@Override
public void onSubscribe(@NonNull Disposable d) {
Log.d(TAG, "onSubscribe");
}
@Override
public void onNext(@NonNull String s) {
Log.d(TAG, "onNext s = " + s);
}
@Override
public void onError(@NonNull Throwable e) {
Log.d(TAG, "onError");
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete");
}
各位看官，莫急莫急，且听老衲娓娓道来。
还记得上面的订阅流吗？订阅流从右往左执行的，执行到最后的observable，执行了它的subscribe方法。我们从使用代码知道，最左端的observable是啥来着，大家还记得吗？当然是ObservableJust
private void test() {
//第一步：just调用
Observable.just("https://img-blog.csdn.net/20160903083319668")
//第二步：map调用
.map(new Function<String, Bitmap>() {
@Override
public Bitmap apply(String s) throws Exception {
//Bitmap bitmap = downloadImage(s);
return null;
}
})
//第三步：subscribeOn、observeOn调用
.subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
//第四步：subscribe调用
.subscribe(new Observer<Bitmap>() {
@Override
public void onSubscribe() {
Log.d(TAG, "onSubscribe");
}
@Override
public void onNext(Bitmap s) {
Log.d(TAG, "onNext s = " + s);
}
@Override
public void onError(Throwable e) {
Log.e(TAG, "onError ", e);
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete");
}
});
}
我们就顺坡下驴，看一下ObservableJust的subscribe方法做啥了
public final class ObservableJust<T> extends Observable<T> implements ScalarSupplier<T> {
private final T value;
public ObservableJust(final T value) {
this.value = value;
}
@Override
protected void subscribeActual(Observer<? super T> observer) {
ScalarDisposable<T> sd = new ScalarDisposable<>(observer, value);
observer.onSubscribe(sd);
sd.run();
}
@Override
public T get() {
return value;
}
}
仔细一看，这里面没有subscribe方法，那么肯定就是调用父类observable的subscribe方法了
@SchedulerSupport(SchedulerSupport.NONE)
@Override
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方法，所以我们继续看子类ObservableJust的subscribeActual方法干啥了？
@Override
protected void subscribeActual(Observer<? super T> observer) {
ScalarDisposable<T> sd = new ScalarDisposable<>(observer, value);
observer.onSubscribe(sd);
sd.run();
}
接续根据ScalarDisposable的run方法
public static final class ScalarDisposable<T>
extends AtomicInteger
implements QueueDisposable<T>, Runnable {
private static final long serialVersionUID = 3880992722410194083L;
final Observer<? super T> observer;
final T value;
//...省略很多代码
@Override
public void run() {
if (get() == START && compareAndSet(START, ON_NEXT)) {
//可以看到这里执行了onNext、onComplete方法
observer.onNext(value);
if (get() == ON_NEXT) {
lazySet(ON_COMPLETE);
observer.onComplete();
}
}
}
}
小结
看到这里，我们知道了，开始一层一层的从左往右去调用observer的相关方法了。 由订阅流可知，每层的observable实际上拥有下一层的observer的代理类，所以自然而然，从最左边开始调用observer的相关方法开始，触发流，就是从左往右，一层一层的剥开之前包裹的observer，然后顺序调用里面的onNext、onComplete等方法。 不信，我们挑一个ObservableMap来验证一下。
public final class ObservableMap<T, U> extends AbstractObservableWithUpstream<T, U> {
final Function<? super T, ? extends U> function;
public ObservableMap(ObservableSource<T> source, Function<? super T, ? extends U> function) {
super(source);
this.function = function;
}
@Override
public void subscribeActual(Observer<? super U> t) {
source.subscribe(new MapObserver<T, U>(t, function));
}
static final class MapObserver<T, U> extends BasicFuseableObserver<T, U> {
final Function<? super T, ? extends U> mapper;
MapObserver(Observer<? super U> actual, Function<? super T, ? extends U> mapper) {
super(actual);
this.mapper = mapper;
}
@Override
public void onNext(T t) {
if (done) {
return;
}
if (sourceMode != NONE) {
downstream.onNext(null);
return;
}
U v;
try {
v = Objects.requireNonNull(mapper.apply(t), "The mapper function returned a null value.");
} catch (Throwable ex) {
fail(ex);
return;
}
//此处调用了下游的observer的onNext方法
downstream.onNext(v);
}
}
}
可以看到里面，的确调用了下游的observer的onNext方法。
总结
整个过程，分为构建流、订阅流、触发流。
构建流：从左到右执行不同的操作符的过程，其实很简单，就是根据不同的操作符，对原始的 observable进行逐层包装，这里可以看出，每层的节点 N* 就持有了上一层的observable。
订阅流：从右到左的 subscribe 调用过程，这个过程中，每个observable内部的subscribeActual执行两个关键操作，一个是对自己已有的observer进行一层重新包装，另外一个就是使用前面节点的observable，订阅包装好的observer。
触发流：在订阅流执行完成之后，执行到最左端的observable，我们发现它内部的subscribeActual实现，实际上就是调用里面拥有的observer的相关回调方法（onNext、onComplete、onError等），那么这层回调流就简单了，就是一层一层的调用里面的observer，最终执行到最右端的observer。
篇幅所限，大家也发现了，我们本节课，我们详细讲解Rxjava线程切换的实现原理，这个有两个原因，一是篇幅所限，本节内容已经够多了，大家先吃透框架，另外一方面是，线程切换我相信我们后面实践环节，待框架自我搭建实现之后，里面的线程切换功能就是水到渠成的事情，相信凭借大家已有的知识，都可以做到的。
所以建议大家，先别看这块Rxjava是如何实现线程切换的，而是想一下，它是怎么实现的？到时我们自己的Rxjava框架搭建起来之后，填充实现一下。
提个醒儿，大家还记得我们之前EventBus源码分析、实践环节吗？其中也说到了线程切换。其实原理差不多，大家先想一下。
加载全部内容