引言
简单阐述RxJava流程源码,RxJava有以下三种流程,向下递增。
- Observable->Observer
- Observable->Operator->Observer
- Observable->Operator->Scheduler->Observer
简单流程
先看一下RxJava简单的代码:
Observable.create(new Observable.OnSubscribe<Boolean>() {
@Override
public void call(Subscriber<? super Boolean> s) {
s.onNext(true);
s.onCompleted();
}
}).subscribe(new Observer<Boolean>() {
@Override
public void onCompleted() {
_log("On complete");
_progress.setVisibility(View.INVISIBLE);
}
@Override
public void onError(Throwable e) {
Timber.e(e, "Error in RxJava Demo concurrency");
_log(String.format("Boo! Error %s", e.getMessage()));
_progress.setVisibility(View.INVISIBLE);
}
@Override
public void onNext(Boolean bool) {
_log(String.format("onNext with return value \"%b\"", bool));
}
});
据上:Obervable create传入Observable.OnSubscribe作为参数。OnSubscribe 会被存储在返回的 Observable 对象中，它的作用相当于一个计划表，当Observable被订阅的时候，即当Obervable.subscribe(observer)的时候,OnSubscribe 的 call() 方法会自动被调用，事件序列就会依照设定依次触发。
看下subscribe method 源码
将源码简化一下就是这样:
public Subscription subscribe(Subscriber subscriber) {
subscriber.onStart();//可选的准备方法
onSubscribe.call(subscriber);//开始执行计划表
return subscriber;//返回Subscription 方便unsubscribe()
}
你会发现触发Observable的时间是subscribe产生订阅关系的时候。
另外你会看到subscribe中参数是Subscriber类型,Subscriber是实现Oberver的抽象类。
带有Operator的流程
这里我们说的Operator是一种抽象的概念,上述简单流程中我们是手写计划表OnSubscribe,而这里我们不需要,我们只需要调用just,map,等转换操作,这些操作内部帮我们实现了OnSubscribe计划表,我将这些操作称之为Operator.
我们看一下带有Operator的简单代码:
Observable.just(true).map(new Func1<Boolean, Boolean>() {
@Override
public Boolean call(Boolean aBoolean) {
_log("Within Observable");
_doSomeLongOperation_thatBlocksCurrentThread();
return aBoolean;
}
}).subscribe(new Observer<Boolean>() {
@Override
public void onCompleted() {
_log("On complete");
_progress.setVisibility(View.INVISIBLE);
}
@Override
public void onError(Throwable e) {
Timber.e(e, "Error in RxJava Demo concurrency");
_log(String.format("Boo! Error %s", e.getMessage()));
_progress.setVisibility(View.INVISIBLE);
}
@Override
public void onNext(Boolean bool) {
_log(String.format("onNext with return value \"%b\"", bool));
}
});
据上:和第一种简单流程的代码相比,subscribe之后的代码是一致的,不同的是subscribe之前的代码。两个方法just和map method,通过源码分析下功能。
- just method
just function:将传入的参数发送给Subscriber订阅者。
你可以根据上面ScalarSynchronousObservable的构造函数,重写OnSubscribe计划表看出来。
- map method
将上面代码简化就是:
public <R> Observable<R> lift(Operator<? extends R, ? super T> operator) {
return Observable.create(new OnSubscribe<R>() {
@Override
public void call(Subscriber subscriber) {
Subscriber newSubscriber = operator.call(subscriber);//通过Operator创建一个新的Subscriber,通过OperatorMap将新旧Subscriber建立一种关系。
newSubscriber.onStart();//准备操作
onSubscribe.call(newSubscriber);//onSubscribe计划表唤醒新的Subscriber,新的Subscriber会联系subscribe订阅的Subscriber(订阅者)。
}
});
}
借助抛物线文章中的图抽象表现出来:
再带上Scheduler的流程
Scheduler function:指明Observable和Observer是运行在哪个线程中。
- subscribeOn指定的是Observable(被观察者)所在线程
- observeOn指定的是Observer(观察者)所在线程
我们在带有Operator的基础上加上Scheduler.
Observable.just(true).map(new Func1<Boolean, Boolean>() {
@Override
public Boolean call(Boolean aBoolean) {
_log("Within Observable");
_doSomeLongOperation_thatBlocksCurrentThread();
return aBoolean;
}
}).subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread()).subscribe(...)
observeOn与subscribeOn也是通过lift变换原理实现线程切换, observeOn切换线程是发生在lift方式中的内建的Subscriber中,subscribeOn切换线程是发生在OnSubscribe计划表中。
调度器源码我也没太搞明白,以后有机会补上。不过调度器作为RxJava的一大特性,在Android编程中使用非常方便。
总结
- RxJava最核心流程是Observable->Oberver 当subscribe订阅的时候,不管增加怎样复杂的变化,一定是OnSubscribe计划表来通知观察者,被观察者发生变化。
- RxJava灵活的特性主要在两个方面,数据序列的变化机制和线程切换。
