资讯专栏INFORMATION COLUMN

Rxjava2教程:Flowable背压支持

qianfeng / 2164人阅读

摘要:相对于最大的更新就是把对背压问题的处理逻辑从中抽取出来产生了新的可观察对象。由于基于发射的数据流,以及对数据加工处理的各操作符都添加了背压支持,附加了额外的逻辑,其运行效率要比慢得多。

背压(backpressure)
当上下游在不同的线程中,通过Observable发射,处理,响应数据流时,如果上游发射数据的速度快于下游接收处理数据的速度,这样对于那些没来得及处理的数据就会造成积压,这些数据既不会丢失,也不会被垃圾回收机制回收,而是存放在一个异步缓存池中,如果缓存池中的数据一直得不到处理,越积越多,最后就会造成内存溢出,这便是响应式编程中的背压(backpressure)问题。
例如,运行以下代码:

public void demo1() {

    Observable
            .create(new ObservableOnSubscribe() {
                @Override
                public void subscribe(ObservableEmitter e) throws Exception {
                    int i = 0;
                    while (true) {
                        i++;
                        e.onNext(i);
                    }
                }
            })
            .subscribeOn(Schedulers.newThread())
            .observeOn(Schedulers.newThread())
            .subscribe(new Consumer() {
                @Override
                public void accept(Integer integer) throws Exception {
                    Thread.sleep(5000);
                    System.out.println(integer);
                }
            });
}

创建一个可观察对象Observable在Schedulers.newThread()的线程中不断发送数据,而观察者Observer在Schedulers.newThread()的另一个线程中每隔5秒接收打印一条数据。
运行后,查看内存使用如下:

由于上游通过Observable发射数据的速度大于下游通过Consumer接收处理数据的速度,而且上下游分别运行在不同的线程中,下游对数据的接收处理不会堵塞上游对数据的发射,造成上游数据积压,内存不断增加,最后便会导致内存溢出。

Flowable
既然在函数响应式编程中会产生背压(backpressure)问题,那么在函数响应式编程中就应该有解决方案。
Rxjava2相对于Rxjava1最大的更新就是把对背压问题的处理逻辑从Observable中抽取出来产生了新的可观察对象Flowable。

在Rxjava2中,Flowable可以看做是为了解决背压问题,在Observable的基础上优化后的产物,与Observable不处在同一组观察者模式下,Observable是ObservableSource/Observer这一组观察者模式中ObservableSource的典型实现,而Flowable是Publisher与Subscriber这一组观察者模式中Publisher的典型实现。

所以在使用Flowable的时候,可观察对象不再是Observable,而是Flowable;观察者不再是Observer,而是Subscriber。Flowable与Subscriber之间依然通过subscribe()进行关联。

虽然在Rxjava2中,Flowable是在Observable的基础上优化后的产物,Observable能解决的问题Flowable也都能解决,但是并不代表Flowable可以完全取代Observable,在使用的过程中,并不能抛弃Observable而只用Flowable。

由于基于Flowable发射的数据流,以及对数据加工处理的各操作符都添加了背压支持,附加了额外的逻辑,其运行效率要比Observable慢得多。

只有在需要处理背压问题时,才需要使用Flowable。

由于只有在上下游运行在不同的线程中,且上游发射数据的速度大于下游接收处理数据的速度时,才会产生背压问题;
所以,如果能够确定:
1、上下游运行在同一个线程中,
2、上下游工作在不同的线程中,但是下游处理数据的速度不慢于上游发射数据的速度,
3、上下游工作在不同的线程中,但是数据流中只有一条数据
则不会产生背压问题,就没有必要使用Flowable,以免影响性能。

文章版权归作者所有,未经允许请勿转载,若此文章存在违规行为,您可以联系管理员删除。

转载请注明本文地址:https://www.ucloud.cn/yun/72149.html

相关文章

  • 「码个蛋」2017年200篇精选干货集合

    摘要:让你收获满满码个蛋从年月日推送第篇文章一年过去了已累积推文近篇文章,本文为年度精选,共计篇,按照类别整理便于读者主题阅读。本篇文章是今年的最后一篇技术文章,为了让大家在家也能好好学习,特此花了几个小时整理了这些文章。 showImg(https://segmentfault.com/img/remote/1460000013241596); 让你收获满满! 码个蛋从2017年02月20...

    wangtdgoodluck 评论0 收藏0
  • Rxjava2教程:Single、Completable、Maybe——简化版的Observable

    摘要:只发射一条单一的数据,或者一条异常通知,不能发射完成通知,其中数据与通知只能发射一个。观察者可观察对象对应的观察者为方法等同于中的。没有用来处理完成通知的方法示例三发射完成通知示例四发射异常通知 在Rxjava2中,Observale和Flowable都是用来发射数据流的,但是,我们在实际应用中,很多时候,需要发射的数据并不是数据流的形式,而只是一条单一的数据,或者一条完成通知,或者一...

    Vicky 评论0 收藏0

发表评论

0条评论

qianfeng

|高级讲师

TA的文章

阅读更多
最新活动
阅读需要支付1元查看
<