RxJava
RxJava – Reactive Extensions for the JVM – a library for composing asynchronous and event-based programs using observable sequences for the Java VM.
RxJava 是java的响应式编程的扩展.
Ps:RxJava 配 Lambda 更合适哦!~
不过这里没用Lambda,本文只关心RxJava语法,因为现在是学好基础语法,打好基础的时候,如果你还不知道Lambda的可以看看我的使用Lambda在Android中的替代匿名类
依赖:compile 'io.reactivex:rxjava:1.0.12'
基础知识
RxJava最核心的两个东西:
- Observables（被观察者，事件源)
- Subscribers（观察者,Observer 的实现类）。
Observables发出一系列事件，Subscribers处理这些事件。
这里的事件可以是任何你感兴趣的东西 （触摸事件，web接口调用返回的数据。。。）
一个Observable可以发出零个或者多个事件，直到结束或者出错。每发出一个事件，就会调用它的Subscriber的onNext方法，最后调用Subscriber.onCompleted()或者Subscriber.onError()结束。
粗看跟观察者模式很相似,但是有一点明显不同，那就是:
如果一个Observerble没有任何的的Subscriber，那么这个Observable是不会发出任何事件的。
Talk is cheap ,show me the code.
接下去练习一下~
练习
1. 创建Observable
通过Observable.create方法很容易创建
static Observable<String> mStringObservable = Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("Hello Rxjava");
//Notice 如果onCompleted 先调用了,那么后面的onNext onError则会被无视掉,但是不是等同于return,后面其他的代码还是会执行
subscriber.onCompleted();
System.out.println("看看onCompleted调用了还会不会有输出");
// subscriber.onError(new Throwable("hehehe"));
}
});
2. 创建Subscriber
接着我们创建一个Subscriber来处理Observable对象发出的字符串。
很简单,只是打印一下数据而已.
static Subscriber<String> mStringSubscriber = new Subscriber<String>() {
@Override
public void onCompleted() {
System.out.println("onCompleted");
}
@Override
public void onError(Throwable e) {
System.out.println("onError:" + e.getMessage());
}
@Override
public void onNext(String s) {
System.out.println(s);
}
};
3. 订阅
main方法添加代码,run起来~
public static void main(String[] args) {
mStringObservable.subscribe(mStringSubscriber);
}
输出Log:
Hello Rxjava
onCompleted
看看onCompleted调用了还会不会有输出
注意点:
- 如果调用了
subscriber.onCompleted(),那么在onCompleted后调用的onNext和onError都不会生效.
其实如果Rx真的每次都需要这么写的话,真的是比Java还啰嗦,接下去练习更简洁的方法.
简化代码
Observable.just方法可以用来创建只发出一个事件就结束的Observable对象
上面创建Observable对象的代码可以简化为一行
static Observable<String> simpleObservable = Observable.just("Hell,simple observable");
跟踪一下just方法的代码到ScalarSynchronousObservable类,会发现,其实它就是创建了一个OnSubscribe对象,顺便帮忙调用了onNext和onCompleted:
protected ScalarSynchronousObservable(final T t) {
super(new OnSubscribe<T>() {
@Override
public void call(Subscriber<? super T> s) {
s.onNext(t);
s.onCompleted();
}
});
this.t = t;
}
同样,订阅一下试试~
simpleObservable.subscribe(mStringSubscriber);
输出Log:
Hell,simple observable
onCompleted
更进一步
RxJava中还有Action1和Action0这两个类.
另外
subscribe方法有一个重载版本，接受三个Action类型的参数，分别对应OnNext，OnComplete， OnError函数。
subscribe(onNextAction, onErrorAction, onCompleteAction);
如果我们并不关心OnComplete和OnError,那么完全可以用Action1来替代上面所写:
static Action1<String> onNextAction = new Action1<String>() {
@Override
public void call(String s) {
System.out.println("Action1<String> onNextAction:" + s);
}
};
在main中增加:
mStringObservable.subscribe(onNextAction);
run起来~
Log:
Action1<String> onNextAction:Hello Rxjava
那么再写一下onComplete和onError看:
// 注意是:Throwable
static Action1<Throwable> onErrorAction = new Action1<Throwable>() {
@Override
public void call(Throwable throwable) {
System.out.println("Action1<Throwable> onErrorAction:");
}
};
//Action0
static Action0 onCompleteAction = new Action0() {
@Override
public void call() {
System.out.println("Action0 onCompleteAction:");
}
};
//Notice 如果调用onError 这个单个的onNextAction 会报错
mStringObservable.subscribe(onNextAction);
mStringObservable.subscribe(onNextAction, onErrorAction);
mStringObservable.subscribe(onNextAction, onErrorAction, onCompleteAction);
需要注意的是onErrorAction 接收的是Throwable,onComplete无类型
另外如果call里面调用了onError则会报错:
Exception in thread "main" rx.exceptions.OnErrorNotImplementedException: hehehe
再简化,单个Action1可以简化成:
Observable.just("Hello simple onaction1").subscribe(new Action1<String>() {
@Override
public void call(String s) {
System.out.println(s);
}
});
输出日志:
Hello simple onaction1
好了,基本语法部分例子学完了,例子很简单,到此结束,下一篇Rx操作符.
更多更快更好的阅读体验请戳我的博客
参考/更多
