当前位置： 首页 > news >正文

聊聊reactive streams的backpressure

news 来源：原创 2024/5/6 19:42:31

序

本文主要研究下reactive streams的backpressure

reactive streams跟传统streams的区别

    @Test
    public void testShowReactiveStreams() throws InterruptedException {
        Flux.interval(Duration.ofMillis(1000))
                .take(500)
                .subscribe(e -> LOGGER.info("get {}",e));

        Thread.sleep(5*60*1000);
    }
复制代码

输出实例如下：

18:52:34.118 [main] DEBUG reactor.util.Loggers$LoggerFactory - Using Slf4j logging framework
18:52:35.157 [parallel-2] INFO com.example.demo.FluxTest - get 0
18:52:36.156 [parallel-2] INFO com.example.demo.FluxTest - get 1
18:52:37.156 [parallel-2] INFO com.example.demo.FluxTest - get 2
18:52:38.159 [parallel-2] INFO com.example.demo.FluxTest - get 3
18:52:39.157 [parallel-2] INFO com.example.demo.FluxTest - get 4
18:52:40.155 [parallel-2] INFO com.example.demo.FluxTest - get 5
18:52:41.154 [parallel-2] INFO com.example.demo.FluxTest - get 6
18:52:42.158 [parallel-2] INFO com.example.demo.FluxTest - get 7
18:52:43.157 [parallel-2] INFO com.example.demo.FluxTest - get 8
18:52:44.156 [parallel-2] INFO com.example.demo.FluxTest - get 9
18:52:45.154 [parallel-2] INFO com.example.demo.FluxTest - get 10
复制代码

传统的list streams不是异步的，好比如一批500件的半成品，得在A环节都处理完，才能下一个环节B，而reactive streams之所以成为reactive，就好比如这批500件的半成品，A环节每处理完一件就可以立即推往下个环节B处理，源源不断，而不是等所有的半成品都在A环节处理再推往B环节。典型的活生生的一个生产流水线的例子。

backpressure

这样一个生产流水线，有个要求就是每个环节的处理要能够协调，就像电影起跑线里头男主角去工厂打工，流水线花花往他那边推送货物，他速度跟不上，导致货物都掉地上了，最后不得不人工关掉流水线。 在应用程序里头，如果发布者速度过快，而订阅者速度慢，那么就会数据就会堆积，控制不好就容易产生内存溢出，而backpressure就专门用来解决这个问题的。

pull模型的backpressure

@Test
    public void testPullBackpressure(){
        Flux.just(1, 2, 3, 4)
                .log()
                .subscribe(new Subscriber<Integer>() {
                    private Subscription s;
                    int onNextAmount;

                    @Override
                    public void onSubscribe(Subscription s) {
                        this.s = s;
                        s.request(2);
                    }

                    @Override
                    public void onNext(Integer integer) {
                        System.out.println(integer);
                        onNextAmount++;
                        if (onNextAmount % 2 == 0) {
                            s.request(2);
                        }
                    }

                    @Override
                    public void onError(Throwable t) {}

                    @Override
                    public void onComplete() {}
                });

        try {
            Thread.sleep(10*1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
复制代码

push模型的backpressure

借助线程相关的操作符，比如timeout(),delayElements(),buffer(),skip(),take()来控制数据产生速度。

delayElements

@Test
    public void testPushBackpressure() throws InterruptedException {
        Flux.range(1, 1000)
                .delayElements(Duration.ofMillis(200))
                .subscribe(e -> {
                    LOGGER.info("subscribe:{}",e);
                    try {
                        Thread.sleep(2000);
                    } catch (InterruptedException e1) {
                        e1.printStackTrace();
                    }
                });
        Thread.sleep(100*1000);
    }
复制代码

输出实例

19:37:00.870 [main] DEBUG reactor.util.Loggers$LoggerFactory - Using Slf4j logging framework
19:37:01.117 [parallel-1] INFO com.example.demo.FluxTest - subscribe:1
19:37:03.326 [parallel-2] INFO com.example.demo.FluxTest - subscribe:2
19:37:05.535 [parallel-3] INFO com.example.demo.FluxTest - subscribe:3
19:37:07.743 [parallel-4] INFO com.example.demo.FluxTest - subscribe:4
19:37:09.953 [parallel-5] INFO com.example.demo.FluxTest - subscribe:5
19:37:12.156 [parallel-6] INFO com.example.demo.FluxTest - subscribe:6
19:37:14.363 [parallel-7] INFO com.example.demo.FluxTest - subscribe:7
19:37:16.568 [parallel-8] INFO com.example.demo.FluxTest - subscribe:8
19:37:18.775 [parallel-1] INFO com.example.demo.FluxTest - subscribe:9
复制代码

这是个delayElements的例子，可以看到数据不丢失，但是延时是生产延时+消费延时

sample

@Test
    public void testSampleBackpressure() throws InterruptedException {
        Flux.range(1, 1000)
                .log()
                .delayElements(Duration.ofMillis(200))
                .sample(Duration.ofMillis(1000))
                .subscribe(e -> {
                    LOGGER.info("subscribe:{}",e);
                    try {
                        Thread.sleep(2000);
                    } catch (InterruptedException e1) {
                        e1.printStackTrace();
                    }
                });
        Thread.sleep(100*1000);
    }
复制代码

输出实例

19:48:40.516 [main] DEBUG reactor.util.Loggers$LoggerFactory - Using Slf4j logging framework
19:48:40.544 [main] INFO reactor.Flux.Range.1 - | onSubscribe([Synchronous Fuseable] FluxRange.RangeSubscription)
19:48:40.546 [main] INFO reactor.Flux.Range.1 - | onNext(1)
19:48:40.770 [parallel-2] INFO reactor.Flux.Range.1 - | onNext(2)
19:48:40.974 [parallel-3] INFO reactor.Flux.Range.1 - | onNext(3)
19:48:41.175 [parallel-4] INFO reactor.Flux.Range.1 - | onNext(4)
19:48:41.378 [parallel-5] INFO reactor.Flux.Range.1 - | onNext(5)
19:48:41.543 [parallel-1] INFO com.example.demo.FluxTest - subscribe:4
19:48:41.583 [parallel-6] INFO reactor.Flux.Range.1 - | onNext(6)
19:48:41.785 [parallel-7] INFO reactor.Flux.Range.1 - | onNext(7)
19:48:41.989 [parallel-8] INFO reactor.Flux.Range.1 - | onNext(8)
19:48:43.547 [parallel-1] INFO reactor.Flux.Range.1 - | onNext(9)
19:48:43.548 [parallel-1] INFO com.example.demo.FluxTest - subscribe:8
19:48:43.751 [parallel-2] INFO reactor.Flux.Range.1 - | onNext(10)
19:48:43.952 [parallel-3] INFO reactor.Flux.Range.1 - | onNext(11)
复制代码

可以看到，由于订阅者速度慢，导致部分数据被丢弃

buffer

@Test
    public void testBufferBackpressure() throws InterruptedException {
        Flux.range(1, 1000)
//                .log()
                .delayElements(Duration.ofMillis(200))
                .buffer(Duration.ofMillis(800))
                .subscribe(e -> {
                    LOGGER.info("subscribe:{}",e);
                    try {
                        Thread.sleep(2000);
                    } catch (InterruptedException e1) {
                        e1.printStackTrace();
                    }
                });
        Thread.sleep(100*1000);
    }
复制代码

输出实例

19:55:06.680 [main] DEBUG reactor.util.Loggers$LoggerFactory - Using Slf4j logging framework
19:55:06.712 [main] INFO reactor.Flux.Range.1 - | onSubscribe([Synchronous Fuseable] FluxRange.RangeSubscription)
19:55:06.714 [main] INFO reactor.Flux.Range.1 - | onNext(1)
19:55:06.940 [parallel-2] INFO reactor.Flux.Range.1 - | onNext(2)
19:55:07.141 [parallel-3] INFO reactor.Flux.Range.1 - | onNext(3)
19:55:07.343 [parallel-4] INFO reactor.Flux.Range.1 - | onNext(4)
19:55:07.509 [parallel-1] INFO com.example.demo.FluxTest - subscribe:[1, 2, 3]
19:55:07.545 [parallel-5] INFO reactor.Flux.Range.1 - | onNext(5)
19:55:07.748 [parallel-6] INFO reactor.Flux.Range.1 - | onNext(6)
19:55:07.951 [parallel-7] INFO reactor.Flux.Range.1 - | onNext(7)
19:55:08.156 [parallel-8] INFO reactor.Flux.Range.1 - | onNext(8)
19:55:09.512 [parallel-1] INFO com.example.demo.FluxTest - subscribe:[4, 5, 6, 7]
19:55:11.515 [parallel-1] INFO reactor.Flux.Range.1 - | onNext(9)
19:55:11.516 [parallel-1] INFO com.example.demo.FluxTest - subscribe:[8]
19:55:11.719 [parallel-2] INFO reactor.Flux.Range.1 - | onNext(10)
19:55:11.923 [parallel-3] INFO reactor.Flux.Range.1 - | onNext(11)
19:55:12.127 [parallel-4] INFO reactor.Flux.Range.1 - | onNext(12)
19:55:12.330 [parallel-5] INFO reactor.Flux.Range.1 - | onNext(13)
19:55:12.533 [parallel-6] INFO reactor.Flux.Range.1 - | onNext(14)
19:55:12.735 [parallel-7] INFO reactor.Flux.Range.1 - | onNext(15)
19:55:12.941 [parallel-8] INFO reactor.Flux.Range.1 - | onNext(16)
19:55:13.516 [parallel-1] INFO com.example.demo.FluxTest - subscribe:[9, 10, 11, 12, 13, 14, 15]
19:55:15.517 [parallel-1] INFO reactor.Flux.Range.1 - | onNext(17)
19:55:15.517 [parallel-1] INFO com.example.demo.FluxTest - subscribe:[16]
19:55:15.721 [parallel-2] INFO reactor.Flux.Range.1 - | onNext(18)
19:55:15.925 [parallel-3] INFO reactor.Flux.Range.1 - | onNext(19)
19:55:16.127 [parallel-4] INFO reactor.Flux.Range.1 - | onNext(20)
19:55:16.331 [parallel-5] INFO reactor.Flux.Range.1 - | onNext(21)
19:55:16.537 [parallel-6] INFO reactor.Flux.Range.1 - | onNext(22)
19:55:16.738 [parallel-7] INFO reactor.Flux.Range.1 - | onNext(23)
19:55:16.942 [parallel-8] INFO reactor.Flux.Range.1 - | onNext(24)
19:55:17.519 [parallel-1] INFO com.example.demo.FluxTest - subscribe:[17, 18, 19, 20, 21, 22, 23]
19:55:19.522 [parallel-1] INFO reactor.Flux.Range.1 - | onNext(25)
19:55:19.522 [parallel-1] INFO com.example.demo.FluxTest - subscribe:[24]
复制代码

将每个800ms内产生的数据堆积为一批次推送给订阅者

skip

@Test
    public void testSkip() throws InterruptedException {
        Flux.range(1, 1000)
                .log()
                .delayElements(Duration.ofMillis(200))
                .skip(Duration.ofMillis(800))
                .subscribe(e -> {
                    LOGGER.info("subscribe:{}",e);
                    try {
                        Thread.sleep(2000);
                    } catch (InterruptedException e1) {
                        e1.printStackTrace();
                    }
                });
        Thread.sleep(100*1000);
    }
复制代码

输出实例

20:02:07.558 [main] DEBUG reactor.util.Loggers$LoggerFactory - Using Slf4j logging framework
20:02:07.606 [main] INFO reactor.Flux.Range.1 - | onSubscribe([Synchronous Fuseable] FluxRange.RangeSubscription)
20:02:07.608 [main] INFO reactor.Flux.Range.1 - | onNext(1)
20:02:07.815 [parallel-2] INFO reactor.Flux.Range.1 - | onNext(2)
20:02:08.016 [parallel-3] INFO reactor.Flux.Range.1 - | onNext(3)
20:02:08.218 [parallel-4] INFO reactor.Flux.Range.1 - | onNext(4)
20:02:08.421 [parallel-5] INFO com.example.demo.FluxTest - subscribe:4
20:02:10.425 [parallel-5] INFO reactor.Flux.Range.1 - | onNext(5)
20:02:10.631 [parallel-6] INFO com.example.demo.FluxTest - subscribe:5
20:02:12.635 [parallel-6] INFO reactor.Flux.Range.1 - | onNext(6)
20:02:12.840 [parallel-7] INFO com.example.demo.FluxTest - subscribe:6
20:02:14.843 [parallel-7] INFO reactor.Flux.Range.1 - | onNext(7)
20:02:15.049 [parallel-8] INFO com.example.demo.FluxTest - subscribe:7
复制代码

通过skip指定跳过最初一个时间段内产生的数据

take

@Test
    public void testTakeBackpressure() throws InterruptedException {
        Flux.range(1, 1000)
                .log()
                .delayElements(Duration.ofMillis(200))
                .take(Duration.ofMillis(4000))
                .subscribe(e -> {
                    LOGGER.info("subscribe:{}",e);
                    try {
                        Thread.sleep(2000);
                    } catch (InterruptedException e1) {
                        e1.printStackTrace();
                    }
                });
        Thread.sleep(100*1000);
    }
复制代码

输出实例

20:05:08.366 [main] DEBUG reactor.util.Loggers$LoggerFactory - Using Slf4j logging framework
20:05:08.419 [main] INFO reactor.Flux.Range.1 - | onSubscribe([Synchronous Fuseable] FluxRange.RangeSubscription)
20:05:08.422 [main] INFO reactor.Flux.Range.1 - | onNext(1)
20:05:08.629 [parallel-2] INFO com.example.demo.FluxTest - subscribe:1
20:05:10.633 [parallel-2] INFO reactor.Flux.Range.1 - | onNext(2)
20:05:10.835 [parallel-3] INFO com.example.demo.FluxTest - subscribe:2
20:05:12.418 [parallel-1] INFO reactor.Flux.Range.1 - | cancel()
复制代码

通过take表示只推送前面几个或前面一段时间产生的数据给订阅者

小结

reactive streams对于具有多个阶段的数据处理来说，非常有用，可以节省很多时间，另外又有backpressure来控制订阅者速度过慢的问题，非常值得使用。

doc

  • 关于RxJava最友好的文章——背压（Backpressure）
  • Java ProjectReactor框架之Flux篇

相关文章：

  • python 运行 hadoop 2.0 mapreduce 程序
  • ListView优化
  • 蓝桥杯第二届试题集锦
  • OpenAI发文怒怼：对抗样本怎么不会对检测产生干扰了？
  • mina编解码(摘录)
  • 团队作业7——第二次项目冲刺-Beta版本项目计划
  • mysql登陆密码忘记，解决办法
  • VMPlayer Ubuntu 16.04 Copy and Paste with Host 主机与宿机之间的复制粘贴
  • MyISAM 与 InnoDB 的区别
  • 牛逼的 弹出层 layer ！！！
  • mysql中OPTIMIZE TABLE的作用
  • Speed up your eclipse as a super fast IDE
  • grep 正则表达式 vim及相关知识
  • 自动部署ubuntu系统时ks.cfg和ks.seed有什么不同
  • Android学习系列(19)--App离线下载
  • 【跃迁之路】【585天】程序员高效学习方法论探索系列（实验阶段342-2018.09.13）...
  • Android开发 - 掌握ConstraintLayout（四）创建基本约束
  • Android开源项目规范总结
  • C++回声服务器_9-epoll边缘触发模式版本服务器
  • ES6系统学习----从Apollo Client看解构赋值
  • iOS小技巧之UIImagePickerController实现头像选择
  • JavaScript新鲜事·第5期
  • Java-详解HashMap
  • Mysql5.6主从复制
  • NSTimer学习笔记
  • Python十分钟制作属于你自己的个性logo
  • React-flux杂记
  • 计算机在识别图像时“看到”了什么？
  • 将回调地狱按在地上摩擦的Promise
  • 软件开发学习的5大技巧，你知道吗？
  • 双管齐下，VMware的容器新战略
  • 详解NodeJs流之一
  • 携程小程序初体验
  • ​业务双活的数据切换思路设计（下）
  • （1）(1.9) MSP (version 4.2)
  • （2.2w字）前端单元测试之Jest详解篇
  • (html转换)StringEscapeUtils类的转义与反转义方法
  • （分享）一个图片添加水印的小demo的页面，可自定义样式
  • （附源码）ssm基于jsp高校选课系统 毕业设计 291627
  • （附源码）计算机毕业设计SSM基于java的云顶博客系统
  • (九)One-Wire总线-DS18B20
  • （算法）Game
  • (转)Linux整合apache和tomcat构建Web服务器
  • .equal()和==的区别 怎样判断字符串为空问题： Illegal invoke-super to void nio.file.AccessDeniedException
  • .gitignore文件设置了忽略但不生效
  • .NET Core引入性能分析引导优化
  • .NET大文件上传知识整理
  • .NET开发人员必知的八个网站
  • .net开发引用程序集提示没有强名称的解决办法
  • .Net转Java自学之路—基础巩固篇十三（集合）
  • :=
  • @31省区市高考时间表来了，祝考试成功
  • @JsonSerialize注解的使用
  • @SuppressWarnings注解
  • [ C++ ] STL priority_queue(优先级队列)使用及其底层模拟实现，容器适配器，deque(双端队列)原理了解