标签:ror call() nim getname etl let ber pointer sage
1.什么是RxJava3.引入框架
目前RxJava已经升级为2.0版本,但为了能够更好的理解RxJava,我们可以从1.0版本开始学习。也为了让我们的安卓项目能够更好的使用RxJava,可以在项目中引入gradle脚本依赖:
compile ‘io.reactivex:rxandroid:1.2.1‘
compile ‘io.reactivex:rxjava:1.1.6‘
现在 我们的项目已经支持RxJava的功能了。
4.响应式的核心
所谓的响应式,无非就是存在这样的2个部分,一部分负责发送事件/消息,另一部分负责响应事件/消息。
以前如果我们想看新闻,一般需要通过看报纸。比如,你对某个报刊杂志比较感兴趣,那么你首先要做3件事:
订阅(也就是 观察者&被观察者之间要相互关联 以便被观察的对象一变化 就会马上通知观察该事件的对象)
上面示例的演示代码如下:
//1.创建被观察者
Observable<String> observable =
Observable.create(new Observable.OnSubscribe<String>() {br/>@Override
public void call(Subscriber<? super String> subscriber) {
//4.开始发送事件
//事件有3个类型 分别是onNext() onCompleted() onError()
//onCompleted() onError() 一般都是用来通知观察者 事件发送完毕了,两者只取其一。
subscriber.onNext("Hello Android !");
subscriber.onNext("Hello Java !");
subscriber.onNext("Hello C !");
subscriber.onCompleted();
}
});
//2.创建观察者
Subscriber<String> subscriber = new Subscriber<String>() {br/>@Override
public void onCompleted() {
Log.i(TAG, "onCompleted ");
}
@Override
public void onError(Throwable e) {
Log.i(TAG, "onError: "+e.getLocalizedMessage());
}
@Override
public void onNext(String s) {
Log.i(TAG, "onNext: "+s);
}
};
//3.订阅
observable.subscribe(subscriber);
输出如下:
com.m520it.rxjava I/IT520: onNext: Hello Android !
com.m520it.rxjava I/IT520: onNext: Hello Java !
com.m520it.rxjava I/IT520: onNext: Hello C !
com.m520it.rxjava I/IT520: onCompleted
代码运行的原理
? 上面的代码中,当观察者subscriber订阅了被观察者observable之后,系统会自动回调observable对象内部的call()。
? 在observable的call()方法实体中,发送了如onNext/onCompleted/onError事件后。
? 接着subscriber就能回调到到对应的方法。
5.被观察者变种
普通的Observable发送需要三个方法onNext, onError, onCompleted,而Single作为Observable的变种,只需要两个方法:
? onSuccess - Single发射单个的值到这个方法
? onError - 如果无法发射需要的值,Single发射一个Throwable对象到这个方法
Single只会调用这两个方法中的一个,而且只会调用一次,调用了任何一个方法之后,订阅关系终止。
final Single<String> single = Single.create(new Single.OnSubscribe<String>() {br/>@Override
public void call(SingleSubscriber<? super String> singleSubscriber) {
//先调用onNext() 最后调用onCompleted()
//singleSubscriber.onSuccess("Hello Android !");
//只调用onError();
singleSubscriber.onError(new NullPointerException("mock Exception !"));
}
});
Observer<String> observer = new Observer<String>() {br/>@Override
public void onCompleted() {
Log.i(TAG, "onCompleted ");
}
@Override
public void onError(Throwable e) {
Log.i(TAG, "onError: "+e.getLocalizedMessage());
}
@Override
public void onNext(String s) {
Log.i(TAG, "onNext: "+s);
}
};
single.subscribe(observer);
6.观察者变种
Observer观察者对象,上面我们用Subscriber对象代替。因为该对象本身就是继承了Observer。
该对象实现了onNext()&onCompleted()&onError()事件,我们如果对哪个事件比较关心,只需要实现对应的方法即可,代码如下:
//创建观察者
Subscriber<String> subscriber = new Subscriber<String>() {br/>@Override
public void onCompleted() {
Log.i(TAG, "onCompleted ");
}
@Override
public void onError(Throwable e) {
Log.i(TAG, "onError: "+e.getLocalizedMessage());
}
@Override
public void onNext(String s) {
Log.i(TAG, "onNext: "+s);
}
};
//订阅
observable.subscribe(subscriber);
上面的代码中,如果你只关心onNext()事件,但却不得不实现onCompleted()&onError()事件.这样的代码就显得很臃肿。鉴于这种需求,RxJava框架在订阅方面做了特定的调整,代码如下:
//为指定的onNext事件创建独立的接口
Action1<String> onNextAction = new Action1<String>() {br/>@Override
public void call(String s) {
Log.i(TAG, "call: "+s);
}
};
//订阅
observable.subscribe(onNextAction);
不知道大家注意到没有,subscribe()订阅的不再是观察者,而是特定的onNext接口对象。类似的函数如下,我们可以根据需要实现对应的订阅:
public Subscription subscribe(final Observer observer)
public Subscription subscribe(final Action1 onNext)
public Subscription subscribe(final Action1 onNext, Action1 onError)
public Subscription subscribe(final Action1 onNext, Action1 onError, Action0 onCompleted)
这里还有一个forEach函数有类似的功能:
public void forEach(final Action1 onNext)
public void forEach(final Action1 onNext, Action1 onError)
public void forEach(final Action1 onNext, Action1 onError, Action0 onComplete)
##7.Subject变种
上面2节中既介绍了被观察者变种,又介绍了观察者变种,这里再介绍一种雌雄同体的对象(既作为被观察者使用,也可以作为观察者)。
针对不同的场景一共有四种类型的Subject。他们并不是在所有的实现中全部都存在。
###AsyncSubject
一个AsyncSubject只在原始Observable完成后,发射来自原始Observable的最后一个值。它会把这最后一个值发射给任何后续的观察者。
以下贴出代码:
//创建被观察者final AsyncSubject<String> subject = AsyncSubject.create();//创建观察者
Subscriber<String> subscriber = new Subscriber<String>() {br/>@Override
public void onCompleted() {
Log.i(TAG, "onCompleted");
}
@Override
public void onError(Throwable e) {
Log.i(TAG, "onError");
}
@Override
public void onNext(String s) {
Log.i(TAG, "s:" + s);
}
};//订阅事件
subject.subscribe(subscriber);//被观察者发出事件 如果调用onCompleted(),onNext()则会打印最后一个事件;如果没有,onNext()则不打印任何事件。
subject.onNext("Hello Android ");
subject.onNext("Hello Java ");
subject.onCompleted();
输出:
s:Hello Java onCompleted
然而,如果原始的Observable因为发生了错误而终止,AsyncSubject将不会发射任何数据,只是简单的向前传递这个错误通知。
上面的观察者被观察者代码相同,现在发出一系列信号,并在最后发出异常 代码如下:
subject.onNext("Hello Android ");
subject.onNext("Hello Java ");//因为发送了异常 所以onNext()无法被打印
subject.onError(null);
###BehaviorSubject
当观察者订阅BehaviorSubject时,他会将订阅前最后一次发送的事件和订阅后的所有发送事件都打印出来,如果订阅前无发送事件,则会默认接收构造器create(T)里面的对象和订阅后的所有事件,代码如下:
BehaviorSubject subject=BehaviorSubject.create("NROMAL");
Subscriber subscriber = new Subscriber() {br/>@Override
public void onCompleted() {
Log.i(TAG, "onCompleted");
}
@Override
public void onError(Throwable e) {
Log.i(TAG, "onError");
}
@Override
public void onNext(Object o) {
Log.i(TAG, "onNext: " + o);
}
};
//subject.onNext("Hello Android !");//subject.onNext("Hello Java !");//subject.onNext("Hello C !");//这里开始订阅 如果上面的3个注释没去掉,则Hello C的事件和订阅后面的事件生效//如果上面的三个注释去掉 则打印构造器NORMAL事件生效后和订阅后面的事件生效
subject.subscribe(subscriber);
subject.onNext("Hello CPP !");
subject.onNext("Hello IOS !");
PublishSubject
PublishSubject只会把在订阅发生的时间点之后来自原始Observable的数据发射给观察者。
需要注意的是,PublishSubject可能会一创建完成就立刻开始发射数据,因此这里有一个风险:在Subject被创建后到有观察者订阅它之前这个时间段内,一个或多个数据可能会丢失。
代码如下:
PublishSubject subject= PublishSubject.create();
Action1<String> onNextAction1 = new Action1<String>(){
@Override
public void call(String s) {
Log.i(TAG, "onNextAction1 call: "+s);
}
};
Action1<String> onNextAction2 = new Action1<String>(){
@Override
public void call(String s) {
Log.i(TAG, "onNextAction2 call: "+s);
}
};
subject.onNext("Hello Android !");
subject.subscribe(onNextAction1);
subject.onNext("Hello Java !");
subject.subscribe(onNextAction2);
subject.onNext("Hello IOS !");
输出如下:
onNextAction1 call: Hello Java !
onNextAction1 call: Hello IOS !
onNextAction2 call: Hello IOS !
ReplaySubject
ReplaySubject会发射所有来自原始Observable的数据给观察者,无论它们是何时订阅的。
代码如下:
ReplaySubject subject= ReplaySubject.create();
Action1<String> onNextAction1 = new Action1<String>(){
@Override
public void call(String s) {
Log.i(TAG, "onNextAction1 call: "+s);
}
};
Action1<String> onNextAction2 = new Action1<String>(){
@Override
public void call(String s) {
Log.i(TAG, "onNextAction2 call: "+s);
}
};
subject.onNext("Hello Android !");
subject.subscribe(onNextAction1);
subject.onNext("Hello Java !");
subject.subscribe(onNextAction2);
subject.onNext("Hello IOS !");
输出如下:
onNextAction1 call: Hello Android !
onNextAction1 call: Hello Java !
onNextAction2 call: Hello Android !
onNextAction2 call: Hello Java !
onNextAction1 call: Hello IOS !
onNextAction2 call: Hello IOS !
###Subject总结
AsyncSubject无论何时订阅 只会接收最后一次onNext()事件,如果最后出现异常,则不会打印任何onNext()
BehaviorSubject会从订阅前最后一次oNext()开始打印直至结束。如果订阅前无调用onNext(),则调用默认creat(T)传入的对象。如果异常后才调用,则不打印onNext()
PublishSubject只会打印订阅后的任何事件。
ReplaySubject无论订阅在何时都会调用发送的事件。
以上用代码演示了 RxJava 一些基础功能是如何实现的,希望能给大家带来不一样的启发。但这只是一个小小的分享,离真正能运用于工程的 Rx 框架还差太远。这也让我们明白到,一个健壮的框架,需要考虑太多东西,比如代码的可拓展性和可读性,性能优化,可测试性,兼容性,极端情况等等。但有时要想深入理解一个复杂框架的实现原理,就需要剥离这些细节代码,多关注主干的调用逻辑,化繁为简。
标签:ror call() nim getname etl let ber pointer sage
原文地址:https://blog.51cto.com/13007966/2456664