spring-in-action-初识反应式编程reactor

一：反应式编程

　　　　1.反应式编程的优点

　　　　1)无阻塞

　　　　 2)处理的数据可以是无限的

　　　　　　打水仗：

　　　　　　命令式编程：使用的是水气球。水气球的容量有限。把水气球装水，再扔向对方。如果想要更多的水，就需要更多的水气球。

　　　　　　反应式编程：使用的是水管。只要水管的另一头不断的输入水，那么这边喷出的水就不会断。

二：反应式流

　　1.基本概念

　　　　他是一种规范。是对无阻塞-回压-异步流处理的标准

　　　　　他是一个异步编程范式，主要涉及数据流及变化的传播，可以看做是观察者设计模式的扩展。

java里头的iterator是以pull模型，即订阅者使用next去拉取下一个数据；而reactive streams则是以push模型为主

　　　　2.JAVA流和反应式流的基本区别

　　　　Java流：通常是同步的，且处理的数据有限

　　　　反应式流：支持任意大小的数据集，包括无限的。且只要数据就绪，就可以实时的处理数据

　　　　3.反应式流规范的定义

　　　　反应式流的规范主要由4个接口组成　　　　　

　　　　(1) Publisher 出版商 - 定义了发布者的方法

　　　　(2) Subscriber 用户 - 接口定义了订阅者的方法

　　　　(3) Subscription 订阅 - 控制数据开始和结束发送，且可以控制数据量

　　　　(4) Processor 他是Publisher 和 Subscriber 的结合。定义了处理器

　　　　1)Publisher源码

        public interface Publisher<T> {

        /**

            消费者订阅

         */

        public void subscribe(Subscriber<? super T> s);

    }

　　　　2）Subscriber源码

        public interface Subscriber<T> {

    /**

         接收Subscription 对象，获得的Subscription 对象可以控制数据传输开始和结束以及数据量

     */

    public void onSubscribe(Subscription s);

    /**

            获取下一个数据

     */

    public void onNext(T t);

    /**

     如果出错了，就会调用本方法

     */

    public void onError(Throwable t);

    /**

    如果publisher没有数据了，也就是数据传输完成了，会调用本方法

     */

    public void onComplete();

}

　　　　3）Subscription源码

public interface Subscription {

    /**

        请求publisher发送数据， n表示数据量

     */

    public void request(long n);

    /**

    取消订阅

     */

    public void cancel();

}

　　　　4）Processor源码

public interface Processor<T, R> extends Subscriber<T>, Publisher<R> {

}

三：spring reactor

　　1.基本概念

　　　　他提供了反应式流规范的实现，提供了一组用于组装反应式流的函数式API

　　　　　　reactor是push数据，当Subscriber调用onSubscribe方法是，传入一个Subscriber对象，此时，Subscriber对象的onSubscribe(Subscription subscription)方法会被调用，订阅者调用subscribe方法订阅，发布者调用订阅者的onNext通知订阅者新消息。也就是说，是发布者主动把数据推送给订阅者。

　　被订阅者 (Publisher) 主动推送数据给 订阅者 (Subscriber)，触发 onNext() 方法。异常和完成时触发另外两个方法。

　　被订阅者 (Publisher) 发生异常，则触发 订阅者 (Subscriber) 的 onError() 方法进行异常捕获处理。

　　被订阅者 (Publisher) 每次推送都会触发一次 onNext() 方法。所有的推送完成且无异常时，onCompleted() 方法将 在最后 触发一次

        @Test
    public void testBackpressure(){
        Flux.just(1, 2, 3, 4)
                .log()
                .subscribe(new Subscriber<Integer>() {
                    private Subscription s;
                    int onNextAmount;

                    @Override
                    public void onSubscribe(Subscription s) {
                        this.s = s;
                        s.request(2);
                    }

                    @Override
                    public void onNext(Integer integer) {
                        try {
                            Thread.sleep(500);
                        } catch (InterruptedException e) {
                            // TODO Auto-generated catch block
                            e.printStackTrace();
                        }
                        System.out.println(integer);
                        onNextAmount++;
                        if (onNextAmount % 2 == 0) {
                            s.request(2);
                        }
                    }

                    @Override
                    public void onError(Throwable t) {}

                    @Override
                    public void onComplete() {}
                });

        try {
            System.out.println("aaa");
        } catch (Exception e) {
            e.printStackTrace();
        }

/*
        日志打印和输出结果

16:41:48.160 [main] DEBUG reactor.util.Loggers$LoggerFactory - Using Slf4j logging framework
16:41:48.174 [main] INFO reactor.Flux.Array.1 - | onSubscribe([Synchronous Fuseable] FluxArray.ArraySubscription) //自动调用onSubscribe方法
16:41:48.178 [main] INFO reactor.Flux.Array.1 - | request(2)　　//调用request方法表示结束数据2个
16:41:48.178 [main] INFO reactor.Flux.Array.1 - | onNext(1)　　　//调用onNext方法接收数据
1
16:41:48.680 [main] INFO reactor.Flux.Array.1 - | onNext(2)　　//调用onNext方法接收数据
2
16:41:49.180 [main] INFO reactor.Flux.Array.1 - | request(2)　　//调用request方法表示结束数据2个
16:41:49.180 [main] INFO reactor.Flux.Array.1 - | onNext(3)　　//调用onNext方法接收数据
3
16:41:49.681 [main] INFO reactor.Flux.Array.1 - | onNext(4)　　//调用onNext方法接收数据
4
16:41:50.182 [main] INFO reactor.Flux.Array.1 - | request(2)　　//调用request方法表示结束数据2个
16:41:50.183 [main] INFO reactor.Flux.Array.1 - | onComplete() //发现没有数据了，结束
aaa
        
        */

　　　　2.Reactor提供的两个核心反应式类型

　　　　　　(1) Mono：允许0-1个数据项

　　　　　　(2) Flux：允许任意数量的数据项

　　　　3.Reactor两个核心反应式类型的操作使用

　　　　1)基本说明：由于Mono和Flux类似，且Flux使用频率更大，所以只拿Flux举例

　　　　　 2)Flux的提供的操作可分为4类　　

　　　　　　　　① 创建

　　　　　　　　② 组合

　　　　　　　　③ 转换

　　　　　　　　④ 逻辑

　　　　　3)创建操作

　　　　　　　（1）Flux.just创建Flux　

　　　　　　　　下面使用　Flux.just创建了一个字符串Flux-just并且通过just.subscribe来订阅flux，把数据打印出来以验证输出的数据是否正确

    @Test

    public void test1(){

        //创建一个flux

        Flux<String> just = Flux.just("字符串1","字符串2","字符串3","字符串4","字符串5","字符串6");

        //订阅flux - 这里没有中间操作，数据直接到了订阅者

        just.subscribe(x->{

            System.out.println(x);

        });

    }

　　　　　　（2）使用StepVerifier来订阅和验证Flux

　　　　　　　　上面我们通过just.subscribe来订阅flux，并且把数据打印出来以验证输出的数据是否正确，Reacotr提供了StepVerifier来订阅及验证Flux　

　　@Test

    public void test2(){ //使用StepVerifier 来验证flux

        //创建一个flux

        Flux<String> just = Flux.just("字符串1","字符串2","字符串3","字符串4","字符串5","字符串6");

        //使用StepVerifier来测试flux

        StepVerifier.create(just)  //StepVerifier订阅flux

        .expectNext("字符串1")    //校验数据

        .expectNext("字符串2")

        .expectNext("字符串3")

        .expectNext("字符串4")

        .expectNext("字符串5")

        .expectNext("字符串6")

        .verifyComplete(); //校验正常完成

    }

　　　　　　(3）使用数组来创建flux-Flux.fromArray　　　

    @Test

    public void test3(){  //使用array创建flux

        String[] arr = {"字符串1","字符串2","字符串3","字符串4","字符串5","字符串6"};

        //创建一个flux

        Flux<String> just = Flux.fromArray(arr);

        //使用StepVerifier来测试flux

        StepVerifier.create(just)  //StepVerifier订阅flux

        .expectNext("字符串1")    //校验数据

        .expectNext("字符串2")

        .expectNext("字符串3")

        .expectNext("字符串4")

        .expectNext("字符串5")

        .expectNext("字符串6")

        .verifyComplete(); //校验正常完成

    }

　　　　　　(4）使用集合来创建flux-Flux.fromIterable

    @Test

    public void test4(){  //使用集合创建flux

        String[] arr = {"字符串1","字符串2","字符串3","字符串4","字符串5","字符串6"};

        List<Object> asList = Arrays.asList(arr);

        //创建一个flux

        Flux<Object> just = Flux.fromIterable(asList);

        //使用StepVerifier来测试flux

        StepVerifier.create(just)  //StepVerifier订阅flux

        .expectNext("字符串1")    //校验数据

        .expectNext("字符串2")

        .expectNext("字符串3")

        .expectNext("字符串4")

        .expectNext("字符串5")

        .expectNext("字符串6")

        .verifyComplete(); //校验正常完成

    }

　　　　　(5)使用流创建Flux-Flux.fromStream

       @Test

    public void test5(){  //使用流创建flux

        String[] arr = {"字符串1","字符串2","字符串3","字符串4","字符串5","字符串6"};

        List<Object> asList = Arrays.asList(arr);

        Stream<Object> stream = asList.stream();

        //创建一个flux

        Flux<Object> just = Flux.fromStream(stream);

        //使用StepVerifier来测试flux

        StepVerifier.create(just)  //StepVerifier订阅flux

        .expectNext("字符串1")    //校验数据

        .expectNext("字符串2")

        .expectNext("字符串3")

        .expectNext("字符串4")

        .expectNext("字符串5")

        .expectNext("字符串6")

        .verifyComplete(); //校验正常完成

    }

　　　　　　(6)生成区间内的整数的flux

        @Test

    public void test6(){  //使用flux生成区间内的整数

        //创建一个flux

        Flux<Integer> just = Flux.range(5, 6); //从5开始，生成6个连续的整数

        //使用StepVerifier来测试flux

        StepVerifier.create(just)  //StepVerifier订阅flux

        .expectNext(5)    //校验数据

        .expectNext(6)

        .expectNext(7)

        .expectNext(8)

        .expectNext(9)

        .expectNext(10)

        .verifyComplete(); //校验正常完成

    }

　　　　　　(7)使用flux按时间间隔生成正整数-Flux.interval

        @Test

    public void test7(){  //使用flux按时间间隔生成正整数

        System.out.println("test7");

        //创建一个flux

        Flux<Long> just = Flux.interval(Duration.ofSeconds(2)).take(6);

        //Duration.ofSeconds(2)每2秒生成一个数字，由0开始，take表示生成6个数字

        //使用StepVerifier来测试flux

        StepVerifier.create(just)  //StepVerifier订阅flux

        .expectNext(0L)    //校验数据

        .expectNext(1L)

        .expectNext(2L)

        .expectNext(3L)

        .expectNext(4L)

        .expectNext(5L)

        .verifyComplete(); //校验正常完成

    }

　　　　4)组合操作

　　　　　　mergeWith

　　　　　　　　把两个Flux组合形成一个新的Flux，新的Flux中的数据顺序和两个Flux的发布数据的时间顺序相同

　　　　　　　(1)使用mergeWith将just和just2组合成mergeWith　　

　　　　　　　　下面代码吧just和just2组合生成一个新的Flux-mergeWith,这里我们注意mergeWith打印的数据的顺序,just打印完后再打印的just2的数据。

　　@Test

    public void test0(){

        //创建一个flux

        Flux<String> just = Flux.just("字符串1","字符串2");

        Flux<String> just2 = Flux.just("字符串1x","字符串2x");

        //订阅flux - 这里没有中间操作，数据直接到了订阅者

        Flux<String> mergeWith = just.mergeWith(just2);

        mergeWith.subscribe(x ->{

            Thread t = Thread.currentThread();

            String name = t.getName();

            System.out.println(name +"-"+ x);

        });

　　　　/* main-字符串1

　　　　　　main-字符串2

　　　　　　main-字符串1x

　　　　　　main-字符串2x

}

　　　　　　(2)使用mergeWith将just和just2组合成mergeWith,且使用delaySubscription方法和delayElements方法来控制组合的数据的顺序　　　　　

　　 @Test

    public void test2(){

        //创建一个flux

        Flux<String> just = Flux.just("字符串1","字符串2").delayElements(Duration.ofSeconds(1)); //每一个数据隔1s发布(第一个数据也是隔1s才发布)

        Flux<String> just2 = Flux.just("字符串1x","字符串2x").

                delaySubscription(Duration.ofMillis(500)). //订阅后延迟0.5s发布开始数据

                delayElements(Duration.ofSeconds(1));    ////每一个数据隔1s发布(第一个数据也是隔1s才发布)

        //订阅flux - 这里没有中间操作，数据直接到了订阅者

        Flux<String> mergeWith = just.mergeWith(just2);

        mergeWith.subscribe(x->{

            Thread t = Thread.currentThread();

            String name = t.getName();

            System.out.println(name +"=="+ x);

        });

        /*

        parallel-1-字符串1

        parallel-5-字符串1x

        parallel-7-字符串2

        parallel-1-字符串2x

        */

        StepVerifier.create(mergeWith)  //如果把这段代码去掉，那么打印结果就为空

        .expectNext("字符串1")

        .expectNext("字符串1x")

        .expectNext("字符串2")

        .expectNext("字符串2x")

        .verifyComplete();

        Thread t = Thread.currentThread();

        String name = t.getName();

        System.out.println(name +"线程");

//打印结果 main线程

        System.out.println("结束");

    }

　　　　　　使用delayElements方法对just进行控制，指定它每个数据发布间隔1秒

　　　　　　使用delayElements方法对just2进行控制，指定它每个数据发布间隔1秒

　　　　　　使用delaySubscription方法对just2进行控制，指定开始时延迟0.5秒发布数据

　　　　　　注意：delayElements对第一个数据同样有效，也就是说just等了1秒第一个数据发布，后面每隔1秒发布一个数据，而just2等了1.5秒(0.5+1)发布第一个数据，后面每隔1秒发布一个数据

　　　　　　Flux特性:这里还有一个有意思的发现，这里出现了4个线程 parallel-1、 parallel-5、 parallel-7、main。数据在管道上处理的每一步，可能是同一个线程，也可能是不同的线程，我们不能判断数据会在哪个线程上执行

　　　　　　组合后Flux数据顺序体现：我们使用delayElements和delaySubscription来控制两个源数据的发布顺序，而新的Flux的数据输出和我们预期的一致，也就验证了组合后的Flux的数据顺序和组合的两个Flux的数据的发布顺序相同

　　　　zipWith

　　　　　　把两个Flux组合形成一个新的Flux，默认情况下他会把两个源flux的数据成对的放入Tuple2，形成一个Tuple2的Flux,也可以自定义数据的组合方式

　　　　　(1)使用zipwith来组合两个Flux　　

　　　　　　下面使用zipwith把just和just2组合生成一个新的Tuple2的Flux

    @Test

    public void testz1(){

        //创建一个flux

        Flux<String> just = Flux.just("字符串1","字符串2");

        Flux<String> just2 = Flux.just("字符串1x","字符串2x");

        //订阅flux - 这里没有中间操作，数据直接到了订阅者

         Flux<Tuple2<String, String>> mergeWith = just.zipWith(just2);  //返回的是一个Tuple2的一个flux，他会把两个源flux的数据承兑放入Tuple2

        mergeWith.subscribe(x ->{

            Thread t = Thread.currentThread();

            String name = t.getName();

            System.out.println(name +"-"+ x);

        });

    /*

        main-[字符串1,字符串1x]

        main-[字符串2,字符串2x]

     */

        }

　　　　　　(2)当两个Flux的数据数量不一致时，形成的Tuple2的Flux中的数据将不会录入无法形成一对一的数据

　　　　　　just有4个数据，just2只有2个数据，使用zipmap组合后结果生成了包含2个Tuple2的flux。just的后面的两个数据没有进入新的Flux

　　@Test

    public void testz2(){

        //创建一个flux

        Flux<String> just = Flux.just("字符串1","字符串2","字符串3","字符串4");

        Flux<String> just2 = Flux.just("字符串1x","字符串2x");

         Flux<Tuple2<String, String>> mergeWith = just.zipWith(just2); //返回的是一个Tuple2的一个flux，他会把两个源flux的数据承兑放入Tuple2，如果两个flux的数据数量不一致，多的无法组成Tuple2，则多的数据不会进入新的flux

        mergeWith.subscribe(x ->{

            Thread t = Thread.currentThread();

            String name = t.getName();

            System.out.println(name +"-"+ x);

        });

    /*

        main-[字符串1,字符串1x]

        main-[字符串2,字符串2x]

     */

    }

　　　　　　(3)自定义组合方式

　　　　　　下面把两个Flux的数据按照数据1&&数据2的格式拼接组合生成一个String的flux，而不是生成Tuple2的flux

　　@Test

    public void testz3(){

        //创建一个flux

        Flux<String> just = Flux.just("字符串1","字符串2");

        Flux<String> just2 = Flux.just("字符串1x","字符串2x","字符串3x","字符串4x");

        Flux<String> mergeWith = Flux.zip(just, just2,(x,y)->{

            return x + "&&" + y;

        });

            mergeWith.subscribe(x ->{

            Thread t = Thread.currentThread();

            String name = t.getName();

            System.out.println(name +"-"+ x);

        });

    /*

        main-字符串1&&字符串1x

        main-字符串2&&字符串2x

     */

    }

　　　　5)中间操作

　　　　Flux的中间操作和Java8的Stream的中间操作基本一致　

1. skip 跳过指定数量的数据

2.take 只取前面指定数量的数据

3. filter过滤

4. Distinct 去重

5. sort排序

6.map

7.flatmap

　　　　　　(1)skip跳过指定数量的数据　

　　@Test

    public void testx1(){

        //创建一个flux

        Flux<String> just = Flux.just("字符串1x","字符串2x","字符串3x","字符串4x").skip(2); //跳过前两个数据

        just.subscribe(x ->{

            Thread t = Thread.currentThread();

            String name = t.getName();

            System.out.println(name +"-"+ x);

        });

    /*

        main-字符串3x

        main-字符串4x

     */

    }

　　　　　　(2)take只取前面指定数量的数据　　　

　　@Test

    public void testx4(){

        //创建一个flux

        Flux<String> just = Flux.just("字符串1x","字符串2x","字符串3x","字符串4x").take(2); //只取前两个数据

        just.subscribe(x ->{

            Thread t = Thread.currentThread();

            String name = t.getName();

            System.out.println(name +"-"+ x);

        });

    /*

        main-字符串1x

        main-字符串2x

     */

    }

　　　　　　(3)filter过滤

　　@Test

    public void testx5(){

        //创建一个flux

        Flux<String> just = Flux.just("字符串1x","字符串2x","字符串3x","字符串4x","字符串","字符串5x").filter(x->{

            Thread t = Thread.currentThread();

            String name = t.getName();

            System.out.println(name + "-filter过滤中");

            return x.contains("x");

        }); //只取前两个数据

        just.subscribe(x ->{

            Thread t = Thread.currentThread();

            String name = t.getName();

            System.out.println(name +"-"+ x);

        });

    /*

        main-filter过滤中

        main-字符串1x

        main-filter过滤中

        main-字符串2x

        main-filter过滤中

        main-字符串3x

        main-filter过滤中

        main-字符串4x

        main-filter过滤中

        main-filter过滤中

        main-字符串5x

     */

    }

　　　　　　(4)Distinct 去重　　　

　　@Test

    public void testx6(){

        //创建一个flux

        Flux<String> just = Flux.just("字符串1x","字符串2x","字符串3x","字符串4x","字符串","字符串").distinct(); //只取前两个数据

        just.subscribe(x ->{

            Thread t = Thread.currentThread();

            String name = t.getName();

            System.out.println(name +"-"+ x);

        });

    /*

        main-字符串1x

        main-字符串2x

        main-字符串3x

        main-字符串4x

        main-字符串

     */

    }

　　　　　　(5)sort 排序

　　　　　　自然排序　　　　

　　@Test

    public void testx7(){

        //创建一个flux

        Flux<String> just = Flux.just("字符串1x","字符串2x","字符串3x","字符串4x","字符串7","字符串5").sort(); //只取前两个数据

        just.subscribe(x ->{

            Thread t = Thread.currentThread();

            String name = t.getName();

            System.out.println(name +"-"+ x);

        });

    /*

        main-字符串1x

        main-字符串2x

        main-字符串3x

        main-字符串4x

        main-字符串5

        main-字符串7

     */

    }

　　　　　　自定义排序　　　

@Test

    public void testx8(){

        //创建一个flux

        Flux<Integer> just = Flux.just(1,3,2,6,9,0,5,3).sort((x,y)->{

            return y.compareTo(x); //倒叙

        }); //只取前两个数据

        just.subscribe(x ->{

            Thread t = Thread.currentThread();

            String name = t.getName();

            System.out.println(name +"-"+ x);

        });

    /*

        main-9

        main-6

        main-5

        main-3

        main-3

        main-2

        main-1

        main-0

     */

    }

　　　　　　(6)map　

　　　　　　对Flux的数据进行处理后形成新的Flux，注意，只有一个主线程main

　　@Test

    public void testx9(){

        //创建一个flux

        Flux<Integer> just = Flux.just(1,3,2,6).map(x->{

            Thread t = Thread.currentThread();

            String name = t.getName();

            System.out.println(name +"-map处理中");

            return x * x;

        });

        just.subscribe(x ->{

            Thread t = Thread.currentThread();

            String name = t.getName();

            System.out.println(name +"-"+ x);

        });

    /*

        main-map处理中

        main-1

        main-map处理中

        main-9

        main-map处理中

        main-4

        main-map处理中

        main-36

     */

    }

　　　　　　(7)flatmap

　　　　　　对Flux的数据进行处理后形成的Flux合流形成新的Flux，注意，只有一个主线程main

　　@Test  //只有一个主线程main

    public void testx10() throws InterruptedException{

        //创建一个flux

        Flux just = Flux.just(1,3,2,6).flatMap(x->{

            Thread t = Thread.currentThread();

            String name = t.getName();

            System.out.println(name +"-" + "flatmap转换中");

            return Flux.just(x).map(a->{

                Thread t1 = Thread.currentThread();

                String name1 = t.getName();

                System.out.println(name1 +"-" + "map转换中");

                return a * a;

            });

        });

        just.subscribe(x ->{

            Thread t = Thread.currentThread();

            String name = t.getName();

            System.out.println(name +"-"+ x);

        });

        Thread.sleep(3000);

    /*

        main-flatmap转换中

        main-map转换中

        main-1

        main-flatmap转换中

        main-map转换中

        main-9

        main-flatmap转换中

        main-map转换中

        main-4

        main-flatmap转换中

        main-map转换中

        main-36

     */

    }

　　　　　　(8)flatmap+subscribeOn　　

　　　　　　使用了subscribeOn后，为订阅者开辟了一个新的线程parallel-1

　　　　　@Test    //使用了subscribeOn 单独开了一个新的线程parallel

        public void testx11() throws InterruptedException{

            //创建一个flux

            Flux just = Flux.just(1,3,2,6).flatMap(x->{

                Thread t = Thread.currentThread();

                String name = t.getName();

                System.out.println(name +"-" + "flatmap转换中");

                return Flux.just(x).map(a->{

                    Thread t1 = Thread.currentThread();

                    String name1 = t.getName();

                    System.out.println(name1 +"-" + "map转换中");

                    return a * a;

                });

            }).subscribeOn(Schedulers.parallel());

            just.subscribe(x ->{

                Thread t = Thread.currentThread();

                String name = t.getName();

                System.out.println(name +"-"+ x);

            });

            Thread.sleep(3000);

        /*

            parallel-1-flatmap转换中

            parallel-1-map转换中

            parallel-1-1

            parallel-1-flatmap转换中

            parallel-1-map转换中

            parallel-1-9

            parallel-1-flatmap转换中

            parallel-1-map转换中

            parallel-1-4

            parallel-1-flatmap转换中

            parallel-1-map转换中

            parallel-1-36

         */

        }

Flatmap使用了subscribeOn(Schedulers.parallel())会让多个订阅者并行执行

　　subscribeOn 方法Schedulers支持的并发模型说明

　　.immediate()在当前线程中执行订阅

　　.single()在一个单一的，可重复的线程中执行订阅。对所有的订阅者重用相同的线程

　　.newSingle()针对每个订阅，使用专门的线程

　　.elastic() 在从无界弹性线程池中拉取的线程中执行订阅。根据需要创建新的工作线程，并销毁空闲的工作线程(默认情况下，空闲60s后销毁)

　　.parallel()从一个固定大小的线程池中拉取线程执行订阅，该线程池大大小和CPU的核心数一致

　　　　　　(9)Collection操作

　　　　　　　　@Test

            public void testx21(){

                //创建一个flux

                Mono<List<Integer>> just = Flux.just(1,3,2,6).map(x->{

                    Thread t = Thread.currentThread();

                    String name = t.getName();

                    System.out.println(name +"-map处理中");

                    return x * x;

                }).collect(Collectors.toList());

                just.subscribe(x ->{

                    Thread t = Thread.currentThread();

                    String name = t.getName();

                    System.out.println(name +"-"+ x);

                });

            }

@Test

            public void testx22(){

                //创建一个flux

                Mono<ConcurrentMap<Integer, List<Integer>>> just = Flux.just(1,3,2,6).map(x->{

                    Thread t = Thread.currentThread();

                    String name = t.getName();

                    System.out.println(name +"-map处理中");

                    return x * x;

                }).collect(Collectors.groupingByConcurrent(x->{

                    return x;

                }));

                just.subscribe(x ->{

                    Thread t = Thread.currentThread();

                    String name = t.getName();

                    System.out.println(name +"-"+ x);

                });

            }

　　　　　　　(10)map有多个订阅者

　　　　　　下面just有两个订阅者。我们发现map这段代码被执行了两次，线程还是只有一个主线程main

　　　　　　@Test    //两个订阅者  我们发现flatmap的操作执行了两遍 - 还是只有主线程main

        public void testx12() throws InterruptedException{

            //创建一个flux

            Flux just = Flux.just(1,3,2,6).flatMap(x->{

                Thread t = Thread.currentThread();

                String name = t.getName();

                System.out.println(name +"-" + "flatmap转换中");

                return Flux.just(x).map(a->{

                    Thread t1 = Thread.currentThread();

                    String name1 = t.getName();

                    System.out.println(name1 +"-" + "map转换中");

                    return a * a;

                });

            });

            just.subscribe(x ->{

                Thread t = Thread.currentThread();

                String name = t.getName();

                System.out.println(name +"-"+ x);

            });

            just.subscribe(x ->{

                Thread t = Thread.currentThread();

                String name = t.getName();

                System.out.println(name +"*"+ x);

            });

            Thread.sleep(3000);

        /*

            main-flatmap转换中

            main-map转换中

            main-1

            main-flatmap转换中

            main-map转换中

            main-9

            main-flatmap转换中

            main-map转换中

            main-4

            main-flatmap转换中

            main-map转换中

            main-36

            main-flatmap转换中

            main-map转换中

            main*1

            main-flatmap转换中

            main-map转换中

            main*9

            main-flatmap转换中

            main-map转换中

            main*4

            main-flatmap转换中

            main-map转换中

            main*36

         */

        }

　　　　　　(11)flatmap + subscribeOn多个订阅者

　　　　just有两个订阅者，flatmap这段代码执行了两次，且每个订阅者都开辟了一个线程：parallel-1，parallel-2。这里很好的体现了reactor的特点，异步并行处理

@Test//flatmap使用了subscribeOn两个订阅者  我们发现flatmap的操作执行了两遍 - 每一个订阅者都创建了一个新的线程在执行

            public void testx14() throws InterruptedException{

                //创建一个flux

                Flux just = Flux.just(1,3,2,6).flatMap(x->{

                    Thread t = Thread.currentThread();

                    String name = t.getName();

                    System.out.println(name +"-" + "flatmap转换中");

                    return Flux.just(x).map(a->{

                        Thread t1 = Thread.currentThread();

                        String name1 = t.getName();

                        System.out.println(name1 +"-" + "map转换中");

                        return a * a;

                    });

                }).subscribeOn(Schedulers.parallel());

                just.subscribe(x ->{

                    Thread t = Thread.currentThread();

                    String name = t.getName();

                    System.out.println(name +"-"+ x);

                });

                just.subscribe(x ->{

                    Thread t = Thread.currentThread();

                    String name = t.getName();

                    System.out.println(name +"*"+ x);

                });

                Thread.sleep(3000);

            /*

                parallel-1-flatmap转换中

                parallel-2-flatmap转换中

                parallel-1-map转换中

                parallel-2-map转换中

                parallel-1-1

                parallel-2*1

                parallel-1-flatmap转换中

                parallel-2-flatmap转换中

                parallel-1-map转换中

                parallel-2-map转换中

                parallel-1-9

                parallel-2*9

                parallel-1-flatmap转换中

                parallel-2-flatmap转换中

                parallel-1-map转换中

                parallel-2-map转换中

                parallel-1-4

                parallel-2*4

                parallel-1-flatmap转换中

                parallel-2-flatmap转换中

                parallel-1-map转换中

                parallel-2-map转换中

                parallel-1-36

                parallel-2*36

             */

            }

　　　　总结：每一个订阅者，Flux的中间操作都会执行一遍

　　　　6)逻辑操作

　　　　　　(1)ALL全部匹配返回true

　　　　　　　　@Test

            public void testx23(){

                //创建一个flux

                Mono<Boolean> just = Flux.just(1,3,2,6).map(x->{

                    Thread t = Thread.currentThread();

                    String name = t.getName();

                    System.out.println(name +"-map处理中");

                    return x * x;

                }).all(y ->{

                    return y > 4;

                });

                just.subscribe(x ->{

                    Thread t = Thread.currentThread();

                    String name = t.getName();

                    System.out.println(name +"-"+ x);

                });

            }

　　　　　　(2)ANY任意一个匹配返回true

　　　　　　　　@Test

            public void testx24(){

                //创建一个flux

                Mono<Boolean> just = Flux.just(1,3,2,6).map(x->{

                    Thread t = Thread.currentThread();

                    String name = t.getName();

                    System.out.println(name +"-map处理中");

                    return x * x;

                }).any(y ->{

                    return y > 4;

                });

                just.subscribe(x ->{

                    Thread t = Thread.currentThread();

                    String name = t.getName();

                    System.out.println(name +"-"+ x);

                });

            }

巴特西

spring-in-action-初识反应式编程reactor

最新文章

热门文章