rxjava : Flowable 和Backpressure背压参考：BackpressureStrategy背压策略

98 阅读 0 评论 65 点赞

我是靠谱客的博主现实帅哥，最近开发中收集的这篇文章主要介绍rxjava : Flowable 和Backpressure背压参考：BackpressureStrategy背压策略，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

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

Observable在 subscribe方法中调用emitter.onNext()才能触发之后的onNext方法；Flowable也相同，也需要一个通知，这里发挥作用的就是subscription.request(n)方法,参数n代表上游发送多少个数据。

参考：

给初学者的RxJava2.0教程(五)

同步和异步的区别仅仅在于是否有水缸

当上下游工作在同一个线程中时, 这时候是一个同步的订阅关系, 也就是说上游每发送一个事件必须等到下游接收处理完了以后才能接着发送下一个事件.

当上下游工作在不同的线程中时, 这时候是一个异步的订阅关系, 这个时候上游发送数据不需要等待下游接收, 为什么呢, 因为两个线程并不能直接进行通信, 因此上游发送的事件并不能直接到下游里去, 这个时候就需要一个田螺姑娘来帮助它们俩, 这个田螺姑娘就是我们刚才说的水缸 ! 上游把事件发送到水缸里去, 下游从水缸里取出事件来处理, 因此, 当上游发事件的速度太快, 下游取事件的速度太慢, 水缸就会迅速装满, 然后溢出来, 最后就OOM了.

给初学者的RxJava2.0教程(六)

异步情况下上下游流速不均衡的问题治理的办法就两种:

一是从数量上进行治理, 减少发送进水缸里的事件

二是从速度上进行治理, 减缓事件发送进水缸的速度

给初学者的RxJava2.0教程(七)

响应式拉取:

Flowable在设计的时候采用了一种新的思路也就是响应式拉取的方式来更好的解决上下游流速不均衡的问题, 与我们之前所讲的控制数量和控制速度不太一样, 这种方式用通俗易懂的话来说就好比是叶问打鬼子, 我们把上游看成小日本, 把下游当作叶问, 当调用Subscription.request(1)时, 叶问就说我要打一个! 然后小日本就拿出一个鬼子给叶问, 让他打, 等叶问打死这个鬼子之后, 再次调用request(10), 叶问就又说我要打十个! 然后小日本又派出十个鬼子给叶问, 然后就在边上看热闹, 看叶问能不能打死十个鬼子, 等叶问打死十个鬼子后再继续要鬼子接着打…

默认水缸：

在Flowable里默认有一个大小为128的水缸, 当上下游工作在不同的线程中时, 上游就会先把事件发送到这个水缸中, 因此, 下游虽然没有调用request, 但是上游在水缸中保存着这些事件, 只有当下游调用request时, 才从水缸里取出事件发给下游.

给初学者的RxJava2.0教程(八)

BackpressureStrategy背压策略

/**
* Represents the options for applying backpressure to a source sequence.
* 表示将背压应用于源序列的选项
*/
public enum BackpressureStrategy {
/**
* OnNext events are written without any buffering or dropping.
* Downstream has to deal with any overflow.
* <p>Useful when one applies one of the custom-parameter onBackpressureXXX operators.
* 写入OnNext事件时不会进行任何缓冲或删除。
* 下游必须处理任何溢出。
* 在将一个自定义参数应用于BackpressureXXX运算符时有用。
*/
MISSING,
/**
* Signals a MissingBackpressureException in case the downstream can't keep up.
* 指示MissingBackpressureException，以防下游无法跟上。
*/
ERROR,
/**
* Buffers <em>all</em> onNext values until the downstream consumes it.
* 缓冲all onNext值，直到下游使用它 ，【BUFFER_SIZE默认为128】
* 这时的FLowable表现出来的特性的确和Observable一模一样，需要注意OOM
*/
BUFFER,
/**
* Drops the most recent onNext value if the downstream can't keep up.
* 如果下游无法跟上，则丢弃最新的onNext值。
* Drop就是直接把存不下的事件丢弃,
*/
DROP,
/**
* Keeps only the latest onNext value, overwriting any previous value if the
* downstream can't keep up.
* 仅保留最新的onNext值，如果下游不能跟上，则覆盖任何先前的值。
* Latest就是只保留最新的事件
*/
LATEST
}