Reactor之发射器(Flux、Mono)创建函数

Flux

发射0到N个元素的异步"发射器
Reactor之发射器(Flux、Mono)创建函数
文章图片
image
  • Flux是一个标准Publisher,表示0到N个发射项的异步序列,可选地以完成信号或错误终止。与Reactive Streams规范中一样,这三种类型的信号转换为对下游订阅者的onNext、onComplete或onError方法的调用。
  • 在这种大范围的可能信号中,Flux是通用的reactive 类型。注意,所有事件,甚至终止事件,都是可选的:没有onNext事件,但是onComplete事件表示一个空的有限序列,但是移除onComplete并且您有一个无限的空序列(除了关于取消的测试之外,没有特别有用)。同样,无限序列不一定是空的。例如,Flux.interval(Duration) 产生一个Flux,它是无限的,从时钟发出规则的数据。
Mono
发射0到1个元素的异步"发射器
Reactor之发射器(Flux、Mono)创建函数
文章图片
image
  • Mono是一个专门的Publisher,它最多发出一个项,然后可选地以onComplete信号或onError信号结束。
  • 它只提供了可用于Flux的操作符的子集,并且一些操作符(特别是那些将Mono与另一个发布者组合的操作符)切换到Flux。
  • 例如,Mono#concatWith(Publisher)返回一个Flux ,而Mono#then(Mono)则返回另一个Mono。
  • 注意,Mono可以用于表示只有完成概念(类似于Runnable)的无值异步进程。若要创建一个,请使用Mono
创建函数
  • create
【Reactor之发射器(Flux、Mono)创建函数】以编程方式创建具有多次发射能力的Flux,
元素通过FluxSink API以同步或异步方式进行。
eg:
Flux.create((t) -> { t.next("create"); t.next("create1"); t.complete(); }).subscribe(System.out::println);

  • generate
以编程方式创建一个的Flux,通过consumer回调逐一生成信号;generate中next只能调1次,否则会报错 reactor.core.Exceptions$ErrorCallbackNotImplemented
Reactor之发射器(Flux、Mono)创建函数
文章图片
image eg:
Flux.generate(t -> { t.next("generate"); //注意generate中next只能调用1次 t.complete(); }).subscribe(System.out::println);

  • just
创建一个Flux,它发出所提供的元素,然后完成。
Reactor之发射器(Flux、Mono)创建函数
文章图片
image eg:
//单个元素 Flux.just("just").subscribe(System.out::println); //多个元素 Flux.just("just", "just1", "just2").subscribe(System.out::println);

  • from
用Flux API装饰指定的Publisher,通过Publisher创建一个Flux
Reactor之发射器(Flux、Mono)创建函数
文章图片
image eg:
//Flux->Flux Flux.from(Flux.just("just", "just1", "just2")) .subscribe(System.out::println); //Mono->Mono Flux.from(Mono.just("just")).subscribe(System.out::println);

  • fromArray
创建一个Flux,它发出包含在提供的数组中的项。
Reactor之发射器(Flux、Mono)创建函数
文章图片
image eg:
Flux.fromArray(new String[] { "arr", "arr", "arr", "arr" }) .subscribe(System.out::println);

  • fromIterable
创建一个个Flux,它发出所提供的Iterable中包含的项。将为每个subscriber创建一个新的Iterable。
Reactor之发射器(Flux、Mono)创建函数
文章图片
image eg:
Set v = new HashSet<>(); v.add("1"); v.add("2"); v.add("3"); Flux.fromIterable(() -> v.iterator()).subscribe(System.out::println);

  • fromStream
创建一个Flux,它发出所提供的Stream中包含的项。请记住,Stream不能被重新使用,这可能是有问题的。多订阅或重订阅的情况(如repeat或retry)Stream是closed由操作员取消,错误或完成。
Reactor之发射器(Flux、Mono)创建函数
文章图片
image
  • defer
每当对得到的Flux进行Subscription时,延迟提供Publisher,因此实际的源实例化被推迟,直到每个订阅和Supplier可以创建订阅者特定的实例。
但是,如果供应商没有生成新的实例,这个操作符将有效地从Publisher起作用。
Reactor之发射器(Flux、Mono)创建函数
文章图片
image eg:
Flux.defer(() -> Flux.just("just", "just1", "just2")) .subscribe(System.out::println);

  • interval
创建一个Flux,它以0开始发射长值并递增
全局计时器上指定的时间间隔。如果需求没有及时产生,一个OnError将用来发出信号。IllegalStateException详细说明无法发出的信息。在正常情况下,Flux将永远不会完成。
Reactor之发射器(Flux、Mono)创建函数
文章图片
image eg:
Flux.interval(Duration.of(500, ChronoUnit.MILLIS)) .subscribe(System.out::println); //防止程序过早退出,放一个CountDownLatch拦住 CountDownLatch latch = new CountDownLatch(1); latch.await();

  • empty
创建一个Flux,完成而不发射任何项目。
Reactor之发射器(Flux、Mono)创建函数
文章图片
image eg:
Flux.empty().subscribe(System.out::println);

  • error
创建一个Flux,它在订阅之后立即以指定的错误终止。
Reactor之发射器(Flux、Mono)创建函数
文章图片
image eg:
Flux.error(new RuntimeException()).subscribe(System.out::println);

  • never
创建一个Flux,它永远不会发出任何数据、错误或完成信号。
Reactor之发射器(Flux、Mono)创建函数
文章图片
image eg:
Flux.never().subscribe(System.out::println);

  • range
建立一个Flux,它只会发出一个count递增整数的序列,从start开始。也就是说,在start(包含)和start + count(排除)之间发出整数,然后完成。
Reactor之发射器(Flux、Mono)创建函数
文章图片
image eg:
Flux.range(0, 100).subscribe(System.out::println);

    推荐阅读