标签（空格分隔）： rxandroid rxjava
RxJava: "a library for composing asynchronous and event-based programs using observable sequences for the Java VM" (一个在Java VM上使用可观测的序列来组成异步的、基于事件的程序的库)
RxJava的观察者模式的基本概念
RxJava有四个基本概念:
Observable(可观察者，即被观察者)
Observer(观察者)
subscribe(订阅)
事件
被观察者Observable中有一张计划表OnSubscribe,当订阅者subcribe发出订阅请求时，被订阅的Observable对象的计划表OnSubscribe的call方法会被调用，事件序列就会依照设定依次触发，订阅者去处理。这样由被观察者调用了观察者的回调方法，就实现了由被观察者向观察者的事件传递，即观察者模式。
一个简单的调用例子
这个例子很简单，甚至在实际中没有什么意义，只是通过这个例子，看下代码的回调执行过程。下面是相关代码：
Observable<String> observable = Observable.create( new Observable.OnSubscribe<String>() {
@Override
public void call( Subscriber<? super String> subscriber ) {
Log.d( TAG, "OnSubscribe call: " );
subscriber.onNext( "123" );
}
} );
observable.subscribe( new Subscriber<String>() {
@Override
public void onCompleted() {
Log.d( TAG, "Subscriber onCompleted: " );
}
@Override
public void onError( Throwable e ) {
Log.d( TAG, "Subscriber onError: " );
}
@Override
public void onNext( String s ) {
Log.d( TAG, "Subscriber onNext: " + s );
}
} );
这段代码很容易，整个过程分为
a.创建一个被观察者Observable对象，同时新建一个OnSubscribe，当这个计划表被触发时，通知相关的订阅者(类似观察者角色)。
b.订阅者subscribe执行订阅，像Observable发出命令，当Observable被订阅时，计划表OnSubscribe的call方法就会执行，事件序列就会依照计划表中的逻辑触发。
下面详细分析
1.Observable.create(OnSubscribe f)
public static <T> Observable<T> create(OnSubscribe<T> f) {
return new Observable<T>(hook.onCreate(f));
}
这里的hook是一个插件管理，可以暂时忽略它，默认返回的是本身，比如hoon.onCreate(f),默认返回的就是f这个对象。在Observable构造方法中，将f赋值给其成员变量onSubscribe，这样一个计划表就完成建立了,同时被观察者对象也被建立。
2.Observable.subscribe(Subscriber subscriber)
public final Subscription subscribe(Subscriber<? super T> subscriber) {
return Observable.subscribe(subscriber, this);
}
static <T> Subscription subscribe(Subscriber<? super T> subscriber, Observable<T> observable) {
// validate and proceed
...
// new Subscriber so onStart it
subscriber.onStart();
// if not already wrapped
if (!(subscriber instanceof SafeSubscriber)) {
// assign to `observer` so we return the protected version
subscriber = new SafeSubscriber<T>(subscriber);
}
try {
// allow the hook to intercept and/or decorate
hook.onSubscribeStart(observable, observable.onSubscribe).call(subscriber);
return hook.onSubscribeReturn(subscriber);
} catch (Throwable e) {
...
if (subscriber.isUnsubscribed()) {
RxJavaPluginUtils.handleException(hook.onSubscribeError(e));
} else {
// if an unhandled error occurs executing the onSubscribe we will propagate it
try {
subscriber.onError(hook.onSubscribeError(e));
} catch (Throwable e2) {
....
throw r;
}
}
return Subscriptions.unsubscribed();
}
}
省去了一些异常检查的代码，进入这个方法之后，参数subscriber就是我们的订阅者，observable就是当前已经建立好计划表的被观察者对象。在这个方法中基本的非空校验通过之后，接下来回调subscriber的onStart方法，在这里事件还没有开始执行可以做一些准备工作。然后将subscriber用一个包装者模式封装成SafeSubscriber对象，这个包装类进行了一些异常逻辑的处理。紧接着调用hook对象的onSubscribeStart方法，传人的参数为被观察者observable和它的计划表onSubscribe对象。
public <T> OnSubscribe<T> onSubscribeStart(Observable<? extends T> observableInstance, final OnSubscribe<T> onSubscribe) {
// pass through by default
return onSubscribe;
}
这个方法的默认返回值为参数中的计划表对象，接下来拿到这个计划表对象onSubscribe之后，主动回调它的call方法，完成一次事件触发，同时将当前订阅者subcriber以参数的形式传入，回到第一步中的计划表对象call实现
@Override
public void call( Subscriber<? super String> subscriber ) {
Log.d( TAG, "OnSubscribe call: " );
subscriber.onNext( "123" );
}
} );
在计划表的call方法中，subercriber.onNext方法即通知订阅者去执行它的下一步。同时在Observable.subscribe还有一些异常处理，一旦某个计划表出现异常，即会回调订阅者的onError方法，告诉订阅者订阅失败。
总结，上述简单例子是RxAndroid的基础，每一个被观察者都对应有一个计划表对象OnSubscribe。下面用图表的形式展示下这段代码的执行过程:
1.Observable.create: 创建一个被观察者对象和一个计划表OnSubscribe对象
2.Observable.subscribe: 创建一个订阅者Subscribe对象,同时触发事件
<img src="http://o767chiwj.bkt.clouddn.com/rxandroid_1-1.png"/>
一次事件的过程为:被观察者Observable对象的计划表ObSubscribe的call方法被调用，同时在call方法中，回调订阅者Subscribe的onNext方法(观察者的一个拓展实现).最后onCompleted和onError方法有且仅有一个被回调。这是只有一层结构的被观察者和订阅者的情形。
网友评论