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

时间:2021-7-4 作者:qvyue

文章首发公众号:Java架构师联盟,每日更新技术好文

背压

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

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

什么是背压问题

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

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

package com.crazymaker.demo.rxJava.basic;
//省略import
@Slf4j
public class BackpressureDemo {
/**
*演示不使用背压
*/
@Test
public void testNoBackpressure() throws InterruptedException {
//被观察者(主题)
Observable observable = Observable.create(
new Observable.OnSubscribe() {
@Override
public void call(Subscriber super String> subscriber) {
//循环10次
for (int i = 0;i
log.info(“produce ->” + i);
subscriber.onNext(String.valueOf(i));
}
}
});
//观察者
Action1 subscriber = new Action1() {
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() {
@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.OperatorObserveOnObserveOnSubscriber.onNext (OperatorObserveOn.java:160) at rx.internal.operators.OperatorSubscribeOnSubscribeOnSubscriber.onNext
(OperatorSubscribeOn.java:74)
at com.crazymaker.demo.rxJava.basic.BackpressureDemo1.call (BackpressureDemo.java:24) at com.crazymaker.demo.rxJava.basic.BackpressureDemo1.call
(BackpressureDemo.java:19)
at rx.Observable.unsafeSubscribe(Observable.java:10327)
at rx.internal.operators.OperatorSubscribeOnSubscribeOnSubscriber.call (OperatorSubscribeOn.java:100) at rx.internal.schedulers.CachedThreadSchedulerEventLoopWorker$1.call
(CachedThreadScheduler.java:230)
… 9 more

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

背压问题的几种应对模式

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

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

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

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

/**
*演示使用“最近模式”背压
*/
@Test
public void testBackpressure() throws InterruptedException {
//主题实例,使用背压
Observable observable = Observable.create(
new Action1> () {
@Override
public void call(Emitter emitter) {
//无限循环
for (int i = 0; ; i++) {
//log.info(“produce ->” + i);
emitter.onNext(String.valueOf(i));
}
}
}, Emitter.BackpressureMode.LATEST);
//订阅者(观察者)
Action1 subscriber = new Action1() {
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接口中通过一个枚举常量定义了以下几种背压模式:

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架构师联盟,每日更新技术好文

声明:本文内容由互联网用户自发贡献自行上传,本网站不拥有所有权,未作人工编辑处理,也不承担相关法律责任。如果您发现有涉嫌版权的内容,欢迎发送邮件至:qvyue@qq.com 进行举报,并提供相关证据,工作人员会在5个工作日内联系你,一经查实,本站将立刻删除涉嫌侵权内容。