Android拾萃 - RxJava操作符列表和响应类型（二）

经过上一篇文章 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

[b]Observable使用场景:[/b]

数据流最长不超过1000个元素，即随着时间的流逝，应用没有机会报OOME(OutOfMemoryError)错误。

处理诸如鼠标移动或触摸事件之类的GUI事件

[b]Flowable使用场景:[/b]

处理超过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 {

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文档中文版