Reactor3 Flux 创建

痛定思痛。 2021-09-16 13:24 427阅读 0赞

## Reactor3 Flux 创建 ##

### 1 创建 Flux ###

#### 1.1 Flux 类中的静态方法 ####

// 1 just()：可以指定序列中包含的全部元素
            // 创建出来的 Flux 序列在发布这些元素之后会自动结束
            Flux.just("1", "2", "3").subscribe(System.out::println);
            // 2 fromArray()，fromIterable()和 fromStream():
            // 可以从一个数组、Iterable 对象或 Stream 对象中创建 Flux 对象
            Flux.fromArray(new Integer[]{ 4, 5, 6}).subscribe(System.out::println);
            // 3 empty()：创建一个不包含任何元素，只发布结束消息的序列
            Flux.empty().subscribe(System.out::println);
            // 4 range(int start, int count)：
            // 创建包含从 start 起始的 count 个数量的 Integer 对象的序列。
            Flux.range(7, 3).subscribe(System.out::println);
            // 5 interval(Duration period)和 interval(Duration delay, Duration period)：
            // 创建一个包含了从 0 开始递增的 Long 对象的序列。其中包含的元素按照指定的间隔来发布。除了间隔时间之外，
            // 还可以指定起始元素发布之前的延迟时间
            // 每2秒产生一个元素
            Flux.interval(Duration.of(2, ChronoUnit.SECONDS)).subscribe(System.out::println);
            // 每1000微秒产生一个元素
            Flux.intervalMillis(1000).subscribe(System.out::println);
            TimeUnit.SECONDS.sleep(5);

#### 1.2 generate()方法 ####

> generate()方法通过同步和逐一的方式来产生 Flux 序列。  
> 序列的产生是通过调用所提供的 SynchronousSink 对象的 next()，complete()和 error(Throwable)方法来完成的。  
> 逐一生成的含义是在具体的生成逻辑中，next()方法只能最多被调用一次。

*  public static <T> Flux <T> generate(Consumer<SynchronousSink<T>> generator)

> SynchronousSink.next()产生元素， SynchronousSink.compelete()结束。

*  public static <T, S> Flux<T> generate(Callable<S> stateSupplier, BiFunction<S, SynchronousSink<T>, S> generator)

> Callable<S> 会提供一个有状态的对象。

*  public static <T, S> Flux<T> generate(Callable<S> stateSupplier, BiFunction<S, SynchronousSink<T>, S> generator, Consumer<? super S> stateConsumer)

> 第三个参数 stateConsumer 是第一个参数stateSupplier的consumer。

//generate
            final AtomicInteger count = new AtomicInteger(1);
            Flux.generate(synchronousSink -> { 
                synchronousSink.next(count + " --> " + new Random().nextInt());
                // 不complete() 将一直产生"Hello" 可以注掉看下效果
                if (count.getAndIncrement() >= 5) { 
                    synchronousSink.complete();
                }
            })
                    .doOnNext(number -> System.out.println(number))
                    .doOnNext(number -> System.out.println(number.toString() + 4))
                    .subscribe();
            System.out.println("-------分割线-------");
    
            // 将names中姓名的首字母转成大写
            final AtomicInteger index = new AtomicInteger(0);
            List<String> names = Arrays.asList("weison", "evan", "elen");
            Flux.generate(() -> names, (nameList, synchronousSink) -> { 
                String s = nameList.get(index.get());
                String s1 = s.substring(0, 1).toUpperCase() + s.substring(1);
                synchronousSink.next(s1);
                if (index.getAndIncrement() == nameList.size() - 1)
                    synchronousSink.complete();
                return nameList;
            }).subscribe(System.out::println);
            System.out.println("-------分割线-------");
    
    
            //第三个参数 stateConsumer 是第一个参数的consumer
            final AtomicInteger index1 = new AtomicInteger(0);
            Flux.generate(() -> names, (nameList, synchronousSink) -> { 
                String s = nameList.get(index1.get());
                String s1 = s.substring(0, 1).toUpperCase() + s.substring(1);
                synchronousSink.next(s1);
                if (index1.getAndIncrement() == nameList.size() - 1)
                    synchronousSink.complete();
                return nameList;
            }, System.out::println).subscribe(System.out::println);

`结果`

--> 1410943965
    2 --> 519628392
    3 --> -1853057073
    4 --> -945435601
    5 --> 510298439
    -------分割线-------
    Weison
    Evan
    Elen
    -------分割线-------
    Weison
    Evan
    Elen
    [weison, evan, elen]

#### 1.3 create()方法 ####

> create()用于阻塞方法与异步代码之间的交互等

Flux.create(sink -> { 
                for (int i = 0; i < 5; i++) { 
                    sink.next(new Random().nextInt());
                }
                sink.complete();
            }).subscribe(System.out::println);
            System.out.println("-------分割线-------");
    
            Flux.create(sink -> { 
                for (int i = 0; i < 5; i++) { 
                    sink.next(new Random().nextInt());
                }
                sink.complete();
            }, FluxSink.OverflowStrategy.BUFFER).subscribe(System.out::println);
            System.out.println("-------分割线-------");
    
            Flux.create((FluxSink<String> sink) -> { 
                List<String> articals = getArticalsFromServer();
                articals.forEach(sink::next);
            }, FluxSink.OverflowStrategy.LATEST)
                    .doOnNext(number -> System.out.println(number))
                    .doOnNext(number -> System.out.println(number + 4))
                    .subscribe(System.out::println);
            System.out.println("-------分割线-------");
    
            Flux.create(fluxSink -> { 
                fluxSink.next("Weison");
                fluxSink.next("Even");
                fluxSink.error(new Exception("Err"));
            }, FluxSink.OverflowStrategy.ERROR)
                    .onErrorReturn("我出错了～")
                    .publishOn(Schedulers.parallel())
                    .subscribe(System.out::println);