RxJava2 中的背压
在RxJava2里，引入了Flowable这个类来处理backpressure，而Observable不包含backpressure处理。Flowable的三种Backpressure策略：
- BackpressureStrategy.BUFFER
onBackpressureBuffer是不丢弃数据的处理方式。把上游收到的全部缓存下来，等下游来请求再发给下游。相当于一个水库。但上游太快，水库（buffer）就会溢出。 - BackpressureStrategy.DROP 与 BackpressureStrategy.LATEST
Drop 和Latest 类似，都会丢弃数据，下游通过request请求产生令牌给上游，上游接收到多少令牌，就发送多少，当令牌为0的时候，上游开始丢弃数据。区别在于，drop直接丢弃数据不缓存数据。而latest缓存最新的一条数据，当上游收到令牌，就把缓存的上一条“最新”数据发送给下游。
例如 ：
Flowable<Integer> flowable =
Flowable.create((FlowableOnSubscribe<Integer>) e -> {
for (int i = 0; i < Integer.MAX_VALUE; i++) {
Log.d(TAG, "onNext : " + i);
e.onNext(i);
Thread.sleep(10);
}
}, BackpressureStrategy.DROP).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread());
Flowable以10毫秒一次派发数据，注意我们让Flowable和订阅者运行在不同的线程，这样才能看出生产与消费在不同效率下时的差异性，如果Flowable和订阅者在同一线程，背压是没什么意义的。假设订阅他们的方法都是:
Subscription mSubscription;
flowable.subscribe(new Subscriber<Integer>() {
@Override
public void onSubscribe(Subscription subscription) {
mSubscription = subscription;
}
@Override
public void onNext(Integer integer) {
Log.d(TAG, "onNext: " + integer);
}
@Override
public void onError(Throwable throwable) {
}
@Override
public void onComplete() {
}
});
我们在onSubscribe中保存了Subscription ，以后 方便我们可以在任何时候request 数据。我们添加一个按钮，以实现手动request数据 ，代码如下：
if(mSubscription != null) {
mSubscription.request(64);
}
我们一开始request 64个数据，我们启动Flowable后，隔一段时间才点击request , log打印 0 ~ 63 ：
D/SimpleExampleActivity: onNext: 0
D/SimpleExampleActivity: onNext: 1
D/SimpleExampleActivity: onNext: 2
...
D/SimpleExampleActivity: onNext: 63
D/SimpleExampleActivity: onNext: 64
隔一段时间第二次点击request , log打印 64~ 127 ：
D/SimpleExampleActivity: onNext: 64
D/SimpleExampleActivity: onNext: 65
D/SimpleExampleActivity: onNext: 66
...
D/SimpleExampleActivity: onNext: 126
D/SimpleExampleActivity: onNext: 127
隔一段时间第三次点击request , log打印 1243~ 1306：
D/SimpleExampleActivity: onNext: 1243
D/SimpleExampleActivity: onNext: 1244
D/SimpleExampleActivity: onNext: 1245
...
D/SimpleExampleActivity: onNext: 1304
D/SimpleExampleActivity: onNext: 1305
D/SimpleExampleActivity: onNext: 1306
我们使用是BackpressureStrategy.DROP ， 与就是 直接丢弃数据不缓存数据 。可是我们一开始隔了点时间再request时，还是打印从 0 ~ 127 ， 这说明 Flowable 本身就会存储缓存 128 个数据，超过128个后执行我们的策略，也就是丢弃。所以 1243~ 1306 其实是我们在第二次点击后，重新缓存的128数据。如果我们换成 BackpressureStrategy.BUFFER ， 那么不管你点击多少次，数据都是连续的，因为 BackpressureStrategy.BUFFER 策略会把数据一直放到内存中，直到发生OutOfMemoryError。
我们现在修改request 的数目 ，改成 每次 request 96 个 ， 代码如下 ：
if(mSubscription != null) {
mSubscription.request(96);
}
第一次点击request , log打印 0 ~ 95 , 没什么问题
D/SimpleExampleActivity: onNext: 0
D/SimpleExampleActivity: onNext: 1
D/SimpleExampleActivity: onNext: 2
...
D/SimpleExampleActivity: onNext: 94
D/SimpleExampleActivity: onNext: 95
隔一段时间第二次点击request , log打印 96 ~ 127 , 188 ~ 251 :
D/SimpleExampleActivity: onNext: 96
D/SimpleExampleActivity: onNext: 97
D/SimpleExampleActivity: onNext: 98
...
D/SimpleExampleActivity: onNext: 126
D/SimpleExampleActivity: onNext: 127
D/SimpleExampleActivity: onNext: 188
D/SimpleExampleActivity: onNext: 189
...
D/SimpleExampleActivity: onNext: 250
D/SimpleExampleActivity: onNext: 251
隔一段时间第三次点击request , log打印 252 ~ 283 , 650 ~ 713:
D/SimpleExampleActivity: onNext: 252
D/SimpleExampleActivity: onNext: 253
...
D/SimpleExampleActivity: onNext: 282
D/SimpleExampleActivity: onNext: 283
D/SimpleExampleActivity: onNext: 650
D/SimpleExampleActivity: onNext: 651
...
D/SimpleExampleActivity: onNext: 712
D/SimpleExampleActivity: onNext: 713
我们可以看到第二次、第三次时打印的数据出现了中断的情况。我们知道Flowable 默认会缓存 127个数据，那么第一次点击之后应该剩下 128 - 96 = 32个 ， 所以第二次首先打印 96 ~ 127 ， 之后再打印 188 ~ 251 64个数据。第三次又打印了 252 ~ 283 32个数据。第二次打印中断之后打印的 64个数据 加上 第三次打印中断前打印的 32个，刚好是 96个数据，也就是打印中断的时间点的数据刚好是96个。
这个96就是Flowable 重新去拉取缓存的限制，这是在源码上设定的，就是说首先缓存了 128个数据之后，被消费了96个数据时才会重新缓存。所以在第二次时，从127后就打印了 188，因为这个188是在第一次点击之后就重新缓存了。
总结
Flowable 有三种Backpressure策略，分别是BackpressureStrategy.BUFFER、BackpressureStrategy.DROP 和 BackpressureStrategy.LATEST。默认会缓存 127个数据，被消费了96个数据后才会重新缓存。
