深入讲解RxJava响应式编程框架，背压问题的几种应对模式

Posted 2021-06-08 Java架构师联盟

背压

本节首先介绍什么是背压（Backpressure）问题，然后介绍背压问题的几种应对模式。

 什么是背压问题

当上下游的流操作处于不同的线程时，如果上游弹射数据的速度快于下游接收处理数据的速度，对于那些没来得及处理的数据就会造成积压，这些数据既不会丢失，又不会被垃圾回收机制回收，而是存放在一个异步缓存池中，如果缓存池中的数据一直得不到处理，越积越多，最后就会造成内存溢出，这便是响应式编程中的背压问题。

一个存在背压问题的演示实例代码如下：

package com.crazymaker.demo.rxJava.basic;
//省略import
@Slf4j
public class BackpressureDemo {
 /**
 *演示不使用背压
 */
 @Test
 public void testNoBackpressure() throws InterruptedException {
 //被观察者（主题）
 Observable observable = Observable.create(
 new Observable.OnSubscribe<String>() {
 @Override
 public void call(Subscriber<? super String> subscriber) {
 //循环10次
 for (int i = 0;i<10 ; i++) {
 log.info("produce ->" + i);
 subscriber.onNext(String.valueOf(i));
 }
 }
 });
 //观察者
 Action1<String> subscriber = new Action1<String>() {
 public void call(String s){
 try {
 //每消费一次间隔50毫秒
 Thread.sleep(50);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 log.info("consumer ->" + s);
 }
 };
 //订阅:observable与subscriber之间依然通过subscribe()进行关联
 observable
 .subscribeOn(Schedulers.io())
 .observeOn(Schedulers.newThread())
 .subscribe(subscriber);
 Thread.sleep(Integer.MAX_VALUE);
 }
}

在实例代码中，observable发射操作执行在一条通过Schedulers.io()调度器获取的IO线程上，而观察者subscriber的消费操作执行在另一条通过Schedulers.newThread()调度器获取的新线程上。observable流不断发送数据，累积发送10次；观察者subscriber每隔50毫秒接收一条数据。

运行上面的演示程序后，输出的结果如下：

17:56:17.719 [Rxioscheduler-2] INFO c.c.d.r.b.BackpressureDemo - produce ->0
17:56:17.723 [RxIoScheduler-2] INFO c.c.d.r.b.BackpressureDemo - produce ->1
17:56:17.723 [RxIoScheduler-2] INFO c.c.d.r.b.BackpressureDemo - produce ->2
17:56:17.723 [RxIoScheduler-2] INFO c.c.d.r.b.BackpressureDemo - produce ->3
17:56:17.723 [RxIoScheduler-2] INFO c.c.d.r.b.BackpressureDemo - produce ->4
17:56:17.723 [RxIoScheduler-2] INFO c.c.d.r.b.BackpressureDemo - produce ->5
17:56:17.723 [RxIoScheduler-2] INFO c.c.d.r.b.BackpressureDemo - produce ->6
17:56:17.723 [RxIoScheduler-2] INFO c.c.d.r.b.BackpressureDemo - produce ->7
17:56:17.723 [RxIoScheduler-2] INFO c.c.d.r.b.BackpressureDemo - produce ->8
17:56:17.723 [RxIoScheduler-2] INFO c.c.d.r.b.BackpressureDemo - produce ->9
17:56:17.774 [RxNewThreadScheduler-1] INFO c.c.d.r.b.BackpressureDemo - consumer ->0
17:56:17.824 [RxNewThreadScheduler-1] INFO c.c.d.r.b.BackpressureDemo - consumer ->1
17:56:17.875 [RxNewThreadScheduler-1] INFO c.c.d.r.b.BackpressureDemo - consumer ->2
17:56:17.925 [RxNewThreadScheduler-1] INFO c.c.d.r.b.BackpressureDemo - consumer ->3
17:56:17.976 [RxNewThreadScheduler-1] INFO c.c.d.r.b.BackpressureDemo - consumer ->4
17:56:18.027 [RxNewThreadScheduler-1] INFO c.c.d.r.b.BackpressureDemo - consumer ->5
17:56:18.078 [RxNewThreadScheduler-1] INFO c.c.d.r.b.BackpressureDemo - consumer ->6
17:56:18.129 [RxNewThreadScheduler-1] INFO c.c.d.r.b.BackpressureDemo - consumer ->7
17:56:18.179 [RxNewThreadScheduler-1] INFO c.c.d.r.b.BackpressureDemo - consumer ->8
17:56:18.230 [RxNewThreadScheduler-1] INFO c.c.d.r.b.BackpressureDemo - consumer ->9

上面的程序有一个特点：生产者observable弹射数据的速度大于下游消费者subscriber接收处理数据的速度，但是由于数据量小，因此上面的程序运行起来没有出现问题。

简单修改一下生产者，将原来的弹射10条改成无限制地弹射，代码如下：

//被观察者（主题）
 Observable observable = Observable.create(
 new Observable.OnSubscribe<String>() {
 @Override
 public void call(Subscriber<? super String> subscriber) {
 //无限制地循环
 for (int i = 0; ; i++) {
 //log.info("produce ->" + i); subscriber.onNext(String.valueOf(i));
 }
 }
 });

再次运行该演示程序后，抛出的异常如下：

Caused by: rx.exceptions.MissingBackpressureException
at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber.onNext
(OperatorObserveOn.java:160)
at rx.internal.operators.OperatorSubscribeOn$SubscribeOnSubscriber.onNext
(OperatorSubscribeOn.java:74)
at com.crazymaker.demo.rxJava.basic.BackpressureDemo$1.call
(BackpressureDemo.java:24)
at com.crazymaker.demo.rxJava.basic.BackpressureDemo$1.call
(BackpressureDemo.java:19)
at rx.Observable.unsafeSubscribe(Observable.java:10327)
at rx.internal.operators.OperatorSubscribeOn$SubscribeOnSubscriber.call
(OperatorSubscribeOn.java:100)
at rx.internal.schedulers.CachedThreadScheduler$EventLoopWorker$1.call
(CachedThreadScheduler.java:230)
 ... 9 more

异常原因：由于上游observable流弹射数据的速度远远大于下游通过subscriber接收的速度，导致observable用于暂存弹射数据的队列空间耗尽，造成上游数据积压。

 背压问题的几种应对模式

如何应对背压问题呢？在创建主题时可以使用Observable类的一个重载的create方法设置具体的背压模式，该方法的源代码如下：

 public static <T> Observable<T> create(Action1<Emitter<T>> emitter, Emitter.BackpressureMode backpressure) {
 return unsafeCreate(new OnSubscribeCreate<T>(emitter, backpressure));
 }

此方法的第二个参数用于指定一种背压模式。背压模式有多种，比较常用的有“最近模式”
Emitter.BackpressureMode.LATEST。这种模式的含义为：如果消费跟不上，那么仅仅缓存最近弹射出来的数据，将老旧一点的数据直接丢弃。

使用“最近模式”背压，改写4.8.1节的测试用例，代码如下：

 /**
 *演示使用“最近模式”背压
 */
 @Test
 public void testBackpressure() throws InterruptedException {
 //主题实例，使用背压
 Observable observable = Observable.create(
 new Action1<Emitter<String>> () {
 @Override
 public void call(Emitter<String> emitter) {
 //无限循环
 for (int i = 0; ; i++) {
 //log.info("produce ->" + i);
 emitter.onNext(String.valueOf(i));
 }
 }
 }, Emitter.BackpressureMode.LATEST);
 //订阅者（观察者）
 Action1<String> subscriber = new Action1<String>() {
 public void call(String s) {
 try {
 //每消费一次间隔50毫秒
 Thread.sleep(3);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 log.info("consumer ->" + s);
 }
 };
 //订阅: observable与subscriber之间依然通过subscribe()进行关联
 observable
 .subscribeOn(Schedulers.io())
 .observeOn(Schedulers.newThread())
 .subscribe(subscriber);
 Thread.sleep(Integer.MAX_VALUE);
 }

运行这个演示程序，部分输出的结果节选如下：

18:51:54.736 [RxNewThreadScheduler-1] INFO c.c.d.r.b.BackpressureDemo - consumer ->0
18:51:54.745 [RxNewThreadScheduler-1] INFO c.c.d.r.b.BackpressureDemo - consumer ->1
//省略部分输出
18:51:55.217 [RxNewThreadScheduler-1] INFO c.c.d.r.b.BackpressureDemo - consumer ->123
18:51:55.220 [RxNewThreadScheduler-1] INFO c.c.d.r.b.BackpressureDemo - consumer ->124
18:51:55.224 [RxNewThreadScheduler-1] INFO c.c.d.r.b.BackpressureDemo - consumer ->125
18:51:55.228 [RxNewThreadScheduler-1] INFO c.c.d.r.b.BackpressureDemo - consumer ->126
18:51:55.232 [RxNewThreadScheduler-1] INFO c.c.d.r.b.BackpressureDemo - consumer ->127
18:51:55.236 [RxNewThreadScheduler-1] INFO c.c.d.r.b.BackpressureDemo - consumer ->7337652
18:51:55.240 [RxNewThreadScheduler-1] INFO c.c.d.r.b.BackpressureDemo - consumer ->7337653
18:51:55.244 [RxNewThreadScheduler-1] INFO c.c.d.r.b.BackpressureDemo - consumer ->7337654
//省略部分输出
18:51:55.595 [RxNewThreadScheduler-1] INFO c.c.d.r.b.BackpressureDemo - consumer ->7337747
18:51:55.598 [RxNewThreadScheduler-1] INFO c.c.d.r.b.BackpressureDemo - consumer ->14161628

从输出的结果可以看到，上游主题连续不断地弹射，下游订阅者在接收完127后直接跳到了7337652，其间弹射出来的几百万数据（相对旧一点的数据）就直接被丢弃了。

除了
Emitter.BackpressureMode.LATEST“最近模式”外，RxJava在Emitter<T>接口中通过一个枚举常量定义了以下几种背压模式：

enum BackpressureMode {
 /**
 *No backpressure is applied（无背压模式）
*可能导致rx.exceptions.MissingBackpressureException异常
*或者IllegalStateException异常
 */
 NONE,
 /**
 *如果消费者跟不上，就抛出rx.exceptions.MissingBackpressureException异常
 */
 ERROR,
 /**
 *缓存所有的onNext方法弹射出来的消息，等待消费者慢慢地消费
 */
 BUFFER,
 /**
 *如果下游消费跟不上，就丢弃onNext方法弹射出来的新消息
 */
 DROP,
 /**
 *如果消费者跟不上，就丢掉旧的消息，缓存onNext方法弹射出来的新消息
 */
 LATEST
 }

对于以上RxJava背压模式，介绍如下：

（1）BackpressureMode.DROP：在这种模式下，Observable主题使用固定大小为128的缓冲区。如果下游订阅者无法处理，流的第一个元素就会缓存下来，后续的会被丢弃。

（2）BackpressureMode.LATEST：这种模式与BackpressureMode.DROP类似，并且Observable主题也使用固定大小为128的缓冲区。BackpressureMode.LATEST的缓存策略不同，使用最新的弹出元素替换缓冲区缓存的元素。当消费者可以处理下一个元素时，它收到的是Observable最近一次弹出的元素。

（3）BackpressureMode.NONE和BackpressureMode.ERROR：在这两种模式中发送的数据不使用背压。如果上游observable主题弹射数据的速度大于下游通过subscriber接收的速度，造成上游数据积压，就会抛出
MissingBackpressureException异常。

（4）BackpressureMode.BUFFER：在这种模式下，有一个无限的缓冲区（初始化时是128），下游消费不了的元素全部会放到缓冲区中。如果缓冲区中持续地积累，就会导致内存耗尽，抛出OutOfMemoryException异常。

• 

关注我，一个勤奋爱劳动的程序员





关注作者微信公众号 —《Java架构师联盟》
了解更多Java后端架构、python知识以及最新面试宝典