响应式编程一、Reactor核心-EW帮帮网

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

一、前置知识

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）响应式编程

在这里插入图片描述

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： 订阅关系绑定的时候触发
            @Override
            protected void hookOnSubscribe(Subscription subscription) {
                // 流被订阅的时候触发
                System.out.println("绑定了..."+subscription);

                //找发布者要数据
                request(1); //要1个数据
//                requestUnbounded(); //要无限数据
            }

            @Override
            protected void hookOnNext(String value) {
                System.out.println("数据到达，正在处理："+value);
                request(1); //要1个数据
            }


            //  hookOnComplete、hookOnError 二选一执行
            @Override
            protected void hookOnComplete() {
                System.out.println("流正常结束...");
            }

            @Override
            protected void hookOnError(Throwable throwable) {
                System.out.println("流异常..."+throwable);
            }

            @Override
            protected void hookOnCancel() {
                System.out.println("流被取消...");
            }

            @Override
            protected void hookFinally(SignalType type) {
                System.out.println("最终回调...一定会被执行");
            }
        });

3、流的取消

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

 flux.subscribe(new BaseSubscriber<String>() {

            // 生命周期钩子1： 订阅关系绑定的时候触发
            @Override
            protected void hookOnSubscribe(Subscription subscription) {
                // 流被订阅的时候触发
                System.out.println("绑定了..."+subscription);

                //找发布者要数据
                request(1); //要1个数据
//                requestUnbounded(); //要无限数据
            }

            @Override
            protected void hookOnNext(String value) {
                System.out.println("数据到达，正在处理："+value);
                if(value.equals("哈哈：5")){
                    cancel(); //取消流
                }
                request(1); //要1个数据
            }


            //  hookOnComplete、hookOnError 二选一执行
            @Override
            protected void hookOnComplete() {
                System.out.println("流正常结束...");
            }

            @Override
            protected void hookOnError(Throwable throwable) {
                System.out.println("流异常..."+throwable);
            }

            @Override
            protected void hookOnCancel() {
                System.out.println("流被取消...");
            }

            @Override
            protected void hookFinally(SignalType type) {
                System.out.println("最终回调...一定会被执行");
            }
        });

4、BaseSubscriber

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

5、背压（Backpressure ）和请求重塑（Reshape Requests）

背压：request(1)

@Override
 protected void hookOnSubscribe(Subscription subscription) 				 {
  request(1); //要1个数据
 }

  @Override
  protected 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();
    }

响应式编程一、Reactor核心

目录

一、前置知识

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、自定义线程调度

网站公告

今日签到

热门文章

最新发布