RxJava2系列之背压策略(一)
前言
通过前7篇RxJava的文章，我们对RxJava1.x版本的内容进行了学习与实践。目前RxJava已经更新到2.x了，有小伙伴问我为什么不直接上RxJava2的教程？RxJava2是在1的基础上进行了更新与优化，有很多相通之处，初学者的话建议还是先从RxJava1的基础理论一步步学习。
本系列主要通过与RxJava1比较来学习RxJava2都有哪些改变。而RxJava2中最大的优化之处就在于它解决了RxJava1中未能有效解决的背压(Backpressure)问题。本篇我们就来看看什么是背压，以及RxJava1中是如何解决这个问题。
背压问题
先来个定义：背压问题是指在RxJava观察者模式的异步场景中，由于被观察者生产事件的速度远远快于观察者消费事件的速度，从而导致生产的事件堆积，最后致使内存溢出，程序崩溃。
再上个场景：工厂生产面包，消费者吃面包。工厂生产面包速度慢了，消费者处于等待状态，来一个吃一个，这没什么影响。而如果工厂生产面包的速度过快，导致消费者来不及吃，面包就会累积的越来越多，最后就会过期浪费。
来看代码演示这个问题：
//被观察者每过1ms发射一个事件
Observable.interval(1, TimeUnit.MILLISECONDS)
.observeOn(Schedulers.newThread())
.subscribe((aLong) -> {
//观察者每过800ms处理一个事件
try {
Thread.sleep(800);
} catch (InterruptedException e) {
e.printStackTrace();
}
Log.e("rx_test", "back_pressure：" + aLong);
}); 以上代码中被观察者发射事件的速度是观察者处理速度的800倍，执行后就会抛出Caused by: rx.exceptions.MissingBackpressureException的异常。背压问题的出现需要两个条件：
- 观察者与被观察者需处于不同线程。
- 被观察者产生事件的速度需远快与观察者消费事件的速度。
由于观察者与被观察者处于不同线程，所以RxJava内部使用队列来存储事件，Android中默认队列缓存buffersize为16，所以当事件累计超过16个时就会抛出MissingBackpressureException的异常。解决这种问题就需要对被观察者进行流速控制了，而背压正是应对这种问题的一种策略。
背压策略
背压策略的解决思路便是响应式拉取。与RxJava观察者模型相反，响应式拉取是观察者主动去被观察者那里拉取事件，而被观察者则是被动等待通知再发射事件。
观察者需要多少事件就从被观察者那里拉取，而不是被动接收。这样实际上就实现了控制被观察者的流速，达到了背压策略的目的。
自绘结构图：
再来看一下代码示例：
//range操作符支持背压策略，发送事件的速度可被控制
Observable.range(1, 10000)
.observeOn(Schedulers.newThread())
.subscribe(new Subscriber<Integer>() {
@Override
public void onStart() {
//一定要在onStart中通知被观察者先发送一个事件
request(1);
}
@Override
public void onCompleted() {
Log.e("rx_test", "reactivePull：onCompleted");
}
@Override
public void onError(Throwable e) {
Log.e("rx_test", "reactivePull：onError：" + e.getMessage());
}
@Override
public void onNext(Integer n) {
try {
Thread.sleep(1000);
Log.e("rx_test", "reactivePull：onNext：" + n);
} catch (InterruptedException e) {
e.printStackTrace();
}
//处理完毕之后，再通知被观察者发送下一个事件
request(1);
}
}); 上述代码示例中，被观察者使用了range操作符发射10000次从1开始自增的数字，在观察者中首先于onStart()中使用request(1)向被观察者请求了第一个事件，之后在onNext()中每延时1000ms后输出日志，处理完事件后再调用request(1)请求一个新的事件。
输出结果：
reactivePull：onNext：1
...1s后...
reactivePull：onNext：2
...1s后...
reactivePull：onNext：3
...1s后...
reactivePull：onNext：4
...1s后...
reactivePull：onNext：5
......由输出结果可看出，每过1秒输出了一个数字，是不是实现了背压限流策略呢？需要多少事件就在观察者中使用request(n)主动拉取。不过RxJava1.x版本中并不是所有操作符都支持request(n)的响应式拉取，例如第一个例子中的interval操作符就不支持背压策略。而这个问题到了RxJava2.x中就得到了完美解决，且看下一篇。
其他解决方法
RxJava1中不支持背压策略的操作符如何解决背压问题呢？
过滤限流
通过使用限流操作符将被观察者产生的大部分事件过滤抛弃掉来达到限流的目的，间接降低事件发射的速度。
- sample：在一段时间内,只处理最后一个数据
- throttleFirst：在一段时间内,只处理第一个数据
-
debounce：发送一个数据,开始计时,到了规定时间内,若没有再发送数据,则开始处理数据,反之重新开始计时。
这里以sample操作符为例子：
//使用sample过滤操作符，每隔300ms取里时间点最近的事件发送
Observable.interval(1, TimeUnit.MILLISECONDS)
.observeOn(Schedulers.newThread())
.sample(300, TimeUnit.MILLISECONDS)
.subscribe((aLong) -> Log.e("rx_test", "controlByFilter：sample：" + aLong));输出结果：
controlByFilter：sample：280
controlByFilter：sample：577
controlByFilter：sample：878
controlByFilter：sample：1178
controlByFilter：sample：1478
controlByFilter：sample：1779
controlByFilter：sample：2078
controlByFilter：sample：2378
controlByFilter：sample：2673
......这种方式虽然实现了限流，但却是以抛弃大部分事件为代价的，在实际场景中并不可取，大家了解就好。
打包缓存
在被观察者发射事件过快，观察者来不及处理的情况下，可以使用缓存类的操作符将其中一部分打包缓存起来，再一点一点的处理其中的事件。
- buffer：将多个事件打包放入一个List中，再一起发射。
-
window：将多个事件打包放入一个Observable中，再一起发射。
这里以buffer操作符为例：
//使用buffer过滤操作符，将100ms内的事件打包为list发送
Observable.interval(1, TimeUnit.MILLISECONDS)
.observeOn(Schedulers.newThread())
.buffer(100, TimeUnit.MILLISECONDS)
.subscribe((longs) -> {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
Log.e("rx_test", "controlByCache：buffer：" + longs.size());
});输出结果：
controlByCache：buffer：79
controlByCache：buffer：1001
controlByCache：buffer：1002
controlByCache：buffer：1000
controlByCache：buffer：1001
controlByCache：buffer：1001
controlByCache：buffer：1001
controlByCache：buffer：1001
controlByCache：buffer：1001
controlByCache：buffer：1000
controlByCache：buffer：1002
controlByCache：buffer：1001背压操作符
RxJava1.x中，还有两种效果优于以上两种的操作符，可使不支持背压策略的操作符支持背压策略。
- onBackpressureDrop：将observable发送的事件抛弃掉，直到subscriber再次调用request(n)方法的时候，就发送给它这之后的n个事件。
-
onBackpressurebuffer：把observable发送出来的事件做缓存，当request(n)方法被调用的时候，给下层流发送一个item(如果给这个缓存区设置了大小，那么超过了这个大小就会抛出异常)。
以onBackpressureDrop为例：
Observable.interval(1, TimeUnit.MILLISECONDS)
.onBackpressureDrop()
.observeOn(Schedulers.newThread())
.subscribe(new Subscriber<Long>() {
@Override
public void onStart() {
super.onStart();
Log.e("rx_test", "controlBySpecialOperator：" + "onStart");
request(1);
}
@Override
public void onCompleted() {
Log.e("rx_test", "controlBySpecialOperator：" + "onCompleted");
}
@Override
public void onError(Throwable e) {
Log.e("rx_test", "controlBySpecialOperator：" + "onError");
}
@Override
public void onNext(Long aLong) {
Log.e("rx_test", "controlBySpecialOperator：onNext：" + aLong);
try {
Thread.sleep(500);
request(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});输出结果：
controlBySpecialOperator：onStart
controlBySpecialOperator：onNext：0
controlBySpecialOperator：onNext：1
controlBySpecialOperator：onNext：2
controlBySpecialOperator：onNext：3
controlBySpecialOperator：onNext：4
controlBySpecialOperator：onNext：5
controlBySpecialOperator：onNext：6
controlBySpecialOperator：onNext：7
controlBySpecialOperator：onNext：8
controlBySpecialOperator：onNext：9
controlBySpecialOperator：onNext：10
controlBySpecialOperator：onNext：11
controlBySpecialOperator：onNext：12
controlBySpecialOperator：onNext：13
controlBySpecialOperator：onNext：14
controlBySpecialOperator：onNext：15
controlBySpecialOperator：onNext：8014
controlBySpecialOperator：onNext：8015
controlBySpecialOperator：onNext：8016
controlBySpecialOperator：onNext：8017
controlBySpecialOperator：onNext：8018
controlBySpecialOperator：onNext：8019
controlBySpecialOperator：onNext：8020
controlBySpecialOperator：onNext：8021
controlBySpecialOperator：onNext：8022
...... 首先输出了0-15的数据，是因为observeOn操作符内部有一个长度为16的缓存区，它会首先请求16个事件缓存起来再输出。使用onBackpressureDrop可使不支持背压的操作符也可响应观察者的request(n)。
注意：需调用.onBackpressureDrop()方法。
总结
以上就是本篇关于RxJava1中存在的背压问题，背压策略的使用方法以及其他解决方法的讲解。背压在实际开发中遇到的不多，除非是大量数据，所以各位码友了解一下就好。本篇旨在为引入正式的Rxjava2做一个铺垫，敬请期待下一篇。
技术渣一枚，有写的不对的地方欢迎大神们留言指正，有什么疑惑或者建议也可以在我Github上RxJavaDemo项目Issues中提出，我会及时回复。
附上RxJavaDemo的地址：
RxJavaDemo