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个数据后才会重新缓存。