Reactive Stream 是什么
之所以在研究webflux之前先研究jdk9的reactive stream，是因为这个是jdk原生自带的，与其他的技术没有什么耦合性，另外，webflux实际上就是jdk8的stream+jdk9的reactive stream的基础上的整合，因此先学习有助于理解。
Reactive Stream（JDK9）
测试项目码云地址：https://gitee.com/josekongng/reactive-stream-test.git
概念
FlowAPI，就是响应式流的所有接口：
| Publisher | 发布者 |
|---|---|
| Subscriber | 订阅者 |
| Subscription | 订阅关系，可以理解为合同，是实现背压的关键 |
| request 告诉发布者，这边需要资源的数量 | |
| cancel | |
| Processor | 没有任何的方法，是对于既是发布者又是订阅者的这种情况来设置的，他继承了发布者和订阅者两个接口 |
实例
订阅和发布
public class ReactiveStreamTest {
public static void main(String[] args) throws InterruptedException {
//定义发布者
//使用jdk自带的SubmissionPublisher
SubmissionPublisher<Integer> submissionPublisher = new SubmissionPublisher<>();
//定义订阅者
Flow.Subscriber<Integer> subscriber = new Flow.Subscriber<Integer>() {
private Flow.Subscription subscription;
@Override
public void onSubscribe(Flow.Subscription subscription) {
//保存订阅关系
this.subscription = subscription;
//请求一个数据
this.subscription.request(1);
}
@Override
public void onNext(Integer item) {
System.out.println("get one:t"+item);
this.subscription.request(1);
//或者已达到了目标，可以调用cancel告诉发布者不再接受数据了
// this.subscription.cancel();
}
@Override
public void onError(Throwable throwable) {
throwable.printStackTrace();
System.out.println("run error:t"+throwable.getMessage());
}
@Override
public void onComplete() {
System.out.println("onComplete*******"+ LocalDateTime.now());
}
};
//发布者订阅者建立关系
submissionPublisher.subscribe(subscriber);
//生产数据，并发布，此处忽略
int data = 111;
submissionPublisher.submit(data);
submissionPublisher.submit(222);
submissionPublisher.submit(333);
submissionPublisher.submit(444);
//关闭发布者
//正式环境应该在finally中或者使用 try-resouce确保关闭
submissionPublisher.close();
//主线程延迟
Thread.currentThread().join(1000);
}
}
输出：
get one: 111
get one: 222
get one: 333
get one: 444
onComplete*******2021-05-12T14:06:39.024737900
处理器
/**
* @author: yunho
* @date: 2021/5/12 11:21
* @description:
* Processor处理器 一般会继承SubmissionPublisher，并实现Processor接口
* 输入源数据 Integer，过滤小于0的，然后转换成字符串发布出去
*/
public class ProcessorTest extends SubmissionPublisher<String> implements Flow.Processor<Integer,String> {
private Flow.Subscription subscription;
@Override
public void onSubscribe(Flow.Subscription subscription) {
this.subscription = subscription;
this.subscription.request(1);
}
@Override
public void onNext(Integer item) {
System.out.println("Processor get one:t"+item);
if(item>0) {
this.submit(item + "_");
}
this.subscription.request(1);
}
@Override
public void onError(Throwable throwable) {
throwable.printStackTrace();
System.out.println("Processor run error:t"+throwable.getMessage());
}
@Override
public void onComplete() {
System.out.println("Processor onComplete*******"+ LocalDateTime.now());
}
public static void main(String[] args) throws InterruptedException {
//定义发布者
//使用jdk自带的SubmissionPublisher
SubmissionPublisher<Integer> submissionPublisher = new SubmissionPublisher<>();
//定义订阅者
Flow.Subscriber<String> subscriber = new Flow.Subscriber<String>() {
private Flow.Subscription subscription;
@Override
public void onSubscribe(Flow.Subscription subscription) {
//保存订阅关系
this.subscription = subscription;
//请求一个数据
this.subscription.request(1);
}
@Override
public void onNext(String item) {
System.out.println("Real ****** get one:t"+item);
this.subscription.request(1);
//或者已达到了目标，可以调用cancel告诉发布者不再接受数据了
// this.subscription.cancel();
}
@Override
public void onError(Throwable throwable) {
throwable.printStackTrace();
System.out.println("Real ****** run error:t"+throwable.getMessage());
}
@Override
public void onComplete() {
System.out.println("Real ****** onComplete*******"+ LocalDateTime.now());
}
};
//处理器
ProcessorTest processorTest = new ProcessorTest();
//发布者订阅者建立关系
submissionPublisher.subscribe(processorTest);
processorTest.subscribe(subscriber);
//生产数据，并发布，此处忽略
int data = 111;
submissionPublisher.submit(data);
submissionPublisher.submit(222);
submissionPublisher.submit(333);
submissionPublisher.submit(444);
//关闭发布者
//正式环境应该在finally中或者使用 try-resouce确保关闭
submissionPublisher.close();
//主线程延迟
Thread.currentThread().join(1000);
}
}
输出：
Processor get one: 111
Processor get one: 222
Processor get one: 333
Processor get one: 444
Real ****** get one: 111_
Real ****** get one: 222_
Real ****** get one: 333_
Real ****** get one: 444_
Processor onComplete*******2021-05-12T14:06:39.024737900
运行机制
发布者的submit方法是一个阻塞方法
通过调试我们发现，在订阅者还没有请求元素时，subscription中已经缓存了对应的元素
因此当订阅者的这个缓存满了，就会导致submit阻塞，测试如下：
//我们将之前的常量数据改为0-999的变量
// int data = 111;
// submissionPublisher.submit(data);
// submissionPublisher.submit(222);
// submissionPublisher.submit(333);
// submissionPublisher.submit(444);
for (int i = 0; i < 1000; i++) {
System.out.println("Publisher produce items："+i);
submissionPublisher.submit(i);
}
//订阅者的消费方法增加阻塞3秒的设置
@Override
public void onNext(Integer item) {
System.out.println("get one:t"+item);
try {TimeUnit.SECONDS.sleep(3);} catch (InterruptedException e) {e.printStackTrace();}
this.subscription.request(1);
//或者已达到了目标，可以调用cancel告诉发布者不再接受数据了
// this.subscription.cancel();
}
输出：
...
Publisher produce items：251
Publisher produce items：252
Publisher produce items：253
Publisher produce items：254
get one: 0
Publisher produce items：255
Publisher produce items：256
Publisher produce items：257
get one: 1
结果发现，缓存了255个元素后，生产者的提交数据开始阻塞。
原文始发于微信公众号（云户）：Reactive Stream 是什么
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 举报，一经查实，本站将立刻删除。
文章由极客之音整理，本文链接：https://www.bmabk.com/index.php/post/25908.html
