Android拾萃 - RxJava５种响应类型的区别和背压（三）

Android拾萃 - RxJava最简单的入门（一）

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

本文首先会给出ReactiveX的核心操作符列表和对应的文档链接，后面会具体的介绍和实践所有的操作符。

如果想实现自己的操作符，可以参考：实现自定义操作符

创建操作

用于创建Observable的操作符

Create

— 通过调用观察者的方法从头创建一个Observable

Defer

— 在观察者订阅之前不创建这个Observable，为每一个观察者创建一个新的Observable

Empty/Never/Throw

— 创建行为受限的特殊Observable

From

— 将其它的对象或数据结构转换为Observable

Interval

— 创建一个定时发射整数序列的Observable

Just

— 将对象或者对象集合转换为一个会发射这些对象的Observable

Range

— 创建发射指定范围的整数序列的Observable

Repeat

— 创建重复发射特定的数据或数据序列的Observable

Start

— 创建发射一个函数的返回值的Observable

Timer

— 创建在一个指定的延迟之后发射单个数据的Observable

变换操作

这些操作符可用于对Observable发射的数据进行变换，详细解释可以看每个操作符的文档

Buffer

— 缓存，可以简单的理解为缓存，它定期从Observable收集数据到一个集合，然后把这些数据集合打包发射，而不是一次发射一个

FlatMap

— 扁平映射，将Observable发射的数据变换为Observables集合，然后将这些Observable发射的数据平坦化的放进一个单独的Observable，可以认为是一个将嵌套的数据结构展开的过程。

GroupBy

— 分组，将原来的Observable分拆为Observable集合，将原始Observable发射的数据按Key分组，每一个Observable发射一组不同的数据

Map

— 映射，通过对序列的每一项都应用一个函数变换Observable发射的数据，实质是对序列中的每一项执行一个函数，函数的参数就是这个数据项

Scan

— 扫描，对Observable发射的每一项数据应用一个函数，然后按顺序依次发射这些值

Window

— 窗口，定期将来自Observable的数据分拆成一些Observable窗口，然后发射这些窗口，而不是每次发射一项。类似于Buffer，但Buffer发射的是数据，Window发射的是Observable，每一个Observable发射原始Observable的数据的一个子集

过滤操作

这些操作符用于从Observable发射的数据中进行选择

Debounce

— 只有在空闲了一段时间后才发射数据，通俗的说，就是如果一段时间没有操作，就执行一次操作

Distinct

— 去重，过滤掉重复数据项

ElementAt

— 取值，取特定位置的数据项

Filter

— 过滤，过滤掉没有通过谓词测试的数据项，只发射通过测试的

First

— 首项，只发射满足条件的第一条数据

IgnoreElements

— 忽略所有的数据，只保留终止通知(onError或onCompleted)

Last

— 末项，只发射最后一条数据

Sample

— 取样，定期发射最新的数据，等于是数据抽样，有的实现里叫ThrottleFirst

Skip

— 跳过前面的若干项数据

SkipLast

— 跳过后面的若干项数据

Take

— 只保留前面的若干项数据

TakeLast

— 只保留后面的若干项数据

组合操作

组合操作符用于将多个Observable组合成一个单一的Observable

And/Then/When

— 通过模式(And条件)和计划(Then次序)组合两个或多个Observable发射的数据集

CombineLatest

— 当两个Observables中的任何一个发射了一个数据时，通过一个指定的函数组合每个Observable发射的最新数据（一共两个数据），然后发射这个函数的结果

Join

— 无论何时，如果一个Observable发射了一个数据项，只要在另一个Observable发射的数据项定义的时间窗口内，就将两个Observable发射的数据合并发射

Merge

— 将两个Observable发射的数据组合并成一个

StartWith

— 在发射原来的Observable的数据序列之前，先发射一个指定的数据序列或数据项

Switch

— 将一个发射Observable序列的Observable转换为这样一个Observable：它逐个发射那些Observable最近发射的数据

Zip

— 打包，使用一个指定的函数将多个Observable发射的数据组合在一起，然后将这个函数的结果作为单项数据发射

错误处理

这些操作符用于从错误通知中恢复

Catch

— 捕获，继续序列操作，将错误替换为正常的数据，从onError通知中恢复

Retry

— 重试，如果Observable发射了一个错误通知，重新订阅它，期待它正常终止

辅助操作

一组用于处理Observable的操作符

Delay

— 延迟一段时间发射结果数据

Do

— 注册一个动作占用一些Observable的生命周期事件，相当于Mock某个操作

Materialize/Dematerialize

— 将发射的数据和通知都当做数据发射，或者反过来

ObserveOn

— 指定观察者观察Observable的调度程序（工作线程）

Serialize

— 强制Observable按次序发射数据并且功能是有效的

Subscribe

— 收到Observable发射的数据和通知后执行的操作

SubscribeOn

— 指定Observable应该在哪个调度程序上执行

TimeInterval

— 将一个Observable转换为发射两个数据之间所耗费时间的Observable

Timeout

— 添加超时机制，如果过了指定的一段时间没有发射数据，就发射一个错误通知

Timestamp

— 给Observable发射的每个数据项添加一个时间戳

Using

— 创建一个只在Observable的生命周期内存在的一次性资源

**条件和布尔操作**v

这些操作符可用于单个或多个数据项，也可用于Observable

All

— 判断Observable发射的所有的数据项是否都满足某个条件

Amb

— 给定多个Observable，只让第一个发射数据的Observable发射全部数据

Contains

— 判断Observable是否会发射一个指定的数据项

DefaultIfEmpty

— 发射来自原始Observable的数据，如果原始Observable没有发射数据，就发射一个默认数据

SequenceEqual

— 判断两个Observable是否按相同的数据序列

SkipUntil

— 丢弃原始Observable发射的数据，直到第二个Observable发射了一个数据，然后发射原始Observable的剩余数据

SkipWhile

— 丢弃原始Observable发射的数据，直到一个特定的条件为假，然后发射原始Observable剩余的数据

TakeUntil

— 发射来自原始Observable的数据，直到第二个Observable发射了一个数据或一个通知

TakeWhile

— 发射原始Observable的数据，直到一个特定的条件为真，然后跳过剩余的数据

算术和聚合操作

这些操作符可用于整个数据序列

Average

— 计算Observable发射的数据序列的平均值，然后发射这个结果

Concat

— 不交错的连接多个Observable的数据

Count

— 计算Observable发射的数据个数，然后发射这个结果

Max

— 计算并发射数据序列的最大值

Min

— 计算并发射数据序列的最小值

Reduce

— 按顺序对数据序列的每一个应用某个函数，然后返回这个值

Sum

— 计算并发射数据序列的和

v连接操作

一些有精确可控的订阅行为的特殊Observable

Connect

— 指示一个可连接的Observable开始发射数据给订阅者

Publish

— 将一个普通的Observable转换为可连接的

RefCount

— 使一个可连接的Observable表现得像一个普通的Observable

Replay

— 确保所有的观察者收到同样的数据序列，即使他们在Observable开始发射数据之后才订阅

转换操作

To

— 将Observable转换为其它的对象或数据结构

Blocking

阻塞Observable的操作符

操作符决策树

几种主要的需求

直接创建一个Observable（创建操作）

组合多个Observable（组合操作）

对Observable发射的数据执行变换操作（变换操作）

从Observable发射的数据中取特定的值（过滤操作）

转发Observable的部分值（条件/布尔/过滤操作）

对Observable发射的数据序列求值（算术/聚合操作）

创建操作

建议在传递给create方法的函数中检查观察者的isUnsubscribed状态，以便在没有观察者的时候，让你的Observable停止发射数据或者做昂贵的运算。

在rxjava2中（本系列都是在rxjava2的基础上进行论述的，rxjava1请自行查阅），

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

再次借用Android拾萃 - RxJava操作符和响应类型（二） 的图片。



下面来看下，五种响应类型具体代码．

[b]1. Observable[/b]

Observable是最完整的，拥有四个回调方法onSubscribe会在收到数据之前调用，可以做一些动画加载之类的操作．onNext收到数据会调用，onError和onComplete是互斥的，只会调用一个．

//ObservableOnSubscribe接口
Observable observable = Observable.create(new ObservableOnSubscribe() {
@Override
public void subscribe(@NonNull ObservableEmitter e) throws Exception {
//判断下是否释放了
if (!e.isDisposed()){
for (int i = 0; i<5; i++) {
e.onNext(i);
}
e.onComplete();
}
}
});

//<Integer>泛型没有添加的话，默认的是object ，然后onext强制转换一下即可
observable.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(@NonNull Disposable d) {

}

@Override
public void onNext(@NonNull Integer o) {
Log.d(TAG, o + "");
}

@Override
public void onError(@NonNull Throwable e) {

}

@Override
public void onComplete() {

}
});

Consumer是简化版的Observer, 可以只关心某个步骤(onNext　onError　　onComplete),对应的有一个参数，两个参数，三个参数的方法如下

可以简单理解微，Action和Consumer可以单独行动，单独存在

Action 对应　　无参的onComplete()

Consumer　对应　单个参数的onError(Throwable t)　或者　onNext(T t)

observable.subscribe(new Consumer() {
@Override
public void accept(Object o) throws Exception {
//对应onNext方法
}
}, new Consumer<Throwable>() {
@Override
public void accept(Throwable throwable) throws Exception {
//对应onError
}
});

observable.subscribe(new Consumer() {
@Override
public void accept(Object o) throws Exception {
//对应onNext方法
}
}, new Consumer<Throwable>() {
@Override
public void accept(Throwable throwable) throws Exception {
//对应onError
}
}, new Action() {
@Override
public void run() throws Exception {
//对应onComplete　　无参返回
}
});

下面的几种case, Consumer就不再讨论了，具体参考上面的代码

[b]2. Flowable[/b]

Backpressure的策略

我们发现 Flowable.create 多了一个参数，背压策略

点进去源码发现总共有五种策略

/**
* Represents the options for applying backpressure to a source sequence.
*/
public enum BackpressureStrategy {
/**
* OnNext events are written without any buffering or dropping.
* Downstream has to deal with any overflow.
* <p>Useful when one applies one of the custom-parameter onBackpressureXXX operators.
*/
MISSING,
/**
* Signals a MissingBackpressureException in case the downstream can't keep up.
*/
ERROR,
/**
* Buffers <em>all</em> onNext values until the downstream consumes it.
*/
BUFFER,
/**
* Drops the most recent onNext value if the downstream can't keep up.
*/
DROP,
/**
* Keeps only the latest onNext value, overwriting any previous value if the
* downstream can't keep up.
*/
LATEST
}

处理Backpressure的策略仅仅是处理Subscriber接收事件的方式，并不影响Flowable发送事件和事件产生速度。

我们知道只有生产者的速度大于消费者的速度，才会产生Backpressure问题。也就是说只会发生在异步的情况下。

Flowable的几种背压策略：

1. BackpressureStrategy.ERROR：缓存区默人大小128，流速不均衡时发射MissingBackpressureException信号。

2. BackpressureStrategy.BUFFER：缓存区不限制大小，使用不当仍会OOM。

3. BackpressureStrategy.DROP：缓存最近的nNext事件。

4. BackpressureStrategy.LATEST：缓存区会保留最后的OnNext事件，覆盖之前缓存的OnNext事件。

5. BackpressureStrategy.MISSING：OnNext事件没有任何缓存和丢弃，下游要处理任何溢出。

onBackpressureBuffer是不丢弃数据的处理方式。把上游收到的全部缓存下来，等下游来请求再发给下游。相当于一个水库。但上游太快，水库（buffer）就会溢出。

onBackpressureDrop和onBackpressureLatest比较类似，都会丢弃数据。这两种策略相当于一种令牌机制（或者配额机制），下游通过request请求产生令牌（配额）给上游，上游接到多少令牌，就给下游发送多少数据。当令牌数消耗到0的时候，上游开始丢弃数据。但这两种策略在令牌数为0的时候有一点微妙的区别：onBackpressureDrop直接丢弃数据，不缓存任何数据；而onBackpressureLatest则缓存最新的一条数据，这样当上游接到新令牌的时候，它就先把缓存的上一条“最新”数据发送给下游。

//我们发现Flowable 多了一个参数，背压策略
Flowable flowable = Flowable.create(new FlowableOnSubscribe<Object>() {
@Override
public void subscribe(@NonNull FlowableEmitter<Object> e) throws Exception {

//判断下是否释放了
if (!e.isCancelled()){
for (int i = 0; i<5; i++) {
e.onNext(i);
}
e.onComplete();
}
}
}, BackpressureStrategy.DROP);

//这里建立关系都是使用了subscribe，只不过对应的方法不一样，请参照上面的表
flowable
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())

.subscribe(new FlowableSubscriber() {
@Override
public void onSubscribe(@NonNull Subscription s) {

}

@Override
public void onNext(Object o) {
Log.d(TAG, o + "");
}

@Override
public void onError(Throwable t) {

}

@Override
public void onComplete() {

}
});

[b]3. Single[/b]

Single 是没有complete的，因为，只发送一个数据之后，默认就会调用complete

Single single = Single.create(new SingleOnSubscribe<Object>() {
@Override
public void subscribe(@NonNull SingleEmitter<Object> e) throws Exception {
e.onSuccess(1);
}
});

single.subscribeOn(Schedulers.io()).subscribe(new SingleObserver() {
@Override
public void onSubscribe(@NonNull Disposable d) {

}

@Override
public void onSuccess(@NonNull Object o) {
Log.d(TAG, o + "");
}

@Override
public void onError(@NonNull Throwable e) {

}
});

[b]4. Completable[/b]

Completable只关心onComplete，所以上游发射的数据，都不会进行处理的．

Completable completable =  Completable.create(new CompletableOnSubscribe() {
@Override
public void subscribe(@NonNull CompletableEmitter e) throws Exception {
e.onComplete();
}
});
completable.subscribe(new CompletableObserver() {
@Override
public void onSubscribe(@NonNull Disposable d) {

}

@Override
public void onComplete() {

}

@Override
public void onError(@NonNull Throwable e) {

}
});

[b]5. Maybe[/b]

Maybe是Single　和　Completable的合体，只会处理一条数据，但是onComplete不会自动调用

Maybe maybe = Maybe.create(new MaybeOnSubscribe() {
@Override
public void subscribe(@NonNull MaybeEmitter e) throws Exception {
e.onSuccess(12);
e.onComplete();
}
});

maybe.subscribe(new MaybeObserver() {
@Override
public void onSubscribe(@NonNull Disposable d) {

}

@Override
public void onSuccess(@NonNull Object o) {
Log.d(TAG, o + "");
}

@Override
public void onError(@NonNull Throwable e) {

}

@Override
public void onComplete() {

}
});