RxJava3.x入门(七)——背压策略

一、背压简介

上下游在不同的线程中，通过Observable发射，处理，响应数据流时，如果上游发射数据的速度快于下游接收处理数据的速度，这样对于那些没来得及处理的数据就会造成积压，这些数据既不会丢失，也不会被垃圾回收机制回收，而是存放在一个异步缓存池中，如果缓存池中的数据一直得不到处理，越积越多，最后就会造成内存溢出，这便是响应式编程中的背压（backpressure）问题。

背压是指在异步场景中，被观察者发送事件速度远快于观察者的处理速度的情况下，一种告诉上游的被观察者降低发送速度的策略。

在Rxjava1.0中，有的Observable支持背压，有的不支持，为了解决这种问题，2.0把支持背压和不支持背压的Observable区分开来：

支持背压的有Flowable类
不支持背压的有Observable，Single, Maybe and Completable类。

注意事项：

在订阅的时候如果使用FlowableSubscriber，那么需要通过s.request(Long.MAX_VALUE)去主动请求上游的数据项。如果遇到背压报错的时候，FlowableSubscriber默认已经将错误try-catch，并通过onError()进行回调，程序并不会崩溃。
在订阅的时候如果使用Consumer，那么不需要主动去请求上游数据，默认已经调用了s.request(Long.MAX_VALUE)。如果遇到背压报错、且对Throwable的Consumer没有new出来，则程序直接崩溃。
背压策略的上游的默认缓存池是128。

二、订阅关系

1.同步订阅

同步订阅就是观察者&被观察者在同一个线程工作。
他们的规则就是被观察者每发一个事情，必须等到观察者接收&处理后，才能继续发送下一个事件。

2.异步订阅

异步订阅就是观察者&被观察者不在同一线程工作。
他们的规则就是被观察者不需要等待观察者接收&处理后才能继续发送下一个事件，而是不断发送（到缓冲区），直到事件完毕（，此后观察者会从缓冲区取出事件）。

此时就有问题暴露出来：

被观察者发送事件速度太快，而观察者来不及接收所有事件，从而导致观察者无法及时响应 / 处理所有发送过来事件的问题，最终导致缓存区溢出、事件丢失 & OOM

那么如何解决这一问题？答案就是背压策略。

三、背压原理

作用：在异步订阅关系中，控制事件发送 & 接收的速度
解决方案 & 原理

而Flowable是RxJava 2.0中被观察者的一种新实现，同时也是背压策略实现的承载者。

所以我们先简单介绍下Flowable的特点和使用。

Flowable 的特点
简单使用

Flowable.create(new FlowableOnSubscribe<Integer>() {@Overridepublic 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, "发送事件 4");emitter.onNext(4);Log.d(TAG, "发送完成");emitter.onComplete();}},//背压模式使用处BackpressureStrategy.ERROR).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Subscriber<Integer>() {@Overridepublic void onSubscribe(Subscription s) {Log.d(TAG, "onSubscribe");//作用：决定观察者能够接收多少个事件// 如设置了s.request(3)，这就说明观察者能够接收3个事件（多出的事件存放在缓存区）// 官方默认推荐使用Long.MAX_VALUE，即s.request(Long.MAX_VALUE);s.request(3);}@Overridepublic void onNext(Integer integer) {Log.d(TAG, "onNext: " + integer);}@Overridepublic void onError(Throwable t) {Log.w(TAG, "onError: ", t);}@Overridepublic void onComplete() {Log.d(TAG, "onComplete");}});

运行结果如下：

可以看出Flowble & Subsciber和之前所讲述的被观察者 & 观察者相似，创建被观察者时多了一个背压策略参数，在观察者中通过request方法设置能接收几个事件。

四、背压策略

1.控制观察者接受事件的速度

简介

1.1 异步订阅情况

情况1. 观察者不接收事件的情况下，被观察者继续发送事件 & 存放到缓存区，此后按需取出

        Button button = findViewById(R.id.btn);button.setOnClickListener(new View.OnClickListener() {@Overridepublic void onClick(View v) {subscription.request(2);}});Flowable.create(new FlowableOnSubscribe<Integer>() {@Overridepublic 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, "发送事件 4");emitter.onNext(4);Log.d(TAG, "发送完成");emitter.onComplete();}},//背压模式使用处BackpressureStrategy.ERROR).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Subscriber<Integer>() {@Overridepublic void onSubscribe(Subscription s) {Log.d(TAG, "onSubscribe");//作用：决定观察者能够接收多少个事件// 如设置了s.request(3)，这就说明观察者能够接收3个事件（多出的事件存放在缓存区）// 官方默认推荐使用Long.MAX_VALUE，即s.request(Long.MAX_VALUE);subscription = s;}@Overridepublic void onNext(Integer integer) {Log.d(TAG, "onNext: " + integer);}@Overridepublic void onError(Throwable t) {Log.w(TAG, "onError: ", t);}@Overridepublic void onComplete() {Log.d(TAG, "onComplete");}});

代码运行结果：

未点击按钮：
点击一次按钮：
再次点击按钮：

情况2. 观察者不接收事件的情况下，被观察者继续发送事件至超出缓存区大小

 Flowable.create(new FlowableOnSubscribe<Integer>() {@Overridepublic void subscribe(FlowableEmitter<Integer> emitter) throws Exception {for (int i = 0; i < 3; i++) {Log.d(TAG, "发送了事件" + i);emitter.onNext(i);}emitter.onComplete();}},BackpressureStrategy.ERROR).subscribe(new Subscriber<Integer>() {@Overridepublic void onSubscribe(Subscription s) {Log.d(TAG, "onSubscribe");// 默认不设置可接收事件大小s.request(3);}@Overridepublic void onNext(Integer integer) {Log.d(TAG, "接收到了事件" + integer);}@Overridepublic void onError(Throwable t) {Log.w(TAG, "onError: ", t);}@Overridepublic void onComplete() {Log.d(TAG, "onComplete");}});

代码运行结果：

1.2 同步订阅情况

与异步订阅方式相比，同步订阅是被观察者在发送1个事件后，必须等待观察者接收后，才能继续发下1个事件

        Flowable.create(new FlowableOnSubscribe<Integer>() {@Overridepublic void subscribe(FlowableEmitter<Integer> emitter) throws Exception {for (int i = 0; i < 3; i++) {Log.d(TAG, "发送了事件" + i);emitter.onNext(i);}emitter.onComplete();}},BackpressureStrategy.ERROR).subscribe(new Subscriber<Integer>() {@Overridepublic void onSubscribe(Subscription s) {Log.d(TAG, "onSubscribe");// 默认不设置可接收事件大小s.request(3);}@Overridepublic void onNext(Integer integer) {Log.d(TAG, "接收到了事件" + integer);}@Overridepublic void onError(Throwable t) {Log.w(TAG, "onError: ", t);}@Overridepublic void onComplete() {Log.d(TAG, "onComplete");}});

代码运行结果：

上述代码能够体现出不会出现被观察者发送事件速度 > 观察者接收事件速度的情况。

可是，却会出现被观察者发送事件数量 > 观察者接收事件数量的问题。

Flowable.create(new FlowableOnSubscribe<Integer>() {@Overridepublic void subscribe(FlowableEmitter<Integer> emitter) throws Exception {for (int i = 0; i < 4; i++) {Log.d(TAG, "发送了事件" + i);emitter.onNext(i);}emitter.onComplete();}},BackpressureStrategy.ERROR).subscribe(new Subscriber<Integer>() {@Overridepublic void onSubscribe(Subscription s) {Log.d(TAG, "onSubscribe");// 默认不设置可接收事件大小s.request(3);}@Overridepublic void onNext(Integer integer) {Log.d(TAG, "接收到了事件" + integer);}@Overridepublic void onError(Throwable t) {Log.w(TAG, "onError: ", t);}@Overridepublic void onComplete() {Log.d(TAG, "onComplete");}});

代码运行结果：

2.控制被观察者发送事件的速度

简介

`FlowableEmitter`类的`requested()`介绍

public interface FlowableEmitter<T> extends Emitter<T> {// FlowableEmitter = 1个接口，继承自Emitter
// Emitter接口方法包括：onNext(),onComplete() & onErrorlong requested();// 作用：返回当前线程中request（a）中的a值// 该request（a）则是措施1中讲解的方法，作用  = 设置
}

每个线程中的requested（）的返回值 = 该线程中的request（a）中a的值

2.1 同步订阅情况

原理

在同步订阅情况中，被观察者通过 FlowableEmitter.requested()获得了观察者自身接收事件能力，从而根据该信息控制事件发送速度，从而达到了观察者反向控制被观察者的效果。

代码演示：

        Flowable.create(new FlowableOnSubscribe<Integer>() {@Overridepublic 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>() {@Overridepublic void onSubscribe(Subscription s) {Log.d(TAG, "onSubscribe");// 设置观察者每次能接受10个事件s.request(10);}@Overridepublic void onNext(Integer integer) {Log.d(TAG, "接收到了事件" + integer);}@Overridepublic void onError(Throwable t) {Log.w(TAG, "onError: ", t);}@Overridepublic void onComplete() {Log.d(TAG, "onComplete");}});

代码运行结果：

在同步订阅情况中使用FlowableEmitter.requested()时，有以下几种使用特性需要注意的：

可叠加性

Subscription.request（a1）；
Subscription.request（a2）；FlowableEmitter.requested()的返回值 = a1 + a2

实时更新性

即每次发送事件后，emitter.requested()会实时更新观察者能接受的事件


Subscription.request（10）；
// FlowableEmitter.requested()的返回值 = 10FlowableEmitter.onNext(1); // 发送了1个事件
// FlowableEmitter.requested()的返回值 = 9

异常

当FlowableEmitter.requested()减到0时，则代表观察者已经不可接收事件，此时被观察者若继续发送事件，则会抛出MissingBackpressureException异常

2.2 异步订阅情况

原理

由于二者处于不同线程，所以被观察者无法通过 FlowableEmitter.requested() 知道观察者自身接收事件能力，即 被观察者不能根据观察者自身接收事件的能力控制发送事件的速度。

在异步订阅关系中，反向控制的原理是：通过RxJava内部固定调用被观察者线程中的 request(n)从而反向控制被观察者的发送事件速度。

RxJava内部调用request(n)（n = 128、96、0）的逻辑如下：

代码演示：

// 被观察者：一共需要发送500个事件，但真正开始发送事件的前提 = FlowableEmitter.requested()返回值 ≠ 0
// 观察者：每次接收事件数量 = 48（点击按钮）Flowable.create(new FlowableOnSubscribe<Integer>() {@Overridepublic 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>() {@Overridepublic void onSubscribe(Subscription s) {Log.d(TAG, "onSubscribe");subscription = s;// 初始状态 = 不接收事件；通过点击按钮接收事件}@Overridepublic void onNext(Integer integer) {Log.d(TAG, "接收到了事件" + integer);}@Overridepublic void onError(Throwable t) {Log.w(TAG, "onError: ", t);}@Overridepublic void onComplete() {Log.d(TAG, "onComplete");}});// 点击按钮才会接收事件 = 48 / 次Button button = findViewById(R.id.btn);button.setOnClickListener(new View.OnClickListener() {@Overridepublic void onClick(View view) {subscription.request(48);// 点击按钮 则 接收48个事件}});

以上代码流程图：

3.采用背压策略模式：BackpressureStrategy

介绍
面向对象：针对缓存区
作用：当缓存区大小存满、被观察者仍然继续发送下1个事件时，该如何处理的策略方式

背压模式有如下几种类型：

模式1：BackpressureStrategy.ERROR

处理方式：直接抛出异常MissingBackpressureException

 // 创建被观察者FlowableFlowable.create(new FlowableOnSubscribe<Integer>() {@Overridepublic 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>() {@Overridepublic void onSubscribe(Subscription s) {Log.d(TAG, "onSubscribe");}@Overridepublic void onNext(Integer integer) {Log.d(TAG, "接收到了事件" + integer);}@Overridepublic void onError(Throwable t) {Log.w(TAG, "onError: ", t);}@Overridepublic void onComplete() {Log.d(TAG, "onComplete");}});

代码运行结果：

BackpressureStrategy.MISSING

处理方式：友好提示：缓存区满了

        // 创建被观察者FlowableFlowable.create(new FlowableOnSubscribe<Integer>() {@Overridepublic 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.MISSING) // 设置背压模式 = BackpressureStrategy.ERROR.subscribeOn(Schedulers.io()) // 设置被观察者在io线程中进行.observeOn(AndroidSchedulers.mainThread()) // 设置观察者在主线程中进行.subscribe(new Subscriber<Integer>() {@Overridepublic void onSubscribe(Subscription s) {Log.d(TAG, "onSubscribe");}@Overridepublic void onNext(Integer integer) {Log.d(TAG, "接收到了事件" + integer);}@Overridepublic void onError(Throwable t) {Log.w(TAG, "onError: ", t);}@Overridepublic void onComplete() {Log.d(TAG, "onComplete");}});

代码运行结果：

模式3：BackpressureStrategy.BUFFER

处理方式：将缓存区大小设置成无限大

// 创建被观察者FlowableFlowable.create(new FlowableOnSubscribe<Integer>() {@Overridepublic void subscribe(FlowableEmitter<Integer> emitter) throws Exception {// 发送 129个事件for (int i = 1;i< 130; i++) {Log.d(TAG, "发送了事件" + i);emitter.onNext(i);}emitter.onComplete();}}, BackpressureStrategy.BUFFER) // 设置背压模式 = BackpressureStrategy.BUFFER.subscribeOn(Schedulers.io()) // 设置被观察者在io线程中进行.observeOn(AndroidSchedulers.mainThread()) // 设置观察者在主线程中进行.subscribe(new Subscriber<Integer>() {@Overridepublic void onSubscribe(Subscription s) {Log.d(TAG, "onSubscribe");}@Overridepublic void onNext(Integer integer) {Log.d(TAG, "接收到了事件" + integer);}@Overridepublic void onError(Throwable t) {Log.w(TAG, "onError: ", t);}@Overridepublic void onComplete() {Log.d(TAG, "onComplete");}});

代码运行结果：

发现可以接收超过原先缓存区大小（128）的事件数量了。

模式4： BackpressureStrategy.DROP

处理方式：超过缓存区大小（128）的事件丢弃

如发送了150个事件，观察者只需要128件事情，仅保存第1 - 第128个事件，第129 -第150事件将被丢弃

// 创建被观察者FlowableFlowable.create(new FlowableOnSubscribe<Integer>() {@Overridepublic void subscribe(FlowableEmitter<Integer> emitter) throws Exception {// 发送 150个事件for (int i = 0;i< 150; i++) {Log.d(TAG, "发送了事件" + i);emitter.onNext(i);}emitter.onComplete();}}, BackpressureStrategy.DROP) // 设置背压模式 = BackpressureStrategy.DROP.subscribeOn(Schedulers.io()) // 设置被观察者在io线程中进行.observeOn(AndroidSchedulers.mainThread()) // 设置观察者在主线程中进行.subscribe(new Subscriber<Integer>() {@Overridepublic void onSubscribe(Subscription s) {Log.d(TAG, "onSubscribe");s.request(128);}@Overridepublic void onNext(Integer integer) {Log.d(TAG, "接收到了事件" + integer);}@Overridepublic void onError(Throwable t) {Log.w(TAG, "onError: ", t);}@Overridepublic void onComplete() {Log.d(TAG, "onComplete");}});

代码运行结果：

模式5：BackpressureStrategy.LATEST

处理方式：只保存最新（最后）事件，超过缓存区大小（128）的事件丢弃

即如果发送了150个事件，缓存区里会保存129个事件（第1-第128 + 第150事件）

// 创建被观察者FlowableFlowable.create(new FlowableOnSubscribe<Integer>() {@Overridepublic void subscribe(FlowableEmitter<Integer> emitter) throws Exception {// 发送 150个事件for (int i = 0;i< 150; i++) {Log.d(TAG, "发送了事件" + i);emitter.onNext(i);}emitter.onComplete();}}, BackpressureStrategy.LATEST) // 设置背压模式 = BackpressureStrategy.LATEST.subscribeOn(Schedulers.io()) // 设置被观察者在io线程中进行.observeOn(AndroidSchedulers.mainThread()) // 设置观察者在主线程中进行.subscribe(new Subscriber<Integer>() {@Overridepublic void onSubscribe(Subscription s) {Log.d(TAG, "onSubscribe");subscription = s;}@Overridepublic void onNext(Integer integer) {Log.d(TAG, "接收到了事件" + integer);}@Overridepublic void onError(Throwable t) {Log.w(TAG, "onError: ", t);}@Overridepublic void onComplete() {Log.d(TAG, "onComplete");}});Button button = findViewById(R.id.btn);button.setOnClickListener(new View.OnClickListener() {@Overridepublic void onClick(View v) {subscription.request(128);}});

代码运行结果：
点击一次按钮

再次点击按钮

参考文献

讲的超详细的一篇文章：
Android RxJava ：图文详解背压策略
当然还有冰炭不投day yyds！
Android之Rxjava2.X 8————Rxjava 背压策略

RxJava3.x系列文章索引

RxJava3.x入门(一)——概述及使用
RxJava3.x入门(二)——创建操作符和变换操作符
RxJava3.x入门(三)——组合操作符
RxJava3.x入门(四)——过滤操作符
RxJava3.x入门(五)——功能操作符
RxJava3.x入门(六)——条件操作符
RxJava3.x入门(七)——背压策略