深入浅出RxJava2.0_01[重新理解背压概念]

1.RxJava1.0缺陷之请求太多来不及处理

RxJava是一个观察者模式的架构，当这个架构中被观察者(Observable)和观察者(Subscriber)处在不同的线程环境中时，由于者各自的工作量不一样，导致它们产生事件和处理事件的速度不一样，这就会出现两种情况：

被观察者产生事件慢一些，观察者处理事件很快。那么观察者就会等着被观察者发送事件，（好比观察者在等报社出新的报纸，程序等待，这没有问题）。

被观察者产生事件的速度很快，而观察者处理很慢。那就出问题了，如果不作处理的话，事件会堆积起来，最终挤爆你的内存，导致程序崩溃。（好比被观察者生产的报纸没人去看，报社很容易倒闭）。

Observable<Integer> observable =
Observable.create(new Observable.OnSubscribe<Integer>() {
@Override
public void call(Subscriber<? super Integer> subscriber) {
for (int i = 0; i < 1000; i++) {
subscriber.onNext(i);
}
}
})
//让观察者在新的线程中运行 这样 观察者与被观察者就运行在不同的线程中了
.observeOn(Schedulers.newThread());

observable.subscribe(new Subscriber<Integer>() {

@Override
public void onNext(Integer value) {
try {
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
Log.i(TAG, "call: " + value);
}

@Override
public void onCompleted() {

}

@Override
public void onError(Throwable e) {
Log.i(TAG, "onError: "+e);
}

});

上面的代码运行结果如下：

03-16 16:26:24.931 3480-3496/? I/IT520: call: rx.exceptions.MissingBackpressureException

上面的事例中，在一个线程中被观察者发送1000个整型数据，在另一个线程中没有及时处理，而经过0.5秒之后，观察者处理了信息（也就是打印了数据）。最后引发了异常MissingBackpressureException：告诉开发者，观察者来不及处理被观察者发送的信息。

而这个MissingBackpressureException异常里面就包含了Backpressure这个单词，看来背压肯定和这种异常情况有关系。

2.背压概念

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

这里先介绍一下响应式拉取：

在RxJava的观察者模型中，被观察者是主动的推送数据给观察者，观察者是被动接收的。而响应式拉取则反过来，观察者主动从被观察者那里去拉取数据，而被观察者变成被动的等待通知再发送数据。

Observable<Integer> observable =
Observable.range(1,1000)
//让观察者在新的线程中运行 这样 观察者与被观察者就运行在不同的线程中了
.observeOn(Schedulers.newThread());

observable.subscribe(new Subscriber<Integer>() {

@Override
public void onStart() {
//向被观察者申请 如果被观察者有数据发送 则先发送1个数据
request(1);
}

@Override
public void onNext(Integer value) {
try {
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
Log.i(TAG, "call: " + value);
//如果以下代码不调用 则打印一次数据后不再打印
//如果调用如下代码，则再次向被观察者申请发送数据
request(1);
}

@Override
public void onCompleted() {

}

@Override
public void onError(Throwable e) {
Log.i(TAG, "onError: "+e);
}

});

3.RxJava1.0对背压的支持

上面已经对背压有一定了了解了，但并不是所有的RxJava1.0语法都支持背压策略，这里列出哪些语法支持背压：

3.1 “热”Observable和”冷”Observable

“冷”Observable：指的是那些在订阅之后才开始发送事件的Observable（每个Subscriber都能接收到完整的事件）。

“热”Observable:指的是那些在创建了Observable之后，（不管是否订阅）就开始发送事件的Observable

“热”Observable这一类是不支持背压的，而是”冷”Observable这一类中也有一部分并不支持背压（比如interval，timer等操作符创建的Observable，也就是跟定时器有关的函数一般都不支持背压）。

3.2 buffer/window

造成背压异常的是被观察者发送的速度太快了，而观察者来不及处理造成了内存超过异常，我们可以在传递的过程中设置一个缓存区，示例如下：

Observable<List<Integer>> observable =
Observable.create(new Observable.OnSubscribe<Integer>() {
@Override
public void call(Subscriber<? super Integer> subscriber) {
for (int i = 0; i < 1000; i++) {
subscriber.onNext(i);
}
}
})
.buffer(10)
.observeOn(Schedulers.newThread());

observable.subscribe(new Subscriber<List<Integer>>() {

@Override
public void onNext(List<Integer> value) {
Log.i(TAG, "call: " + value);
try {
Thread.sleep(50);
} catch (InterruptedException e) {
e.printStackTrace();
}
}

@Override
public void onCompleted() {

}

@Override
public void onError(Throwable e) {
Log.i(TAG, "onError: "+e);
}

});

Buffer函数只能将发送的数据缓存起来，尽量的减少数据发送的速度，但不能根本的解决问题，运行效果如下：

03-16 17:39:17.341 5373-5389/? I/IT520: call: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
03-16 17:39:17.421 5373-5389/? I/IT520: onError: rx.exceptions.MissingBackpressureException

3.3 对于不支持背压的Observable除了使用上述两类生硬的操作符之外，还有更好的选择:onBackpressurebuffer，onBackpressureDrop。

onBackpressurebuffer：把observable发送出来的事件做缓存.

示例修改如下：

Observable<Integer> observable =
Observable.create(new Observable.OnSubscribe<Integer>() {
@Override
public void call(Subscriber<? super Integer> subscriber) {
for (int i = 0; i < 1000; i++) {
subscriber.onNext(i);
}
}
})
//每发送1个数据 当观察者执行完成后再往被观察者拉取1个请求
.onBackpressureBuffer()
.observeOn(Schedulers.newThread());

...
...

上面的代码运行结果如下：

03-16 17:21:07.591 5184-5201/? I/IT520: call: 0
03-16 17:21:08.091 5184-5201/? I/IT520: call: 1
03-16 17:21:08.591 5184-5201/? I/IT520: call: 2
03-16 17:21:10.601 5184-5201/? I/IT520: call: ...
03-16 17:21:08.591 5184-5201/? I/IT520: call: 999

onBackpressureDrop：将observable发送的事件抛弃掉.

Observable.interval(1, TimeUnit.MILLISECONDS)
.onBackpressureDrop()
.observeOn(Schedulers.newThread())
.subscribe(new Subscriber<Long>() {

@Override
public void onCompleted() {

}
@Override
public void onError(Throwable e) {
Log.e(TAG,e.toString());
}

@Override
public void onNext(Long aLong) {
Log.w(TAG,"---->"+aLong);
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});

打印如下：

03-16 17:50:03.081 5626-5642/? W/IT520: ---->0
03-16 17:50:03.191 5626-5642/? W/IT520: ---->1
03-16 17:50:03.301 5626-5642/? W/IT520: ---->2
03-16 17:50:03.391 5626-5642/? W/IT520: ---->3
03-16 17:50:03.491 5626-5642/? W/IT520: ---->4
03-16 17:50:03.601 5626-5642/? W/IT520: ---->5
03-16 17:50:03.701 5626-5642/? W/IT520: ---->6
03-16 17:50:03.801 5626-5642/? W/IT520: ---->7
03-16 17:50:03.901 5626-5642/? W/IT520: ---->8
03-16 17:50:04.001 5626-5642/? W/IT520: ---->9
03-16 17:50:04.101 5626-5642/? W/IT520: ---->10
03-16 17:50:04.201 5626-5642/? W/IT520: ---->11
03-16 17:50:04.301 5626-5642/? W/IT520: ---->12
03-16 17:50:04.411 5626-5642/? W/IT520: ---->13
03-16 17:50:04.511 5626-5642/? W/IT520: ---->14
03-16 17:50:04.611 5626-5642/? W/IT520: ---->15
03-16 17:50:04.711 5626-5642/? W/IT520: ---->1222
03-16 17:50:04.811 5626-5642/? W/IT520: ---->1223
03-16 17:50:04.911 5626-5642/? W/IT520: ---->1224

之所以出现0-15这样连贯的数据，就是是因为observeOn操作符内部有一个长度为16的缓存区，它会首先请求16个事件缓存起来

你可能会觉得这两个操作符和上面讲的过滤和缓存很类似，确实，功能上是有些类似，但是这两个操作符提供了更多的特性，那就是可以响应下游观察者的request(n)方法了，也就是说，使用了这两种操作符，可以让原本不支持背压的Observable“支持”背压了。