对RxJava中Backpressure概念的理解

英文原文地址请点击这里

一. Backpressure（背压、反压力）

在rxjava中会常常遇到一种状况就是被观察者发送消息太快以致于它的操做符或者订阅者不能及时处理相关的消息。那么随之而来的就是如何处理这些未处理的消息。

举个例子，使用 zip 操做符将两个无限大的Observable压缩在一块儿，其中一个被观察者发送消息的速度是另外一个的两倍。一个比较不靠谱的作法就是把发送比较快的消息缓存起来，当比较慢的Observable发送消息的时候取出来并将他们结合在一块儿。这样作就使得rxjava变得笨重并且十分占用系统资源。

在rxjava中有多重控制流以及背压（backpressure）策略用来应对当一个快速发送消息的被观察者遇到一个处理消息缓慢的观察者。下面的解释将会向你展现你应当怎么设计属于你本身的被观察者和操做符去应对流量控制（flow control）。

二. Hot and cold Observables, and multicasted Observables

Observable 数据流有两种类型：hot 和 cold。这两种类型有很大的不一样。本节介绍他们的区别，以及做为开发者应该如何正确的使用他们。

1. Cold observables

只有当有订阅者订阅的时候， Cold Observable 才开始执行发射数据流的代码。而且每一个订阅者订阅的时候都独立的执行一遍数据流代码。 Observable.interval 就是一个 Cold Observable。每个订阅者都会独立的收到他们的数据流。

咱们常常用到的Observable.create 就是 Cold Observable，而 just, range, timer 和 from 这些建立的一样是 Cold Observable。

2. Hot observables

Hot observable 无论有没有订阅者订阅，他们建立后就开发发射数据流。 一个比较好的示例就是鼠标事件。 无论系统有没有订阅者监听鼠标事件，鼠标事件一直在发生，当有订阅者订阅后，从订阅后的事件开始发送给这个订阅者，以前的事件这个订阅者是接受不到的；若是订阅者取消订阅了，鼠标事件依然继续发射。

了解更多Hot and cold Observables，参考这里

当一个cold observable是multicast（多路广播）（当转换完成时或者方法被调用）的时候，为了应对背压，应当把cold observable转换成hot observable。

cold observable 至关于响应式拉（就是observer处理完了一个事件就从observable拉取下一个事件），hot observable一般不能很好的处理响应式拉模型，但它倒是处理流量控制问题的不二候选人，例如使用onBackpressureBuffer或者onBackpressureDrop 操做符，和其余操做符比operations, throttling, buffers, or windows。

三. 能从某种意义上实现背压的运算符

防止过分建立observable的第一道防线就是使用普通数组去减小observable发送消息的数量，在这一节会使用一些操做符去应对突发的observable发送爆发性数据（一会没有，一会不少）就像下面的这张图片所示：

 

            

 

这些操做符能够经过微调参数确保slow-consuming观察者不被生产可观测的。

1. Throttling节流

操做符中好比 sample( ) 、 throttleLast( )、 throttleFirst( )、 throttleWithTimeout( ) 、 debounce( ) 容许你经过调节速率来改变Observable发射消息的速度。

如下图表展现如何使用这些操做符。

(1). sample (或 throttleLast)

sample 操做符按期收集observable发送的数据items，并发射出最后一个数据item。 

           

Observable<Integer> burstySampled = bursty.sample(500, TimeUnit.MILLISECONDS);

 

上面代码解释，按期且一次收集5个item，发射出最后一个item。

(2). throttleFirst

跟sample有点相似，可是并非把观测到的最后一个item发送出去，而是把该时间段第一个item发送出去。

 

            

Observable<Integer> burstyThrottled = bursty.throttleFirst(500, TimeUnit.MILLISECONDS);

 

(3). debounce (or throttleWithTimeout)

debounce操做符会只发送两个在规定间隔内的时间发送的序列的最后一个。

 

              

Observable<Integer> burstyDebounced = bursty.debounce(10, TimeUnit.MILLISECONDS);

 

(4). Buffers and windows 缓冲区和窗口

可使用操做符好比buffer( ) 或者window( ) 收集过分生成消息的Observable的数据项，而后发射出较少使用的数据。缓慢的消费者能够决定是否处理每一个集合中的某一个特定的项目，或处理集合中的某种组合，或为集合中的每一项预约计划工做，这都要视状况处理。

如下图表展现如何使用这些操做符。

• buffer

你能够按期关闭并释放突发性的 Observable 缓冲区。

 

            

Observable<List<Integer>> burstyBuffered = bursty.buffer(500, TimeUnit.MILLISECONDS);

 

在突发期间你能够获得的想要的，并在缓冲区收集数据和最终在突发结束的时候释放缓存。使用debounce操做符释放缓存并关闭指示器buffer操做符。

此段超过本人翻译水平，特提供原文以下，若有好的翻译建议请提出。 
Or you could get fancy, and collect items in buffers during the bursty periods and emit them at the end of each burst, by using the debounce operator to emit a buffer closing indicator to the buffer operator:

             

// we have to multicast the original bursty Observable so we can use it
// both as our source and as the source for our buffer closing selector:
Observable<Integer> burstyMulticast = bursty.publish().refCount();
// burstyDebounced will be our buffer closing selector:
Observable<Integer> burstyDebounced = burstMulticast.debounce(10, TimeUnit.MILLISECONDS);
// and this, finally, is the Observable of buffers we're interested in:
Observable<List<Integer>> burstyBuffered = burstyMulticast.buffer(burstyDebounced);

• window

与buffer相似，在一个window转换中容许你发送一个周期性的生成消息的Observable的数据项窗口：

 

             

Observable<Observable<Integer>> burstyWindowed = bursty.window(500, TimeUnit.MILLISECONDS);

当你每次从源被观察者收集了特定数量的数据项后也能够选择从新发送一个新的window:

               

Observable<Observable<Integer>> burstyWindowed = bursty.window(5);

 

四. 使用线程阻塞

处理过快生产item的其余策略就是使用线程阻塞，可是这么作违背了响应式设计和非阻塞模型设计，可是它的确是一个可行的选择。在rxJava中并无操做符能够作到这一点。

若是observable发送消息，subscriber消耗消息都是在同一个线程这将很好的处理这个问题，可是你要知道，在rxJava中，不少时候生产者和消费者都不在同一个线程。

 

五. 如何创建“响应式拉动（reactive pull）”backpressure

当subscribe订阅observable的时候能够经过调用subscribe.request（n），n是你想要的observable发送出来的量。

当在onNext()方法里处理完某个数据项（或一些数据项）后，你能从新调用 request()方法，通知Observable发射数据项。下面是个例子：

someObservable.subscribe(new Subscriber<t>() {
    @Override
    public void onStart() {
      request(1);
    }

    @Override
    public void onCompleted() {
      // gracefully handle sequence-complete
    }

    @Override
    public void onError(Throwable e) {
      // gracefully handle error
    }

    @Override
    public void onNext(t n) {
      // do something with the emitted item "n"
      // request another item:
      request(1);
    }
});

 

你能够经过一个神奇数字request， request(Long.MAX_VALUE)，禁用反应拉背力和要求Observable按照本身的步伐发射数据。request(0)是一个合法的调用,但不会奏效。请求值小于零的请求会致使抛出一个异常。

1. Reactive pull backpressure isn’t magic

backpressure 不会使得过分生产的observable的问题消失，这只是提供了一种更好的解决问题的方法。 让咱们更仔细的研究刚刚说到的zip操做符的问题。

这里有两个observable，a和b，b发射item比a更加的频繁，当你想zip这两个observable的时候，你须要把a发送出来的第n个和b发送出来的第n个对象处理，然而因为b发送出来的速率更快，这时候b已经发送出了n+1~n+m个消息了，这时候你要想要把a的n+1~n+m个消息结合的话，就必须持有b已经发送出来的n+1~n+m消息，同时，这意味着缓存的数量在不断的增加。

固然你能够给b添加操做符throttling，可是这意味着你将丢失某些从b发送出来的项，你真正想要作的其实就是告诉b：“b你须要慢下来，可是你要保持你给个人数据是完整的”。

响应式拉（reective pull）模型能够帮你作到这一点，subscriber从observable那里拉取数据，这与一般状况下从observable那里推送数据这种模式相比造成鲜明的对比。

在rxJava中，zip操做符正是使用了这种技巧。它给每一个源observable维护了一个小的缓存池，当它的缓存池满了之后，它将不会从源observable那里拉取item。每当zip发送一个item的时候，他从它的缓存池里面移除相应的项，并从源observable那里拉取下一个项。

在rxJava中，不少操做符都使用了这种模式（响应式拉），可是有的操做符并无使用这种模式，由于他们也许执行的操做跟源observable处于相同的进程。在这种状况下，因为消耗事件会阻塞本进程，因此这一项的工做完成后，才有机会收到下一项。还有另一种状况，backpressure也是不适合的，由于他们有指定的其余方式去处理流量控制，这些特殊的状况在rxJava的javadocs里面都会有详细说明。

可是，observable a和b必须正确的响应request()方法，若是一个observable尚未被支持响应式拉（并非每一个observable都会支持），你能够采起如下其中一种操做均可以达到backpressure的行为：

(1). onBackpressurebuffer

为全部从源observable发送出来的数据项维护一个缓存区，根据他们生成的request发送给下层流。

 

             

这个操做符还有一个实验性的版本容许去设置这个缓存池的大小，但当缓存池满了之后将会终止执行并抛出异常。

(2). onBackpressureDrop

终止发送来自源observable的事件，除非来自下层流的subscriber即将调用request（n）方法的时候，此时才会发送足够的数据项给以知足requst。

             

(3). onBackpressureBlock (实验性的,   不支持RxJava 1.0)

阻塞源Observable正在操做的线程的线程直到某个Subscriber发出请求，而后只要有即将发出的请求就结束阻塞。

 

             

若是你不容许这些操做符操做用在一个不支持背压的Observable上，而且 若是做为Subscriber的你或者在你和Observable之间的一些操做符尝试去应用响应式拉背压，你将会在onError回调事件中遭遇 MissBackpresssureException的警告。

六. Flowable与Observable

最后，为了你们更好的理解backpressure概念，这里补充说一下Flowable。

Observable在RxJava2.0中新的实现叫作Flowable， 同时旧的Observable也保留了。由于在 RxJava1.x 中，有不少事件不被能正确的背压，从而抛出MissingBackpressureException。

举个简单的例子，在 RxJava1.x 中的 observeOn， 由于是切换了消费者的线程，所以内部实现用队列存储事件。在 Android 中默认的 buffersize 大小是16，所以当消费比生产慢时， 队列中的数目积累到超过16个，就会抛出MissingBackpressureException， 初学者很难明白为何会这样，使得学习曲线异常得陡峭。

而在2.0 中，Observable 再也不支持背压，而Flowable 支持非阻塞式的背压。Flowable是RxJava2.0中专门用于应对背压（Backpressure）问题。所谓背压，即生产者的速度大于消费者的速度带来的问题，好比在Android中常见的点击事件，点击过快则常常会形成点击两次的效果。其中，Flowable默认队列大小为128。而且规范要求，全部的操做符强制支持背压。幸运的是， Flowable 中的操做符大多与旧有的 Observable 相似。