1.背压的产生

Backpressure（背压）是指在异步场景中，被观察者发送事件速度远快于观察者的处理速度的情况下，一种告诉上游的被观察者降低发送速度的策略。在 Observable/Observer 组合的使用中是不支持背压的，下面通过一个例子来看一下这种场景：

Observable.create(new ObservableOnSubscribe<Integer>() {@Overridepublic void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {int i = 0;while(i < Long.MAX_VALUE){e.onNext(i);i++;}}}).subscribeOn(Schedulers.newThread()).observeOn(Schedulers.newThread()).subscribe(new Consumer<Integer>() {@Overridepublic void accept(Integer integer) throws Exception {Thread.sleep(1000);Log.e("Test","i = "+integer);}});

2.Flowable

Flowable 就是为了解决背压问题的产物，因此才会把它们和 Observable/Observer 区分开来使用。
由于基于Flowable发射的数据流，以及对数据加工处理的各操作符都添加了背压支持，附加了额外的逻辑，其运行效率要比 Observable 低得多。
因为只有上下游运行在各自的线程中，且上游发射数据速度大于下游接收处理数据的速度时，才会产生背压问题。
所以，如果能够确定上下游在同一个线程中工作，或者上下游工作在不同的线程中，而下游处理数据的速度高于上游发射数据的速度，则不会产生背压问题，就没有必要使用 Flowable，以免影响性能。

Flowable.create(new FlowableOnSubscribe<Integer>() {@Overridepublic void subscribe(@NonNull FlowableEmitter<Integer> e) throws Exception {int i = 0;while(i < Long.MAX_VALUE){e.onNext(i);i++;}}
}, BackpressureStrategy.DROP).subscribeOn(Schedulers.io()).observeOn(Schedulers.newThread()).subscribe(new Subscriber<Integer>() {@Overridepublic void onSubscribe(Subscription s) {s.request(Long.MAX_VALUE);}@Overridepublic void onNext(Integer integer) {try {Thread.sleep(100);} catch (InterruptedException e) {e.printStackTrace();}Log.e("Test","i = "+integer);}@Overridepublic void onError(Throwable t) {}@Overridepublic void onComplete() {}});

这里注意三点：

Flowable.create 参数中多了个 BackpressureStrategy。
onSubscribe 回调的参数不是 Disposable 而是 Subscription。而且需要调用 Subscription.request 发起数据请求，否则Subscriber不会接受数据。
数据发射器是 FlowableEmitter 而不是 ObservableEmitter。

3.背压策略（BackpressureStrategy）

Flowable 的异步缓存池不同于 Observable，Observable的异步缓存池没有大小限制，可以无限制向里添加数据，直至OOM,而 Flowable 的异步缓存池有个固定容量，其大小为128。
BackpressureStrategy 的作用便是用来设置 Flowable 异步缓存池中的存储数据超限时的策略。
BackpressureStrategy 提供了一下几种背压策略：

MISSING：这种策略模式下相当于没有指定任何的背压策略，不会对数据做缓存或丢弃处理，需要下游通过背压操作符（onBackpressureBuffer()/onBackpressureDrop()/onBackpressureLatest()）指定背压策略。
ERROR：这种策略模式下如果缓存池中的数据超限了，则会抛出 MissingBackpressureException 异常
BUFFER：这种策略模式下没有为异步缓存池限制大小，可以无限制向里添加数据，不会抛出 MissingBackpressureException 异常，但会导致OOM。
DROP：这种策略模式下如果异步缓存池满了，会丢掉将要放入缓存池中的数据。
LATEST：这种策略模式下与 Drop 策略一样，如果缓存池满了，会丢掉将要放入缓存池中的数据，不同的是，不管缓存池的状态如何，LATEST都会将最后一条数据强行放入缓存池中。

参考：https://www.heqiangfly.com/2017/10/14/open-source-rxjava-guide-flowable/

7.2 Flowable 背压相关推荐

flowable背压取消_使用Flowable.generate（）生成可感知背压的流– RxJava常见问题解答...
flowable背压取消 RxJava缺少创建无限自然数流的工厂. 这样的流很有用,例如,当您想通过压缩两个事件的顺序来为可能的无限事件流分配唯一的序列号时: Flowable<Long> ...
使用Flowable.generate（）生成可感知背压的流– RxJava常见问题解答
RxJava缺少创建无限自然数流的工厂. 这样的流很有用,例如,当您想通过压缩两个事件的唯一序列号给可能的无限事件流时: Flowable<Long> naturalNumbers = / ...
rxjava背压_Android Rxjava ：最简单全面背压讲解 (Flowable)
1.前言阅读本文需要对Rxjava了解,如果还没有了解或者使用过Rxjava的兄die们,可以观看我另外一篇 Android Rxjava:不一样的诠释进行学习. Rxjava背压:被观察者发送事件 ...
如何形象地描述 RxJava 中的背压和流控机制？
之前我在知乎上受邀回答过一个关于RxJava背压(Backpressure)机制的问题,今天我把它整理出来,希望对更多的人能有帮助. RxJava的官方文档中对于背压(Backpressure)机制比 ...
Android RxJava 2.0中backpressure(背压)概念的理解
英文原文:https://github.com/ReactiveX/RxJava/wiki/Backpressure Backpressure(背压.反压力) 在rxjava中会经常遇到一种情况就是被 ...
rxjava背压_如何形象地描述RxJava中的背压和流控机制？
之前我在知乎上受邀回答过一个关于RxJava背压(Backpressure)机制的问题,今天我把它整理出来,希望对更多的人能有帮助. RxJava的官方文档中对于背压(Backpressure)机制比 ...
rxjava 背压_背压加载文件– RxJava常见问题解答
rxjava 背压事实证明,将文件作为流进行处理非常有效且方便. 许多人似乎忘记了,自Java 8(3年以上!)以来,我们可以很容易地将任何文件变成一行代码: String filePath = & ...
背压加载文件– RxJava常见问题解答
事实证明,将文件作为流进行处理非常有效且方便. 许多人似乎忘记了,自Java 8(3年多!)以来,我们可以很容易地将任何文件变成一行代码: String filePath = "foobar ...
RxJava之背压策略
转载请以链接形式标明出处: 本文出自:103style的博客本文基于 RxJava 2.x 版本目录 RxJava背压策略简介 Observable背压导致崩溃的原因 Flowable 使用介绍 ...
RxJava 2.0中backpressure(背压)概念的理解
英文原文:https://github.com/ReactiveX/RxJava/wiki/Backpressure Backpressure(背压.反压力) 在rxjava中会经常遇到一种情况就是被 ...

7.2 Flowable 背压

1.背压的产生

2.Flowable

3.背压策略（BackpressureStrategy）

7.2 Flowable 背压相关推荐

最新文章

热门文章