Reactive Streams：一种支持背压的异步数据流处理标准，主流实现有RxJava和Reactor，Spring WebFlux默认集成的是Reactor。
Reactive Streams主要解决背压（back-pressure）问题。当传入的任务速率大于系统处理能力时，数据处理将会对未处理数据产生一个缓冲区。
背压依我的理解来说，是指订阅者能和发布者交互（通过代码里面的调用request和cancel方法交互），可以调节发布者发布数据的速率，解决把订阅者压垮的问题。关键在于上面例子里面的订阅关系Subscription这个接口，他有request和cancel 2个方法，用于通知发布者需要数据和通知发布者不再接受数据。
我们重点理解背压在jdk9里面是如何实现的。关键在于发布者Publisher的实现类SubmissionPublisher的submit方法是block方法。订阅者会有一个缓冲池，默认为Flow.defaultBufferSize() = 256。当订阅者的缓冲池满了之后，发布者调用submit方法发布数据就会被阻塞，发布者就会停（慢）下来；订阅者消费了数据之后（调用Subscription.request方法），缓冲池有位置了，submit方法就会继续执行下去，就是通过这样的机制，实现了调节发布者发布数据的速率，消费得快，生成就快，消费得慢，发布者就会被阻塞，当然就会慢下来了。
Reactive Streams由4个Java接口构成：
- 处理器（Processor）
- 发布商（Publisher）
- 订阅用户（Subscriber）
- 订阅（Subscription）
Flow类允许相互关联的接口和静态方法来建立流控制组件，其中发布者产生由一个或多个订阅者消费的项目，每个订阅者由订阅管理。
Reactive Streams构建在java.util.concurrent.Flow容器对象下，开发者可以在这里找到Flow.Publisher，一个用作lambda表达式或方法引用的赋值目标功能接口。该接口可以让开发者更容易生成Flow.Subscription元素，并且将它们链接在一起。
另一个元素Flow.Subscriber，是异步工作机制，由请求触发。它可以从Flow.Subscription请求多个元素，开发者还可以根据需要自定义缓冲区大小。
背压示例代码1
/**
* reactive stream
* 背压
*/
@Slf4j
public class ReactiveExample1 {
public static void main(String[] args) throws InterruptedException {
//1.发布者
SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
//2. 订阅者
Flow.Subscriber subscriber = new Flow.Subscriber() {
private Flow.Subscription subscription;
@Override
public void onSubscribe(Flow.Subscription subscription) {
log.info("onSubscribe");
//请求数据
subscription.request(1);
this.subscription = subscription;
}
/**
* 处理数据
* @param item
*/
@Override
public void onNext(Object item) {
log.info("item:{}", item);
log.info("onNext");
try {
TimeUnit.SECONDS.sleep(2);
} catch (InterruptedException e) {
e.printStackTrace();
}
this.subscription.request(1);
}
/**
* 数据处理异常
* @param throwable
*/
@Override
public void onError(Throwable throwable) {
log.info("onError");
}
/**
* 数据完成
*/
@Override
public void onComplete() {
log.info("onComplete");
}
};
// 3. 建立关系
publisher.subscribe(subscriber);
// 4. 生产数据
for (int i = 0; i < 500; i++) {
publisher.submit("test" + i);
log.info("submit:{}","test" + i);
}
// 5 .结束关闭
publisher.close();
TimeUnit.SECONDS.sleep(10);
}
}
11:28:53.864 [ForkJoinPool.commonPool-worker-3] INFO cn.wyj.learn.reactive.ReactiveExample1 - onSubscribe
11:28:53.878 [ForkJoinPool.commonPool-worker-3] INFO cn.wyj.learn.reactive.ReactiveExample1 - item:test0
11:28:53.879 [main] INFO cn.wyj.learn.reactive.ReactiveExample1 - submit:test0
11:28:53.881 [ForkJoinPool.commonPool-worker-3] INFO cn.wyj.learn.reactive.ReactiveExample1 - onNext
11:28:53.881 [main] INFO cn.wyj.learn.reactive.ReactiveExample1 - submit:test1
11:28:53.881 [main] INFO cn.wyj.learn.reactive.ReactiveExample1 - submit:test2
.
.
.
.
.
.
11:28:53.896 [main] INFO cn.wyj.learn.reactive.ReactiveExample1 - submit:test252
11:28:53.896 [main] INFO cn.wyj.learn.reactive.ReactiveExample1 - submit:test253
11:28:53.896 [main] INFO cn.wyj.learn.reactive.ReactiveExample1 - submit:test254
11:28:53.897 [main] INFO cn.wyj.learn.reactive.ReactiveExample1 - submit:test255
11:28:53.897 [main] INFO cn.wyj.learn.reactive.ReactiveExample1 - submit:test256
11:28:55.882 [main] INFO cn.wyj.learn.reactive.ReactiveExample1 - submit:test257
11:28:55.882 [ForkJoinPool.commonPool-worker-3] INFO cn.wyj.learn.reactive.ReactiveExample1 - item:test1
11:28:55.883 [ForkJoinPool.commonPool-worker-3] INFO cn.wyj.learn.reactive.ReactiveExample1 - onNext
11:28:57.884 [ForkJoinPool.commonPool-worker-3] INFO cn.wyj.learn.reactive.ReactiveExample1 - item:test2
11:28:57.884 [main] INFO cn.wyj.learn.reactive.ReactiveExample1 - submit:test258
11:28:57.884 [ForkJoinPool.commonPool-worker-3] INFO cn.wyj.learn.reactive.ReactiveExample1 - onNext
11:28:59.885 [ForkJoinPool.commonPool-worker-3] INFO cn.wyj.learn.reactive.ReactiveExample1 - item:test3
11:28:59.885 [main] INFO cn.wyj.learn.reactive.ReactiveExample1 - submit:test259
11:28:59.885 [ForkJoinPool.commonPool-worker-3] INFO cn.wyj.learn.reactive.ReactiveExample1 - onNext
11:29:01.886 [ForkJoinPool.commonPool-worker-3] INFO cn.wyj.learn.reactive.ReactiveExample1 - item:test4
11:29:01.886 [ForkJoinPool.commonPool-worker-3] INFO cn.wyj.learn.reactive.ReactiveExample1 - onNext
11:29:01.886 [main] INFO cn.wyj.learn.reactive.ReactiveExample1 - submit:test260
11:29:03.886 [ForkJoinPool.commonPool-worker-3] INFO cn.wyj.learn.reactive.ReactiveExample1 - item:test5
11:29:03.886 [main] INFO cn.wyj.learn.reactive.ReactiveExample1 - submit:test261
11:29:03.886 [ForkJoinPool.commonPool-worker-3] INFO cn.wyj.learn.reactive.ReactiveExample1 - onNext
11:29:05.887 [ForkJoinPool.commonPool-worker-3] INFO cn.wyj.learn.reactive.ReactiveExample1 - item:test6
11:29:05.887 [main] INFO cn.wyj.learn.reactive.ReactiveExample1 - submit:test262
11:29:05.887 [ForkJoinPool.commonPool-worker-3] INFO cn.wyj.learn.reactive.ReactiveExample1 - onNext
11:29:07.888 [ForkJoinPool.commonPool-worker-3] INFO cn.wyj.learn.reactive.ReactiveExample1 - item:test7
11:29:07.888 [main] INFO cn.wyj.learn.reactive.ReactiveExample1 - submit:test263
11:29:07.888 [ForkJoinPool.commonPool-worker-3] INFO cn.wyj.learn.reactive.ReactiveExample1 - onNext
根据运行结果我们可以看到任务提交256 后就不能直接提交了, 消费者消费完一条消息后又可以提交一条数据,这些就起到了流控的作用
传统的发布订阅模式, 生产者并不能够根据消费者调节生成速率
@Slf4j
public class NormalPublisherSubscriber {
public static void main(String[] args) {
BlockingQueue<String> queue = new LinkedBlockingDeque();
ExecutorService executorService = Executors.newCachedThreadPool();
//订阅者,消费者
executorService.submit(() -> {
try {
while (true) {
String take = queue.take();
log.info("Received :{}", take);
}
} catch (InterruptedException e) {
e.printStackTrace();
}
});
//发布者,生产者
executorService.submit(() -> {
try {
queue.put("test1");
queue.put("test2");
queue.put("test3");
queue.put("test4");
} catch (InterruptedException e) {
e.printStackTrace();
}
});
executorService.shutdown();
}
}
参考1 https://www.imooc.com/article/27181
参考2 https://blog.csdn.net/houzhizhen/article/details/78195210
