了解背压的看管请继续右转您内,请一定看最末，你会发现你其他地方没见过的东西。

背压二三事

不知大家还记不记得在看（一）的时候，当时为了讲解Flowable的create方法时，第二个参数backpressure，我们直接选择了跳过，用了枚举中的其中一个。

• 引入

但是你有没有想过，有没有这么一种情况，上游的选手的速度非常的快，在拼命的往下游扔，而下游的选手速度没那么快，来不及处理上游扔下来的那么多东西。

再就比如我们上一节使用的zip，上游有两个选手，一个选手速度特别快，一个选手速度特别慢（反正没那么快就行），那么中游的拦截者zip在拦截的时候，他会先把上游速度快的那家伙扔下来的东西，统统收集起来，等到速度慢的那个的第一个东西到了再把速度快的第一个东西跟它打包扔下去。（不知道你可以理解么。。）
 
那么这个时候你哟偶没有想过，太多的东西放在下游或者是中游会不会导致放不下了，崩掉呢。

其实我们所说的背压就是这种上游和下游（其实中游也是一样的）速度产生了差异所导致的，系统不得不去对上游的大量的请求做缓存而导致了内存崩溃，让我们继续往下看

• 实验

我们先给出如下的代码：

Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                for(;;){
                    e.onNext("hello on next");
                }
            }
        });
observable.subscribe(new Consumer<String>() {
            @Override
            public void accept(@NonNull String s) throws Exception {
                Log.e("Consumer", "onNext    :    s  " + s);
            }
        });

我们发现，完全没有问题，上下如果都是超音速的选手，那么控制台就会打出一大坨的log，但是完全是流畅的运行哦，没有任何问题哦。

于是，我们把下游的选手的速度变慢了（这里要注意，其实背压的问题仅限于上游的选手速度比下游快很多，毕竟是上游扔东西），我们把下游的编一个稍微缓慢一点的选手：

observable.subscribe(new Consumer<String>() {
            @Override
            public void accept(@NonNull String s) throws Exception {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                Log.e("Consumer", "onNext    :    s  " + s);
            }
        });

让他一秒钟收一个包裹，然后我很开心的按下了build键。。。。

<br /> <br /> <br /> <br /> <br /> <br /> <br /> <br /><br /><br /><br /><br /><br />
你会奇怪我干嘛去了，我会告诉你，看到输出的log信息，我震惊了。它不急不慢的给我每隔一秒钟输出一条信息来。并没看到任何崩溃甚至是卡顿的趋势。。我在想是不是就此封笔，跟大家说再见。
还是先让我，开包辣条，先缓缓。。

经过一段时间的格叽格叽之后，我发现上游和下游是在一个线程中的，要是让他们在不同的线程呢？

OK,各位可以开始数秒蹦了，当你数到10秒左右（视机子情况），你会看到可爱的dialog谈了出来。

 那么这个怎么又从`上下游方法`去分析呢，我们说上游的选手比下游的速度快上很多，如果这两个大兄弟是一个公司（线程）的，那么上有的即使速度再快他也会等下游的大兄弟，处理完了再发下一个；但是如果他俩不是一个公司的，那我的任务我速度我快做完了就好了，我管你下游。。。

• 天将降大任于斯人也

那么我们怎么去尽量减少背压给我们带来的伤害呢。

首先上面的这种方法肯定是可行的，我们让上下游在一个线程中去工作，就不会出现这种背压的问题。但是既然我们使用了响应式变成，如果你依然让他们在同一个线程中去工作的话，那意义就变得不是很大了。我们来看一下，如何去限制两个不同公司的选手。答案就是中游的拦截者——操作符

* 苦其心志
  如果你还记得我们在[(三)](http://www.jianshu.com/p/96df60fd35c6)中介绍的操作符知识的话，那么你应该可以想到我们可以想到的缓解背压的方法。
 我们可以使用flat方法来筛选上游扔下来的包裹，你扔好了，我只挑我要的，可能你扔了1000个，经过我的筛选，我只留了10个，那你想，压力是不是瞬间就变小了。

 * 饿其体肤
   使用sample进行采样，我们每隔一段时间去取一个包裹，你就扔吧，你扔的再多我就每隔一段时间取一个，随便你。

• 缓解疼痛

很爽了把，反正最起码现在系统是不会崩溃的。

不过话还是说回来，这么做终归还是指标不治本的，因为你丢弃掉了上游扔下来的很多包裹，说不定这些被你丢弃掉的包裹中有one_piece哦。那么如果你想这些包裹你一个都不想丢弃掉又不想让系统崩溃的话就要回过头来真正的来认识一下我的朋友backpressure。

• 你好，背压哥！

我可以告诉你，背压是个帮派！背压不是一个哥，共有五个背压哥：
```
/**

• 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.
  /
  MISSING,
  /*
  * Signals a MissingBackpressureException in case the downstream can't keep up.
  /
  ERROR,
  /*
  * Buffers <em>all</em> onNext values until the downstream consumes it.
  /
  BUFFER,
  /*
  * Drops the most recent onNext value if the downstream can't keep up.
  /
  DROP,
  /*
  * Keeps only the latest onNext value, overwriting any previous value if the
  * downstream can't keep up.
  */
  LATEST
  }

    一手抖，把英文注释都贴出来了，我知道大家不是很喜欢看英文的注释，待我们一一的先去认识这五位大哥。
       * MISSING
        顾名思义，就是当你上游一股脑的抛下来很多东西的时候，我就只接一部分，其他的我都miss掉。
       * ERROR
         当下游无法再继续接受请求的时候会抛出异常MissingBackpressureException
        上游，你拼命抛吧，我这边要是拿不下了，我就报警，老子不干了！
       * BUFFER
         上游不断的发出onNext请求，直到下游处理完
         上游，你抛吧，我就屯着，实在不行了，系统崩溃，咱俩同归于尽
       * DROP
         如果下游已经溢出了，会丢弃掉溢出了的onNext请求。
         上游，你就拼命抛吧，我这里就拿那么多，我处理完了我再去你最新的里面拿
       * LATEST
         如果下游已经溢出了，会丢弃掉溢出了的onNext请求，但是会在内部缓存一个最新的onNext请求，在下一次请求的时候会把这个最新的先返回出去。
        
  大体上看来BUFFER最好理解，MISSING和ERROR，DROP和LATEST你如果不去自己写两段代码测试一下，你绝壁不知道到底哪里不一样了。
  
     * **MISSING和ERROR的区别**
       先看一下我们的测试demo：

Flowable<Integer> flowable = Flowable.create(new FlowableOnSubscribe<Integer>() {
@Override
public void subscribe(FlowableEmitter<Integer> e) throws Exception {
//每隔100毫秒发送一个事件，总共发送400个
for(int i = 0; i < 400; i++){
Thread.sleep(100);
Log.e("subscribe", "i : " + i);
e.onNext(i);
}
}
}, BackpressureStrategy.MISSING);
flowable.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Subscriber<Integer>() {
@Override
public void onSubscribe(Subscription s) {
//获取到Subscription对象，拿到外面来对其进行操作
subscription = s;
}

        @Override
        public void onNext(Integer integer) {
            Log.e("onNext" ,"   i  ：    " + integer);
        }

        @Override
        public void onError(Throwable t) {
            Log.e("onError" ,"   onError  ：    ");
            t.printStackTrace();
        }
        @Override
        public void onComplete() {

        }
    });

   然后我们来跑一下我们的demo，如果你写的没有问题的话，你看到logcat，应该是如下这样的：

02-22 12:53:37.312 4143-4184/org.ding.testmulti E/subscribe: i : 126
02-22 12:53:37.413 4143-4184/org.ding.testmulti E/subscribe: i : 127
02-22 12:53:37.516 4143-4184/org.ding.testmulti E/subscribe: i : 128
02-22 12:53:37.516 4143-4143/org.ding.testmulti E/onError: onError ：
02-22 12:53:37.516 4143-4143/org.ding.testmulti W/System.err: io.reactivex.exceptions.MissingBackpressureException: Queue is full?!
02-22 12:53:37.517 4143-4143/org.ding.testmulti W/System.err: at io.reactivex.internal.operators.flowable.FlowableObserveOn$BaseObserveOnSubscriber.onNext(FlowableObserveOn.java:114)
02-22 12:53:37.517 4143-4143/org.ding.testmulti W/System.err: at io.reactivex.internal.operators.flowable.FlowableSubscribeOn$SubscribeOnSubscriber.onNext(FlowableSubscribeOn.java:97)
02-22 12:53:37.517 4143-4143/org.ding.testmulti W/System.err: at io.reactivex.internal.operators.flowable.FlowableCreate$MissingEmitter.onNext(FlowableCreate.java:338)
02-22 12:53:37.517 4143-4143/org.ding.testmulti W/System.err: at org.ding.testmulti.MainActivity$7.subscribe(MainActivity.java:197)
02-22 12:53:37.517 4143-4143/org.ding.testmulti W/System.err: at io.reactivex.internal.operators.flowable.FlowableCreate.subscribeActual(FlowableCreate.java:72)
02-22 12:53:37.517 4143-4143/org.ding.testmulti W/System.err: at io.reactivex.Flowable.subscribe(Flowable.java:12901)
02-22 12:53:37.517 4143-4143/org.ding.testmulti W/System.err: at io.reactivex.internal.operators.flowable.FlowableSubscribeOn$SubscribeOnSubscriber.run(FlowableSubscribeOn.java:82)
02-22 12:53:37.517 4143-4143/org.ding.testmulti W/System.err: at io.reactivex.internal.schedulers.ScheduledRunnable.run(ScheduledRunnable.java:59)
02-22 12:53:37.517 4143-4143/org.ding.testmulti W/System.err: at io.reactivex.internal.schedulers.ScheduledRunnable.call(ScheduledRunnable.java:51)
02-22 12:53:37.517 4143-4143/org.ding.testmulti W/System.err: at java.util.concurrent.FutureTask.run(FutureTask.java:237)
02-22 12:53:37.517 4143-4143/org.ding.testmulti W/System.err: at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:272)
02-22 12:53:37.517 4143-4143/org.ding.testmulti W/System.err: at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1133)
02-22 12:53:37.517 4143-4143/org.ding.testmulti W/System.err: at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:607)
02-22 12:53:37.517 4143-4143/org.ding.testmulti W/System.err: at java.lang.Thread.run(Thread.java:761)

紧接着我们把MISSING换成ERROR再看看，你会发现，哎哟喂。。。。是一样的。。。
但是你如果看的稍微仔细点，你会发现具体的error明细是不一样的，是这样的

could not emit value due to lack of requests

然而并没有什么*用，怎么办，就这么好了？

         * 为什么是128
我们先一样一样来，首先为什么是128，如果你比较喜欢点看源码，你会好奇的点进Flowable源码看一下，首先映入眼帘的就是如下这一坨：

/** The default buffer size. */
static final int BUFFER_SIZE;
static {
BUFFER_SIZE = Math.max(16, Integer.getInteger("rx2.buffer-size", 128));
}

    默认的buffer大小就是128，Flowable将它用来存储上游的事件。那么我们就可以理解了，因为上游发了128个事件将队列撑爆了。

         * 到底有什么不一样
           除了两个error的明细不一样之外，实在是看不出还有哪里不同了。

         我就又回去读了一下MISSING的英文备注，写道`OnNext events are written without any buffering or dropping`，onNext的事件消耗既不缓存又不丢弃。那么反过来讲，它既然着重这么说，那就说明ERROR会做缓存或者是丢弃，由于DROP和LATEST的重点是丢弃，那ERROR应该就和缓存有关。
       也许你会问我，为什么不去源码里看个究竟，整个系列主要面向的初学者，教会大家使用的优先级要远高于探个究竟。能够想到去源码里深究的各位，不要停止你们的步伐。为了不把整体阅读难度抬高，基本都是能简单就简单，故而这里选择了从注释出发。
       
        那好，既然我们猜他和onNext缓存有关系，那我就先消耗掉几个onNext看看，我们在onSubscribe方法中，先去request掉几个事件。

@Override
public void onSubscribe(Subscription s) {
subscription = s;
s.request(20);
}

 我们选择先消耗掉20个事件，再来看一下两种类型的log。你会发现ERROR依然是128，MISSING变成了148！！
         * 结论
     MISSING维护的队列，你如果request掉了，就不会再对这部分有缓存，会继续往这个队列里面存，直到存满128个事件，告诉你队列满了。
      ERROR只存128个，只要一存满，立马抛异常，除非你可以把这128个请求完全消耗掉。

        那么形象一点，基于`上下游方法`方法来分析：
     MISSING这位背压哥，可以一个人扛128个包裹。上游一下子扔了400个包裹，下游立马从MISSING身上拿了20个立马解决掉（1~20），MISSING便继续往下接了20个（21~148），扛不住了，倒掉。
      ERROR这位背压哥，也是一个人可以扛128个包裹，上游扔了400个包裹，下游说，来，我先拆个20个。ERROR说，可以，你拆归拆，但你要把我这128个都拆完，我才会再去扛128个。所以即使下游拆了20个，但实际上那20个依然还在ERROR身上。最后，ERROR扛到第129个，卒。

     * **DROP和LATEST的区别**
       只看字面的注释，你完全无法猜透，我们依然通过上面的demo去测试，删除刚才提前request代码，并且搞一个按钮逻辑：

public void click(View view){
subscription.request(129);
}

        这里为什么用129呢，我可以告诉你，我之前用的是10。然后我发现无论是DROP还是LATEST，最后返回给我的都是一样的。
        因为我们先分析了ERROR和MISSING，所以我们大致可以猜到了，这两个都是对onNext是有缓存的，也就是说你要么把128个都消耗掉，你不消耗掉，我就不去存新的。
       那么这个时候我们消耗129个是什么意思呢，拆解一下，就是128 + 1，一下子消耗128个就会把队列中的清空掉，会让背压哥再去扛（如果有的话）
        
       一开始，这边没有等400个请求都发送完，我便按下了按钮，做了request操作，发现两者的结果是一样的，比如当前正好发送第300个，那么两者的结果都是126，127，300。DROP确实把美誉接收到的丢弃了，LATEST确实又是给了你一个最新的，没毛病啊。
       这样我们基本看不到区别，如果换一种思路，我们直接等400个发送完毕，再去做request请求。
       好的我们先来看一下，两个背压大哥分别返回的log日志：
DROP:

02-22 14:55:13.111 32729-32729/org.ding.testmulti E/onNext: i ： 123
02-22 14:55:13.111 32729-32729/org.ding.testmulti E/onNext: i ： 124
02-22 14:55:13.111 32729-32729/org.ding.testmulti E/onNext: i ： 125
02-22 14:55:13.111 32729-32729/org.ding.testmulti E/onNext: i ： 126
02-22 14:55:13.111 32729-32729/org.ding.testmulti E/onNext: i ： 127

LATEST:

02-22 15:02:10.379 3664-3664/org.ding.testmulti E/onNext: i ： 125
02-22 15:02:10.379 3664-3664/org.ding.testmulti E/onNext: i ： 126
02-22 15:02:10.379 3664-3664/org.ding.testmulti E/onNext: i ： 127
02-22 15:02:10.379 3664-3664/org.ding.testmulti E/onNext: i ： 399

明了了吧，你问DROP要129个，他只给了你128个（0~127），因为你问他要的时候已经请求结束了，他只会把它自己保存的128个统统给你，还有的他都没有接收到。
你问LATEST要129个，他先把128个都给你，他自己还存了一个最新的第400个（399）给你。
这两个背压哥，我们从形象的去分析可以这样理解：

       DROP哥：
身上扛128个包裹，你把都拆完，我再继续接，注意了！！！是继续**接**，也就是说你把我身上的128个都拆完，我就把这128个拆完的包裹都扔掉，继续接上游扔下来的。

      LATEST哥：
身上扛128个包裹，并且自己有个小口袋，你上游扔一个我就往我小口袋里放一个，你上游扔一个我就换一个，这个小口袋里老是放的是上游扔下来的最新的包裹。最后你问我要几个我就先从身上给你，如果128个我就把身上128个都给你，如果超过128个我就先把口袋里的给你，再去摊开双手继续接上游扔下来的。

-----
 ##有个巨坑哦！！
  首先，Flowable的默认BUFFER_SIZE是128这个没有问题。
     但是问题是，如果一下子把128个都清空掉，那么下一次再往里面填是填多少个！！！！
      
  是128个么？？是么？？反正我这儿不是。大伙可以去试一下，第一次在onSubscribe方法中直接request(128)个之后，Flowable会再去往缓存中存请求，我这边经过验证存的是**96个！！！**

那么第二次存96个会有什么问题呢

   * 如果你使用的是ERROR：
     第一次把缓存中的128个完全用掉，促使缓存再次接收请求，第二次缓存存满96个就抛异常

   * 如果你使用的是DROP：
     第一次把缓存中的128个用光，第二次只存96个，之后的我都抛弃掉，我都不要

   * 如果你使用的是LATEST：
    第一次先存满128个，并且在第129个存当前的最新值。如果把128个都用光，导致再次开启缓存，那么会再存96个，并且在第97个存最新值。
    如果你第二次request(96),那么就只把那存的96个返回给你。如果你请求96个以上，那就把第97个最新的那个也返回给你

   * MISSING
    不受任何影响，它的队列大小永远是128。

-------

##电梯
  [抱着陌生的态度再看Rxjava（一）](http://www.jianshu.com/p/fbfdd6fa6154)   
  [抱着陌生的态度再看Rxjava（二）](http://www.jianshu.com/p/f435dc51221f)
    [抱着陌生的态度再看Rxjava（三）](http://www.jianshu.com/p/96df60fd35c6)