目录
- 一、前置知识
- 1、Lambda表达式
- 2、==函数式接口 Function==
- 3、==StreamAPI==
- 4、Reactive-Stream
- 1）几个实际的问题
- 2）Reactive-Stream是什么？
- 3）==核心接口==
- 4）处理器 Processor
- 5）总结
- 二、Reactor核心
- 1、Reactor
- 1）介绍
- 2）Reactor的三个核心特性
- 3）响应式编程
课程内容
一、前置知识
1、Lambda表达式
interface MyInterface {
int sum(int i, int j);
}
interface MyHaha {
int haha();
default int heihei() {
return 2;
}
; //默认实现
}
@FunctionalInterface //检查注解，帮我们快速检查我们写的接口是否函数式接口
interface MyHehe {
int hehe(int i);
}
/**
* lambda简化函数式接口实例创建
*
* @param args
*/
public static void aaaa(String[] args) {
//1、自己创建实现类对象
MyInterface myInterface = new MyInterfaceImpl();
System.out.println(myInterface.sum(1, 2));
//2、创建匿名实现类
MyInterface myInterface1 = new MyInterface() {
@Override
public int sum(int i, int j) {
return i * i + j * j;
}
};
// System.out.println(myInterface1.sum(2, 3));
//冗余写法
//3、lambda表达式:语法糖 参数列表 + 箭头 + 方法体
MyInterface myInterface2 = (x, y) -> {
return x * x + y * y;
};
System.out.println(myInterface2.sum(2, 3));
}
//参数位置最少情况
MyHaha myHaha = () -> {
return 1;
};
MyHehe myHehe = y -> {
return y * y;
};
MyHehe hehe2 = y -> y - 1;
//总结：
//1)、参数类型可以不写，只写(参数名)，参数变量名随意定义;
// 参数表最少可以只有一个 ()，或者只有一个参数名；
//2、方法体如果只有一句话，{} 可以省略
2、函数式接口 Function
接口中有且只有一个未实现的方法，这个接口就叫做函数式接口
函数式接口的出入参定义：
1、有入参，无出参【消费者】BiConsumer
BiConsumer<String,String> function = (a,b)->{ //能接受两个入参
System.out.println("哈哈："+a+"；呵呵："+b);
};
function.accept("1","2");
2、有入参，有出参【多功能函数】： Function
Function<String,Integer> function = (String x) -> Integer.parseInt(x);
System.out.println(function.apply("2"));
3、无入参，无出参【普通函数】：
Runnable runnable = () -> System.out.println("aaa");
new Thread(runnable).start();
4、无入参 ，有出参【提供者】： supplier
Supplier<String> supplier = ()-> UUID.randomUUID().toString();
String s = supplier.get();
System.out.println(s);
java.util.function包下的所有function定义：
● Consumer： 消费者
● function： 功能函数
● Supplier： 提供者
● Predicate： 断言
get/test/apply/accept调用的函数方法；
3、StreamAPI
中间操作：Intermediate Operations
- filter：过滤； 挑出我们用的元素
- map： 映射： 一一映射，a 变成 b
- mapToInt、mapToLong、mapToDouble
- flatMap：一对多映射
filter、 map、mapToInt、mapToLong、mapToDouble flatMap、flatMapToInt、flatMapToLong、flatMapToDouble mapMulti、mapMultiToInt、mapMultiToLong、mapMultiToDouble、 parallel、unordered、onClose、sequential distinct、sorted、peek、limit、skip、takeWhile、dropWhile、
终止操作：Terminal Operation
forEach、forEachOrdered、toArray、reduce、collect、toList、min、 max、count、anyMatch、allMatch、noneMatch、findFirst、findAny、iterator
4、Reactive-Stream
1）几个实际的问题
当请求量巨大的时候，tomcat会被压垮，此时就需要采取背压的策略，让tomcat根据自己的能力主动去消费请求。
服务器核心数固定时，多线程情况下：线程越多越好吗？
当核心数固定时，线程并不是越多越好，操作系统是分时间片执行任务的，当线程越多时，线程间的切换就越频繁，导致cpu性能消耗越多。
2）Reactive-Stream是什么？
Reactive-Streams 是一个标准规范，定义了异步数据流处理的 API 和行为规则，专注于解决异步流式数据的**背压（Backpressure）**问题。
主要特性
- 异步：通过非阻塞方式处理数据流。
- 流式处理：支持连续数据流的逐步消费，避免一次性加载大量数据。
- 背压机制：允许消费者控制生产者的速率，防止消费者被超量数据淹没。
- 非阻塞：在处理数据时不阻塞线程，提高资源利用率。
背压机制（Backpressure）
Reactive-Streams 的核心之一是通过 Subscription 提供背压支持。
消费者可以通过 request(n) 方法控制生产者的生产速率。
如果消费者处理能力不足，可以减少请求数据量，避免内存溢出或阻塞。
使用场景
事件流处理：如消息队列、用户事件。
高性能网络请求：如 RESTful API、WebSocket。
大数据流处理：需要逐步消费大规模数据的场景。
异步系统集成：将不同系统间的数据流通过异步方式连接起来。
3）核心接口
- Publisher：发布者；产生数据流
- Subscriber：订阅者； 消费数据流
- Subscription：订阅关系；
订阅关系是发布者和订阅者之间的关键接口。订阅者通过订阅来表示对发布者产生的数据的兴趣。订阅者可以请求一定数量的元素，也可以取消订阅。 - Processor：处理器；
处理器是同时实现了发布者和订阅者接口的组件。它可以接收来自一个发布者的数据，进行处理，并将结果发布给下一个订阅者。处理器在Reactor中充当中间环节，代表一个处理阶段，允许你在数据流中进行转换、过滤和其他操作。
【扩展】
以前的编程模型是命令式编程： 过程编程，全自定义
流式编程是响应式|声明式编程，说清楚要干什么，最终结果要怎么样
public class MyFlowDemo {
public static void main(String[] args) {
// 1、定义一个发布者，发布数据
SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
//3、定义一个订阅者； 订阅者感兴趣发布者的数据；
Flow.Subscriber<String> subscriber = new Flow.Subscriber<String>() {
private Flow.Subscription subscription;
@Override //在订阅时 onXxxx：在xxx事件发生时，执行这个回调
public void onSubscribe(Flow.Subscription subscription) {
System.out.println(Thread.currentThread() + "订阅开始了：" + subscription);
this.subscription = subscription;
//从上游请求一个数据
subscription.request(1);
}
@Override //在下一个元素到达时； 执行这个回调； 接受到新数据
public void onNext(String item) {
System.out.println(Thread.currentThread() + "订阅者，接受到数据：" + item);
//从上游请求一个数据
subscription.request(1);
}
@Override //在错误发生时，
public void onError(Throwable throwable) {
System.out.println(Thread.currentThread() + "订阅者，接受到错误信号：" + throwable);
}
@Override //在完成时
public void onComplete() {
System.out.println(Thread.currentThread() + "订阅者，接受到完成信号：");
}
};
publisher.subscribe(subscriber);
for (int i = 0; i < 10; i++) {
publisher.submit("p-" + i);
}
try {
Thread.sleep(20000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
}
4）处理器 Processor
public class FlowDemo {
//定义流中间操作处理器； 只用写订阅者的接口
static class MyProcessor extends SubmissionPublisher<String> implements Flow.Processor<String,String> {
private Flow.Subscription subscription; //保存绑定关系
@Override
public void onSubscribe(Flow.Subscription subscription) {
System.out.println("processor订阅绑定完成");
this.subscription = subscription;
subscription.request(1); //找上游要一个数据
}
@Override //数据到达，触发这个回调
public void onNext(String item) {
System.out.println("processor拿到数据："+item);
//再加工
item += "：哈哈";
submit(item);//把我加工后的数据发出去
subscription.request(1); //再要新数据
}
@Override
public void onError(Throwable throwable) {
}
@Override
public void onComplete() {
}
}
/**
* 1、Publisher：发布者
* 2、Subscriber：订阅者
* 3、Subscription： 订阅关系
* 4、Processor： 处理器
* @param args
*/
//发布订阅模型：观察者模式，
public static void main(String[] args) throws InterruptedException {
//1、定义一个发布者； 发布数据；
SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
//2、定一个中间操作： 给每个元素加个 哈哈 前缀
MyProcessor myProcessor1 = new MyProcessor();
//3、定义一个订阅者； 订阅者感兴趣发布者的数据；
Flow.Subscriber<String> subscriber = new Flow.Subscriber<String>() {
private Flow.Subscription subscription;
@Override //在订阅时 onXxxx：在xxx事件发生时，执行这个回调
public void onSubscribe(Flow.Subscription subscription) {
System.out.println(Thread.currentThread()+"订阅开始了："+subscription);
this.subscription = subscription;
//从上游请求一个数据
subscription.request(1);
}
@Override //在下一个元素到达时； 执行这个回调； 接受到新数据
public void onNext(String item) {
System.out.println(Thread.currentThread()+"订阅者，接受到数据："+item);
if(item.equals("p-7")){
subscription.cancel(); //取消订阅
}else {
subscription.request(1);
}
}
@Override //在错误发生时，
public void onError(Throwable throwable) {
System.out.println(Thread.currentThread()+"订阅者，接受到错误信号："+throwable);
}
@Override //在完成时
public void onComplete() {
System.out.println(Thread.currentThread()+"订阅者，接受到完成信号：");
}
};
//4、绑定发布者和订阅者
publisher.subscribe(myProcessor1); //此时处理器相当于订阅者
myProcessor1.subscribe(subscriber); //此时处理器相当于发布者
//绑定操作；就是发布者，记住了所有订阅者都有谁，有数据后，给所有订阅者把数据推送过去。
// publisher.subscribe(subscriber);
for (int i = 0; i < 10; i++) {
//发布10条数据
if(i == 5){
// publisher.closeExceptionally(new RuntimeException("5555"));
}else {
publisher.submit("p-"+i);
}
//publisher发布的所有数据在它的buffer区；
//中断
// publisher.closeExceptionally();
}
//ReactiveStream
//jvm底层对于整个发布订阅关系做好了 异步+缓存区处理 = 响应式系统；
//发布者通道关闭
publisher.close();
// publisher.subscribe(subscriber2);
//发布者有数据，订阅者就会拿到
Thread.sleep(20000);
}
}
5）总结
二、Reactor核心
响应式编程：
1、底层：基于数据缓冲队列 + 消息驱动模型 + 异步回调机制
2、编码：流式编程 + 链式调用 + 声明式API
3、效果：优雅全异步 + 消息实时处理 + 高吞吐量 + 占用少量资源
回调机制：类似于SpringBoot的事件机制，在SpringBoot应用的启动过程中触发事件。
1、Reactor
1）介绍
Reactor 是一个用于在JVM构建非阻塞应用的响应式编程框架 ！
2）Reactor的三个核心特性
3）响应式编程
