经过上一篇文章 Android拾萃 - RxJava最简单的入门(一) 的学习,我们对整个流程有了一定的把握。今天我们就来看看操作符。RxJava中提供了大量不同种类,不同场景的Operators(操作符),RxJava的强大性就来自于它所定义的操作符。
- 操作符让你可以变换、组合、操纵和处理Observable发射的数据。
- Rx的操作符让你可以用声明式的风格组合异步操作序列,它拥有回调的所有效率优势,同时又避免了典型的异步系统中嵌套回调的缺点。
主要分类:
RxJava 的操作符 | 说明 | 例如 |
---|---|---|
创建操作 | 用于创建Observable的操作符 | create、defer、from、just、start、repeat、range |
变换操作 | 用于对Observable发射的数据进行变换 | buffer、window、map、flatMap、groupBy、scan |
过滤操作 | 用于从Observable发射的数据中进行选择 | debounce、distinct、filter、sample、skip、take |
组合操作 | 用于将多个Observable组合成一个单一的Observable | and、startwith、join、merge、switch、zip |
异常处理 | 用于从错误通知中恢复 | catch、retry |
辅助操作 | 用于处理Observable的操作符 | delay、do、observeOn、subscribeOn、subscribe |
条件和布尔操作 | all、amb、contains、skipUntil、takeUntil | |
算法和聚合操作 | average、concat、count、max、min、sum、reduce | |
异步操作 | start、toAsync、startFuture、FromAction、FromCallable、runAsync | |
连接操作 | connect、publish、refcount、replay | |
转换操作 | toFuture、toList、toIterable、toMap、toMultiMap | |
阻塞操作 | forEach、first、last、mostRecent、next、single | |
字符串操作 | byLine、decode、encode、from、join、split、stringConcat |
其中有一些高频使用的操作符如下:
常用操作符 | 说明 |
---|---|
interval | 延时几秒,每隔几秒开始执行 |
take | 超过多少秒停止执行 |
map | 类型转换 |
observeOn | 在主线程运行 |
doOnSubscribe | 在执行的过程中 |
subscribe | 订阅 |
RxJava 2.x的5种响应类型。
上游响应类型可以简单理解为被观察者的回调函数,下游消费类型就是观察者,这样上面的图就很清晰了。
在RxJava 2.x中,这5种类型的用法是非常相似的,它们的接口命名规则相同,只要你知道其中一种,就知道其他几种类型该如何在上游发射数据和在下游消费数据。create
接收的类型都为xxxOnSubscrible(xxx为5种类型对应的名字),发射器为xxxEmitter,具体如下表:
RxJava 2.x 类型 | create参数(响应接口) | 发射器 | Observer |
---|---|---|---|
Observable | ObservableOnSubscribe | ObservableEmitter | Observer |
Flowable | FlowableOnSubscribe | FlowableEmitter | FlowableSubscriber |
Single | SingleOnSubscribe | SingleEmitter | SingleObserver |
Completable | CompletableOnSubscribe | CompletableEmitter | CompletableObserver |
Maybe | MaybeOnSubscribe | MaybeEmitter | MaybeObserver |
ObservableOnSubscribe: Observable 对应的上游类型是ObservableOnSubscribe 我们需要发射数据,所以实现了ObservableEmitter方法
ObservableEmitter: ObservableEmitter可以理解为发射器,这个就是用来发出事件的,它可以发出三种类型的事件。
Observer: 观察者订阅者
Consumer : 消费者 如果我们并不关心OnComplete和OnError,我们只需要在onNext的时候做一些处理,这时候就可以使用Consumer类。所以我们可以认为Consumer是一个不完整的Observer(其实和rxjava1.0的action一样,但是rxjava2.0开始,actiOn表示的无参数返回,具体见下面对比)
Single/SingleObserver: 是只发射一次,一次性的
Completable/CompletableObserver:只关注完成状态的
Maybe/MaybeObserver:可以说是前两者的复合体
Action 在Rx1.0是从Action0,Action1...往后排(数字代表可接受的参数),Rx2.0做出了改动
Rx1.0-----------Rx2.0
Action0--------Action(无参的onComplete())
Action1--------Consumer(单个参数的onError(Throwable t)和onNext(T t))
Action2--------BiConsumer
后面的Action都去掉了,只保留了ActionN
或者理解为 无参数的Action0 被操作符io.reactivex.functions.Action和Scheduler代替。Action1被重命名为Consumer。Action2 被重命名为BiConsumer。 ActionN 被Consumer<Object[]> 代替。
Observable使用场景:
- 数据流最长不超过1000个元素,即随着时间的流逝,应用没有机会报OOME(OutOfMemoryError)错误。
- 处理诸如鼠标移动或触摸事件之类的GUI事件
Flowable使用场景:
- 处理超过10K+ 的元素流
- 从磁盘读取(解析文件)
- 从数据库读取数据
- 从网络获取数据流
在onSubscribe/onStart中调用request
注意,在Subscriber.onSubscribe或ResourceSubscriber.onStart中调用request去请求资源,参数就是要请求的数量,一般如果不限制请求数量,可以写成Long.MAX_VALUE。如果你不调用request,Subscriber的onNext和onComplete方法将不会被调用。当调用request(n)时,将会立即调用onNext。
实例代码如下:
Flowable.range(1, 3).subscribe(new Subscriber<Integer>() {
//rx2 新加入的方法,在订阅后发送数据之前,
//回首先调用这个方法,而Disposable可用于取消订阅
@Override
public void onSubscribe(Subscription s) {
System.out.println("OnSubscribe start");
s.request(Long.MAX_VALUE);//用request(n)将会立即调用onNext,批量发数据
System.out.println("OnSubscribe end");
}
@Override
public void onNext(Integer v) {
System.out.println(v);
}
@Override
public void onError(Throwable e) {
e.printStackTrace();
}
@Override
public void onComplete() {
System.out.println("Done");
}
});
打印结果
OnSubscribe start
1
2
3
Done
OnSubscribe end
###### Single 使用(监听单一事件)
是Observable的变种,只能发射一个值或一个错误通知,有3个方法:
Single只会调用onSuccess onError这两个方法中的一个,而且只会调用一次,调用了任何一个方法之后,订阅关系终止。
interface SingleObserver<T> {
void onSubscribe(Disposable d);
void onSuccess(T value);
void onError(Throwable error);
}
###### Completable使用(只监听 completed or failed 事件)
Completable 本质上来说和 Observable 与 Single 不一样,因为它**不发射数据**。因此 Completable 的操作符也有所区别,最常用的是andThen。在这个操作符中你可以传任何Observable、Single、Flowable、Maybe或者其他Completable,它们会在原来的 Completable 结束后执行。
这里不发射数据了,也就没有sumer发射器了,但是多了一个Action,当然,它也是不发射数据的。
###### Maybe
Maybe 是 RxJava2.x 之后才有的新类型,可以看成是Single和Completable的结合。
Maybe创建之后,MaybeEmitter 和 SingleEmitter 一样并没有onNext()方法,同样需要通过onSuccess()方法来发射数据。
Maybe也只能发射0或者1个数据,即使发射多个数据,后面发射的数据也不会处理。
####参考
+ [ReactiveX/RxJava文档中文版](https://mcxiaoke.gitbooks.io/rxdocs/content/https://mcxiaoke.gitbooks.io/rxdocs/content/)