1 两种订阅关系:同步 & 异步
- 对于异步订阅关系,存在 被观察者发送事件速度 与观察者接收事件速度 不匹配的情况
- 发送 & 接收事件速度 = 单位时间内 发送&接收事件的数量
- 大多数情况,主要是 被观察者发送事件速度 > 观察者接收事件速度
2 问题
- 被观察者 发送事件速度太快,而观察者 来不及接收所有事件,从而导致观察者无法及时响应 / 处理所有发送过来事件的问题,最终导致缓存区溢出、事件丢失 & OOM
- 如,点击按钮事件:连续过快的点击按钮10次,则只会造成点击2次的效果;
- 解释:因为点击速度太快了,所以按钮来不及响应
解决方案
采用 背压策略。
3 背压策略的原理
4 背压策略的具体实现:Flowable
4.1 Flowable
介绍
- 定义:在 RxJava2.0中,被观察者(
Observable
)的一种新实现 - 作用:实现 非阻塞式背压 策略
4.2 Flowable 特点
-
Flowable
的特点 具体如下
实际上,RxJava2.0 也有保留(被观察者)Observerble - Observer(观察者)的观察者模型
4.3 Flowable的基础使用
/**
* 步骤1:创建被观察者 = Flowable
*/
Flowable<Integer> upstream = Flowable.create(new FlowableOnSubscribe<Integer>() {
@Override
public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
emitter.onComplete();
}
}, BackpressureStrategy.ERROR);
// 需要传入背压参数BackpressureStrategy,下面会详细讲解
/**
* 步骤2:创建观察者 = Subscriber
*/
Subscriber<Integer> downstream = new Subscriber<Integer>() {
@Override
public void onSubscribe(Subscription s) {
// 对比Observer传入的Disposable参数,Subscriber此处传入的参数 = Subscription
// 相同点:Subscription具备Disposable参数的作用,即Disposable.dispose()切断连接, 同样的调用Subscription.cancel()切断连接
// 不同点:Subscription增加了void request(long n)
Log.d(TAG, "onSubscribe");
s.request(Long.MAX_VALUE);
// 关于request()下面会继续详细说明
}
@Override
public void onNext(Integer integer) {
Log.d(TAG, "onNext: " + integer);
}
@Override
public void onError(Throwable t) {
Log.w(TAG, "onError: ", t);
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete");
}
};
/**
* 步骤3:建立订阅关系
*/
upstream.subscribe(downstream);
// 步骤1:创建被观察者 = Flowable
Flowable.create(new FlowableOnSubscribe<Integer>() {
@Override
public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
Log.d(TAG, "发送事件 1");
emitter.onNext(1);
Log.d(TAG, "发送事件 2");
emitter.onNext(2);
Log.d(TAG, "发送事件 3");
emitter.onNext(3);
Log.d(TAG, "发送完成");
emitter.onComplete();
}
}, BackpressureStrategy.ERROR)
.subscribe(new Subscriber<Integer>() {
// 步骤2:创建观察者 = Subscriber & 建立订阅关系
@Override
public void onSubscribe(Subscription s) {
Log.d(TAG, "onSubscribe");
s.request(3);
}
@Override
public void onNext(Integer integer) {
Log.d(TAG, "接收到了事件" + integer);
}
@Override
public void onError(Throwable t) {
Log.w(TAG, "onError: ", t);
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete");
}
});
5. 背压策略的使用
5.1 控制 观察者接收事件 的速度
5.1.1 异步订阅情况
// 1. 创建被观察者Flowable
Flowable.create(new FlowableOnSubscribe<Integer>() {
@Override
public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
// 一共发送4个事件
Log.d(TAG, "发送事件 1");
emitter.onNext(1);
Log.d(TAG, "发送事件 2");
emitter.onNext(2);
Log.d(TAG, "发送事件 3");
emitter.onNext(3);
Log.d(TAG, "发送事件 4");
emitter.onNext(4);
Log.d(TAG, "发送完成");
emitter.onComplete();
}
}, BackpressureStrategy.ERROR).subscribeOn(Schedulers.io()) // 设置被观察者在io线程中进行
.observeOn(AndroidSchedulers.mainThread()) // 设置观察者在主线程中进行
.subscribe(new Subscriber<Integer>() {
@Override
public void onSubscribe(Subscription s) {
// 对比Observer传入的Disposable参数,Subscriber此处传入的参数 = Subscription
// 相同点:Subscription参数具备Disposable参数的作用,即Disposable.dispose()切断连接, 同样的调用Subscription.cancel()切断连接
// 不同点:Subscription增加了void request(long n)
s.request(3);
// 作用:决定观察者能够接收多少个事件
// 如设置了s.request(3),这就说明观察者能够接收3个事件(多出的事件存放在缓存区)
// 官方默认推荐使用Long.MAX_VALUE,即s.request(Long.MAX_VALUE);
}
@Override
public void onNext(Integer integer) {
Log.d(TAG, "接收到了事件" + integer);
}
@Override
public void onError(Throwable t) {
Log.w(TAG, "onError: ", t);
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete");
}
});
注意
观察者不接收事件的情况下,被观察者继续发送事件至超出缓存区大小(128)
Flowable.create(new FlowableOnSubscribe<Integer>() {
@Override
public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
// 一共发送129个事件,即超出了缓存区的大小
for (int i = 0;i< 129; i++) {
Log.d(TAG, "发送了事件" + i);
emitter.onNext(i);
}
emitter.onComplete();
}
}, BackpressureStrategy.ERROR).subscribeOn(Schedulers.io()) // 设置被观察者在io线程中进行
.observeOn(AndroidSchedulers.mainThread()) // 设置观察者在主线程中进行
.subscribe(new Subscriber<Integer>() {
@Override
public void onSubscribe(Subscription s) {
Log.d(TAG, "onSubscribe");
// 默认不设置可接收事件大小
}
@Override
public void onNext(Integer integer) {
Log.d(TAG, "接收到了事件" + integer);
}
@Override
public void onError(Throwable t) {
Log.w(TAG, "onError: ", t);
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete");
}
});
5.1.2 同步订阅情况
同步订阅 & 异步订阅 的区别在于:
- 同步订阅中,被观察者 & 观察者工作于同1线程
- 同步订阅关系中没有缓存区
实际上并不会出现被观察者发送事件速度 > 观察者接收事件速度的情况。可是,却会出现被观察者发送事件数量 > 观察者接收事件数量的问题。
/**
* 同步情况
*/
/**
* 步骤1:创建被观察者 = Flowable
*/
Flowable<Integer> upstream = Flowable.create(new FlowableOnSubscribe<Integer>() {
@Override
public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
Log.d(TAG, "发送了事件1");
emitter.onNext(1);
Log.d(TAG, "发送了事件2");
emitter.onNext(2);
Log.d(TAG, "发送了事件3");
emitter.onNext(3);
emitter.onComplete();
}
}, BackpressureStrategy.ERROR);
/**
* 步骤2:创建观察者 = Subscriber
*/
Subscriber<Integer> downstream = new Subscriber<Integer>() {
@Override
public void onSubscribe(Subscription s) {
Log.d(TAG, "onSubscribe");
// 不设置request(long n)
// s.request(Long.MAX_VALUE);
}
@Override
public void onNext(Integer integer) {
Log.d(TAG, "onNext: " + integer);
}
@Override
public void onError(Throwable t) {
Log.w(TAG, "onError: ", t);
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete");
}
};
/**
* 步骤3:建立订阅关系
*/
upstream.subscribe(downstream);
在被观察者发送第1个事件后, 就抛出MissingBackpressureException
异常 & 观察者没有收到任何事件
5.2 控制 被观察者发送事件 的速度
对应于同步 & 异步订阅情况 的原理图
5.2.1 同步订阅情况
即在同步订阅情况中,被观察者 通过 FlowableEmitter.requested()
获得了观察者自身接收事件能力,从而根据该信息控制事件发送速度,从而达到了观察者反向控制被观察者的效果
Flowable.create(new FlowableOnSubscribe<Integer>() {
@Override
public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
// 调用emitter.requested()获取当前观察者需要接收的事件数量
long n = emitter.requested();
Log.d(TAG, "观察者可接收事件" + n);
// 根据emitter.requested()的值,即当前观察者需要接收的事件数量来发送事件
for (int i = 0; i < n; i++) {
Log.d(TAG, "发送了事件" + i);
emitter.onNext(i);
}
}
}, BackpressureStrategy.ERROR)
.subscribe(new Subscriber<Integer>() {
@Override
public void onSubscribe(Subscription s) {
Log.d(TAG, "onSubscribe");
// 设置观察者每次能接受10个事件
s.request(10);
}
@Override
public void onNext(Integer integer) {
Log.d(TAG, "接收到了事件" + integer);
}
@Override
public void onError(Throwable t) {
Log.w(TAG, "onError: ", t);
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete");
}
});
特别注意
情况1:可叠加性
- 即:观察者可连续要求接收事件,被观察者会进行叠加并一起发送
Flowable.create(new FlowableOnSubscribe<Integer>() {
@Override
public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
// 调用emitter.requested()获取当前观察者需要接收的事件数量
Log.d(TAG, "观察者可接收事件" + emitter.requested());
}
}, BackpressureStrategy.ERROR)
.subscribe(new Subscriber<Integer>() {
@Override
public void onSubscribe(Subscription s) {
Log.d(TAG, "onSubscribe");
s.request(10); // 第1次设置观察者每次能接受10个事件
s.request(20); // 第2次设置观察者每次能接受20个事件
}
@Override
public void onNext(Integer integer) {
Log.d(TAG, "接收到了事件" + integer);
}
@Override
public void onError(Throwable t) {
Log.w(TAG, "onError: ", t);
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete");
}
});
情况2:实时更新性
- 即,每次发送事件后,
emitter.requested()
会实时更新观察者能接受的事件
- 即一开始观察者要接收10个事件,发送了1个后,会实时更新为9个
- 仅计算
Next
事件,complete & error
事件不算
Flowable.create(new FlowableOnSubscribe<Integer>() {
@Override
public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
// 1. 调用emitter.requested()获取当前观察者需要接收的事件数量
Log.d(TAG, "观察者可接收事件数量 = " + emitter.requested());
// 2. 每次发送事件后,emitter.requested()会实时更新观察者能接受的事件
// 即一开始观察者要接收10个事件,发送了1个后,会实时更新为9个
Log.d(TAG, "发送了事件 1");
emitter.onNext(1);
Log.d(TAG, "发送了事件1后, 还需要发送事件数量 = " + emitter.requested());
Log.d(TAG, "发送了事件 2");
emitter.onNext(2);
Log.d(TAG, "发送事件2后, 还需要发送事件数量 = " + emitter.requested());
Log.d(TAG, "发送了事件 3");
emitter.onNext(3);
Log.d(TAG, "发送事件3后, 还需要发送事件数量 = " + emitter.requested());
emitter.onComplete();
}
}, BackpressureStrategy.ERROR)
.subscribe(new Subscriber<Integer>() {
@Override
public void onSubscribe(Subscription s) {
Log.d(TAG, "onSubscribe");
s.request(10); // 设置观察者每次能接受10个事件
}
@Override
public void onNext(Integer integer) {
Log.d(TAG, "接收到了事件" + integer);
}
@Override
public void onError(Throwable t) {
Log.w(TAG, "onError: ", t);
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete");
}
});
情况3:异常
- 当
FlowableEmitter.requested()
减到0时,则代表观察者已经不可接收事件 - 此时被观察者若继续发送事件,则会抛出
MissingBackpressureException
异常
如观察者可接收事件数量 = 1,当被观察者发送第2个事件时,就会抛出异常
额外
- 若观察者没有设置可接收事件数量,即无调用
Subscription.request()
- 那么被观察者默认观察者可接收事件数量 = 0,即
FlowableEmitter.requested()
的返回值 = 0
5.2.2 异步订阅情况
- 被观察者 无法通过
FlowableEmitter.requested()
知道观察者自身接收事件能力,即 被观察者不能根据 观察者自身接收事件的能力 控制发送事件的速度。 - 而在异步订阅关系中,反向控制的原理是:通过
RxJava
内部固定调用被观察者线程中的request(n)
从而 反向控制被观察者的发送事件速度
具体使用
RxJava内部调用request(n)(n = 128、96、0)至于为什么是调用request(128) & request(96) & request(0),感兴趣的读者可自己阅读 Flowable的源码
// 被观察者:一共需要发送500个事件,但真正开始发送事件的前提 = FlowableEmitter.requested()返回值 ≠ 0
// 观察者:每次接收事件数量 = 48(点击按钮)
Flowable.create(new FlowableOnSubscribe<Integer>() {
@Override
public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
Log.d(TAG, "观察者可接收事件数量 = " + emitter.requested());
boolean flag; //设置标记位控制
// 被观察者一共需要发送500个事件
for (int i = 0; i < 500; i++) {
flag = false;
// 若requested() == 0则不发送
while (emitter.requested() == 0) {
if (!flag) {
Log.d(TAG, "不再发送");
flag = true;
}
}
// requested() ≠ 0 才发送
Log.d(TAG, "发送了事件" + i + ",观察者可接收事件数量 = " + emitter.requested());
emitter.onNext(i);
}
}
}, BackpressureStrategy.ERROR).subscribeOn(Schedulers.io()) // 设置被观察者在io线程中进行
.observeOn(AndroidSchedulers.mainThread()) // 设置观察者在主线程中进行
.subscribe(new Subscriber<Integer>() {
@Override
public void onSubscribe(Subscription s) {
Log.d(TAG, "onSubscribe");
mSubscription = s;
// 初始状态 = 不接收事件;通过点击按钮接收事件
}
@Override
public void onNext(Integer integer) {
Log.d(TAG, "接收到了事件" + integer);
}
@Override
public void onError(Throwable t) {
Log.w(TAG, "onError: ", t);
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete");
}
});
// 点击按钮才会接收事件 = 48 / 次
btn = (Button) findViewById(R.id.btn);
btn.setOnClickListener(new View.OnClickListener() {
@Override
public void onClick(View view) {
mSubscription.request(48);
// 点击按钮 则 接收48个事件
}
});
整个流程 & 测试结果 请看下图
5.3 采用背压策略模式:BackpressureStrategy
5.3.1 背压模式介绍
- 在Flowable的使用中,会被要求传入背压模式参数
- 面向对象:针对缓存区
- 作用:当缓存区大小存满、被观察者仍然继续发送下1个事件时,该如何处理的策略方式
5.3.2 背压模式类型
具体表现是:出现当缓存区大小存满(默认缓存区大小 = 128)、被观察者仍然继续发送下1个事件时
模式 | 处理方式 |
---|---|
BackpressureStrategy.ERROR | 直接抛出异常MissingBackpressureException
|
BackpressureStrategy.MISSING | 友好提示:缓存区满了 |
BackpressureStrategy.BUFFER | 将缓存区大小设置成无限大(但要注意内存情况,防止出现OOM) |
BackpressureStrategy.DROP | 超过缓存区大小(128)的事件丢弃 |
BackpressureStrategy.LATEST | 只保存最新(最后)事件,超过缓存区大小(128)的事件丢弃(即如果发送了150个事件,缓存区里会保存129个事件(第1-第128 + 第150事件)) |
// 创建被观察者Flowable
Flowable.create(new FlowableOnSubscribe<Integer>() {
@Override
public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
// 发送 129个事件
for (int i = 0;i< 129; i++) {
Log.d(TAG, "发送了事件" + i);
emitter.onNext(i);
}
emitter.onComplete();
}
}, BackpressureStrategy.ERROR) // 设置背压模式 = BackpressureStrategy.ERROR
.subscribeOn(Schedulers.io()) // 设置被观察者在io线程中进行
.observeOn(AndroidSchedulers.mainThread()) // 设置观察者在主线程中进行
.subscribe(new Subscriber<Integer>() {
@Override
public void onSubscribe(Subscription s) {
Log.d(TAG, "onSubscribe");
}
@Override
public void onNext(Integer integer) {
Log.d(TAG, "接收到了事件" + integer);
}
@Override
public void onError(Throwable t) {
Log.w(TAG, "onError: ", t);
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete");
}
});
5.3.3 特别注意
在使用背压策略模式的时候,有1种情况是需要注意的:
背景
FLowable
可通过自己创建(如上面例子),或通过其他方式自动创建,如interval
操作符
interval操作符简介
- 作用:每隔1段时间就产生1个数字(Long型),从0开始、1次递增1,直至无穷大
- 默认运行在1个新线程上
- 与timer操作符区别:timer操作符可结束发送
冲突
- 对于自身手动创建
FLowable
的情况,可通过传入背压模式参数选择背压策略
(即上面描述的) - 对于自动创建FLowable,却无法手动传入传入背压模式参数,RxJava 2.0内部提供 封装了背压策略模式的方法.
onBackpressureBuffer()
onBackpressureDrop()
onBackpressureLatest()
- 默认采用
BackpressureStrategy.ERROR
模式
Flowable.interval(1, TimeUnit.MILLISECONDS)
.onBackpressureBuffer() // 添加背压策略封装好的方法,此处选择Buffer模式,即缓存区大小无限制
.observeOn(Schedulers.newThread())
.subscribe(new Subscriber<Long>() {
@Override
public void onSubscribe(Subscription s) {
Log.d(TAG, "onSubscribe");
mSubscription = s;
s.request(Long.MAX_VALUE);
}
@Override
public void onNext(Long aLong) {
Log.d(TAG, "onNext: " + aLong);
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
@Override
public void onError(Throwable t) {
Log.w(TAG, "onError: ", t);
}
@Override
public void onComplete() {
Log.d(TAG, "onComplete");
}
});