您的位置:首页 > 新闻 > 热点要闻 > 网站制作模板免费下载_百度seo优化技术_工具seo_seo技术代理

网站制作模板免费下载_百度seo优化技术_工具seo_seo技术代理

2024/12/23 8:45:11 来源:https://blog.csdn.net/hc1285653662/article/details/144150603  浏览:    关键词:网站制作模板免费下载_百度seo优化技术_工具seo_seo技术代理
网站制作模板免费下载_百度seo优化技术_工具seo_seo技术代理

目录

  • 一、前置知识
    • 1、Lambda表达式
    • 2、==函数式接口 Function==
    • 3、==StreamAPI==
    • 4、Reactive-Stream
      • 1)几个实际的问题
      • 2)Reactive-Stream是什么?
      • 3)==核心接口==
      • 4)处理器 Processor
      • 5)总结
  • 二、Reactor核心
    • 1、Reactor
      • 1)介绍
      • 2)响应式编程
      • 3)Reactor核心特性
        • 1、Mono和Flux
        • 2、subscribe()
        • 3、流的取消
        • 4、BaseSubscriber
        • 5、背压(Backpressure )和请求重塑(Reshape Requests)
        • 6、以编程的方式创建队列
        • 7、handle()的使用
        • 8、自定义线程调度

课程内容
在这里插入图片描述

一、前置知识

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() {@Overridepublic 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; //保存绑定关系@Overridepublic 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); //再要新数据}@Overridepublic void onError(Throwable throwable) {}@Overridepublic 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)响应式编程

在这里插入图片描述

3)Reactor核心特性

1、Mono和Flux

Mono: 0|1 数据流
Flux: N数据流

响应式流:元素(内容) + 信号(完成/异常);

2、subscribe()

传递钩子函数

flux.subscribe(v-> System.out.println("v = " + v), //流元素消费throwable -> System.out.println("throwable = " + throwable), //感知异常结束()-> System.out.println("流结束了...") //感知正常结束
);

自定义消费者

flux.subscribe(new BaseSubscriber<String>() {// 生命周期钩子1: 订阅关系绑定的时候触发@Overrideprotected void hookOnSubscribe(Subscription subscription) {// 流被订阅的时候触发System.out.println("绑定了..."+subscription);//找发布者要数据request(1); //要1个数据
//                requestUnbounded(); //要无限数据}@Overrideprotected void hookOnNext(String value) {System.out.println("数据到达,正在处理:"+value);request(1); //要1个数据}//  hookOnComplete、hookOnError 二选一执行@Overrideprotected void hookOnComplete() {System.out.println("流正常结束...");}@Overrideprotected void hookOnError(Throwable throwable) {System.out.println("流异常..."+throwable);}@Overrideprotected void hookOnCancel() {System.out.println("流被取消...");}@Overrideprotected void hookFinally(SignalType type) {System.out.println("最终回调...一定会被执行");}});
3、流的取消

消费者调用 cancle() 取消流的订阅;

 flux.subscribe(new BaseSubscriber<String>() {// 生命周期钩子1: 订阅关系绑定的时候触发@Overrideprotected void hookOnSubscribe(Subscription subscription) {// 流被订阅的时候触发System.out.println("绑定了..."+subscription);//找发布者要数据request(1); //要1个数据
//                requestUnbounded(); //要无限数据}@Overrideprotected void hookOnNext(String value) {System.out.println("数据到达,正在处理:"+value);if(value.equals("哈哈:5")){cancel(); //取消流}request(1); //要1个数据}//  hookOnComplete、hookOnError 二选一执行@Overrideprotected void hookOnComplete() {System.out.println("流正常结束...");}@Overrideprotected void hookOnError(Throwable throwable) {System.out.println("流异常..."+throwable);}@Overrideprotected void hookOnCancel() {System.out.println("流被取消...");}@Overrideprotected void hookFinally(SignalType type) {System.out.println("最终回调...一定会被执行");}});
4、BaseSubscriber

自定义消费者,推荐直接编写 BaseSubscriber 的逻辑;

5、背压(Backpressure )和请求重塑(Reshape Requests)

背压 :request(1)

@Overrideprotected void hookOnSubscribe(Subscription subscription) 				 {request(1); //要1个数据}@Overrideprotected void hookOnNext(Integer value) {request(1); //要1个数据}

buffer:缓冲

Flux<List<Integer>> flux = Flux.range(1, 10)  //原始流10个.buffer(3).log();//缓冲区:缓冲3个元素: 消费一次最多可以拿到三个元素; 凑满数批量发给消费者
//
//        //一次发一个,一个一个发;
// 10元素,buffer(3);消费者请求4次,数据消费完成

在这里插入图片描述

6、以编程的方式创建队列

同步环境-generate

    // 同步情况下创建流public void generate() {// 以编程方式创建序列
//        Flux<Object> flux = Flux.generate(sink -> {
//            for (int i = 0; i < 10; i++) {
//                sink.next("哈哈哈" + i);
//            }
//        });//        Flux<Object> flux = Flux.generate(() -> 0, (state, sink) -> {
//            sink.next(state);
//            return state + 1;
//        });Flux<Object> flux = Flux.generate(() -> 0, (state, sink) -> {if (state <= 5) {sink.next(state);} else {sink.complete(); // 流创建完成}return state + 1;});flux.log().subscribe(System.out::println);}

流可以被取消

// 可取消public static void disposable() throws IOException {Flux<Object> flux = Flux.generate(() -> 0, (state, sink) -> {sink.next(state);return state + 1;});// disposable可以被取消Disposable disposable = flux.log().subscribe(System.out::println);new Thread(() ->{try {Thread.sleep(2000);} catch (InterruptedException e) {throw new RuntimeException(e);}disposable.dispose();}).start();System.in.read();}

多线程-create
异步创建flux对象

7、handle()的使用

自定义流中元素处理规则

 // 测试handle// 自定义流中元素处理规则public static void handle() {Flux.range(1,10).handle((value, sink)->{System.out.println("拿到的值:" + value);String v = "转换" + value;sink.next(v);}).subscribe(System.out::println);}
8、自定义线程调度

指定发布者和订阅者的处理线程

 // 自定义线程调度public static void  custom() {Scheduler s = Schedulers.newParallel("parallel-scheduler", 4);final Flux<String> flux = Flux.range(1, 2).map(i -> 10 + i).log().publishOn(s)  // 改变发布者的线程
//                .subscribeOn()  //改变订阅者的线程.map(i -> "value " + i);//只要不指定线程池,默认发布者用的线程就是订阅者的线程;new Thread(() -> flux.subscribe(System.out::println)).start();}

版权声明:

本网仅为发布的内容提供存储空间,不对发表、转载的内容提供任何形式的保证。凡本网注明“来源:XXX网络”的作品,均转载自其它媒体,著作权归作者所有,商业转载请联系作者获得授权,非商业转载请注明出处。

我们尊重并感谢每一位作者,均已注明文章来源和作者。如因作品内容、版权或其它问题,请及时与我们联系,联系邮箱:809451989@qq.com,投稿邮箱:809451989@qq.com