一 初始Reactive Stream
- 概念
Reactive目标是使用非阻塞背压方式提供一个标准的异步流处理”。目前主要的挑战不是找不到解决方案，而是要以最大化的互操作性集成不同的现有方案模式到共同的一个应用中。更确切地说，Reactive流目的是“找到最小的一组接口，方法和协议，用来描述必要的操作和实体以实现这样的目标：以非阻塞背压方式实现数据的异步流”；
- 背压
首先异步消费者会向生产者订阅接收消息，然后当有新的信息可用时，消费者会通过之前订阅时提供的回调函数被再次激活调用。如果生产者发出的信息比消费者能够处理消息最大量还要多，消费者可能会被迫一直在抓消息，耗费越来越多的资源，埋下潜在的崩溃风险。为了防止这一点，需要有一种机制使消费者可以通知生产者，降低消息的生成速度。生产者可以采用多种策略来实现这一要求，这种机制称为背压。
二 Reactive Stream主要接口(Flow API)
public interface Publisher<T> {
public void subscribe(Subscriber<? super T> s);
}
public interface Subscriber<T> {
public void onSubscribe(Subscription s);
public void onNext(T t);
public void onError(Throwable t);
public void onComplete();
}
public interface Subscription {
public void request(long n);
public void cancel();
}
public interface Processor<T,R> extends Subscriber<T>, Publisher<R> {
}
三 完整实例
1⃣️ demo1
package jdk9;
import java.util.concurrent.Flow.Subscriber;
import java.util.concurrent.Flow.Subscription;
import java.util.concurrent.SubmissionPublisher;
import java.util.concurrent.TimeUnit;
public class FlowDemo {
public static void main(String[] args) throws Exception {
// 1. 定义发布者，发布数据类型是Integer
// ֱ直接使用jdk自带的SubmissionPublisher, 它实现了Publisher接口
SubmissionPublisher<Integer> publiser = new SubmissionPublisher<Integer>();
// 2. 定义订阅者
Subscriber<Integer> subscriber = new Subscriber<Integer>() {
private Subscription subscription;
@Override
public void onSubscribe(Subscription subscription) {
// 保存订阅关系，需要用它来给发布者响应
this.subscription = subscription;
// 请求一个数据
this.subscription.request(1);
}
@Override
public void onNext(Integer item) {
// 接受到一个数据，处理
System.out.println("接受到数据: " + item);
try {
TimeUnit.SECONDS.sleep(3);
} catch (InterruptedException e) {
e.printStackTrace();
}
// 处理完调用一个request在请求一个数据
this.subscription.request(1);
// 或者已经达到目标,调用cancel告诉发布者不再接受数据了
// this.subscription.cancel();
}
@Override
public void onError(Throwable throwable) {
// 出现了异常(例如处理数据的时候产生了异常)
throwable.printStackTrace();
// 我们可以告诉发布者，后面不接受数据了
this.subscription.cancel();
}
@Override
public void onComplete() {
// 全部的数据都处理完了(发布者关闭了)
System.out.println("处理完了!");
}
};
// 3. 发布者和订阅者建立订阅关系
publiser.subscribe(subscriber);
// 4. 生产数据并发布，这里忽略数据生产过程
for (int i = 0; i < 1000; i++) {
System.out.println("生产数据:" + i);
publiser.submit(i);
}
// 5. 结束后，关闭发布者
// 正式环境应该使用finally或者使用try-resouce确保关闭
publiser.close();
// 主线程延时停止，否则数据没有消费就退出
Thread.currentThread().join(1000);
System.out.println();
}
}
2⃣️ demo2
package jdk9;
import java.util.concurrent.Flow.Processor;
import java.util.concurrent.Flow.Subscriber;
import java.util.concurrent.Flow.Subscription;
import java.util.concurrent.SubmissionPublisher;
/**
* 带process的flow demo
*/
/**
* Processor需要继承SubmissionPublisher并实现Processor接口
*
* 输入源数据integer, 过滤掉小于0的, 然后转换成字符串发布出去
*/
class MyProcessor extends SubmissionPublisher<String>
implements Processor<Integer, String> {
private Subscription subscription;
@Override
public void onSubscribe(Subscription subscription) {
// 保存订阅关系，需要用它来给发布者响应
this.subscription = subscription;
// 请求一个数据
this.subscription.request(1);
}
@Override
public void onNext(Integer item) {
// 接受到一个数据，处理
System.out.println("处理器接受到数据: " + item);
// 过滤掉小于0的，然后发布出去
if (item > 0) {
this.submit("转换后的数据:" + item);
}
// 处理完调用request在请求一个数据
this.subscription.request(1);
// 或者已经达到目标, 调用cancel告诉发布者不再接受数据了
// this.subscription.cancel();
}
@Override
public void onError(Throwable throwable) {
// 出现了异常(例如处理数据的时候产生了异常)
throwable.printStackTrace();
// 我们可以告诉发布者，后面不再接受数据了
this.subscription.cancel();
}
@Override
public void onComplete() {
// 全部数据处理完了(发布者关闭了)
System.out.println("处理器处理完了!");
// 关闭发布者
this.close();
}
}
public class FlowDemo2 {
public static void main(String[] args) throws Exception {
// 1. 定义发布者，发布数据类型是Integer
// ֱ直接使用jdk自带的SubmissionPublisher
SubmissionPublisher<Integer> publiser = new SubmissionPublisher<Integer>();
// 2. 定义处理器，对数据过滤，并转换成String类型
MyProcessor processor = new MyProcessor();
// 3. 发布者和处理器建立订阅关系
publiser.subscribe(processor);
// 4. 定义最终订阅者，消费String类型数据
Subscriber<String> subscriber = new Subscriber<String>() {
private Subscription subscription;
@Override
public void onSubscribe(Subscription subscription) {
// 保存订阅关系，需要用它来给发布者响应
this.subscription = subscription;
// 请求一个数据
this.subscription.request(1);
}
@Override
public void onNext(String item) {
// 接受到一个数据，处理
System.out.println("接受到数据: " + item);
// 处理完调用一个request在请求一个数据
this.subscription.request(1);
// 或者已经达到目标,调用cancel告诉发布者不再接受数据了
// this.subscription.cancel();
}
@Override
public void onError(Throwable throwable) {
// 出现了异常(例如处理数据的时候产生了异常)
throwable.printStackTrace();
// 我们可以告诉发布者，后面不接受数据了
this.subscription.cancel();
}
@Override
public void onComplete() {
// 全部的数据都处理完了(发布者关闭了)
System.out.println("处理完了!");
}
};
// 5. 处理器和最终的订阅者建立订阅关系
processor.subscribe(subscriber);
// 6. 生产数据并发布
// 这里忽略数据生产过程
publiser.submit(-111);
publiser.submit(111);
// 7. 结束后，关闭发布者
// 正式环境应该使用finally或者使用try-resouce确保关闭
publiser.close();
// 主线程延时停止，否则数据没有消费就退出
Thread.currentThread().join(1000);
}
}
