聊聊reactive streams的backpressure
序
本文主要研究下reactive streams的backpressurereact
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);
}
输出实例ide
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);
}
输出实例this
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);
}
输出实例code
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);
}
输出实例ip
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
相关文章
- 1. 聊聊reactive streams的processors
- 2. 聊聊reactive streams的schedulers
- 3. 聊聊reactive streams的parallel flux
- 4. 聊聊reactive streams的Mono及Flux
- 5. 聊聊reactive streams publisher的doOn方法
- 6. 聊聊reactive streams的tranform操做
- 7. Spring Boot Reactive Streams
- 8. Java 9 Reactive Streams
- 9. java9之Reactive Streams
- 10. What can Reactive Streams offer EE4J?
- 更多相关文章...
- • PHP imagecolorclosest - 取得与指定的颜色最接近的颜色的索引值 - PHP参考手册
- • PHP imagecolorclosesthwb - 取得与指定的颜色最接近的色度的黑白色的索引 - PHP参考手册
- • 漫谈MySQL的锁机制
- • 互联网组织的未来：剖析GitHub员工的任性之源
相关标签/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
最新文章
- 1. 「插件」Runner更新Pro版，帮助设计师远离996
- 2. 错误 707 Could not load file or assembly ‘Newtonsoft.Json, Version=12.0.0.0, Culture=neutral, PublicKe
- 3. Jenkins 2018 报告速览，Kubernetes使用率跃升235%！
- 4. TVI-Android技术篇之注解Annotation
- 5. android studio启动项目
- 6. Android的ADIL
- 7. Android卡顿的检测及优化方法汇总（线下+线上）
- 8. 登录注册的业务逻辑流程梳理
- 9. NDK(1)创建自己的C/C++文件
- 10. 小菜的系统框架界面设计-你的评估是我的决策
欢迎关注本站公众号,获取更多信息
相关文章
- 1. 聊聊reactive streams的processors
- 2. 聊聊reactive streams的schedulers
- 3. 聊聊reactive streams的parallel flux
- 4. 聊聊reactive streams的Mono及Flux
- 5. 聊聊reactive streams publisher的doOn方法
- 6. 聊聊reactive streams的tranform操做
- 7. Spring Boot Reactive Streams
- 8. Java 9 Reactive Streams
- 9. java9之Reactive Streams
- 10. What can Reactive Streams offer EE4J?