响应式编程——初识 Flux 和 Mono

旧城等待， 2023-10-15 12:29 21阅读 0赞

## 前言 ##

Reactor 是一个响应式编程的基础类库，其中有两个很关键的类：Flux 和 Mono。掌握这两个类和相关概念有助于我们学习响应式编程。

Flux 和 Mono 都是数据流的发布者，使用 Flux 和 Mono 都可以发出三种数据信号：元素值，错误信号，完成信号；错误信号和完成信号都代表终止信号，终止信号用于告诉订阅者数据流结束了，错误信号终止数据流同时把错误信息传递给订阅者。

## 一、Flux ##

具有 rx 运算符的响应式流发布器，发出 0 到 N 个元素，然后完成（成功或有错误）。

下图显示了 Flux 如何转换项目：

![Flux转换项目][Flux]

Flux是一个标准的Publisher，表示一个异步的0到N个发出的项目序列，可选择终止于完成信号或错误信号。根据Reactive Streams规范，这三种类型的信号转换为对下游Subscriber的onNext、onComplete和onError方法的调用。  
由于可能出现的信号范围很大，Flux是通用的响应式类型。请注意，所有事件，包括终止事件，都是可选的：没有onNext事件但有onComplete事件表示一个空的有限序列，但如果去掉onComplete，则得到一个无限的空序列（除了用于取消测试之外，不是特别有用）。同样，无限序列不一定为空。例如，Flux.interval(Duration)会生成一个无限的Flux，从时钟发出定期的滴答声。Flux 是标准的 Publisher，它表示 0 到 N 个发出项的异步序列，可以选择由完成信号或错误终止。与 Reactive Streams 规范中一样，这三种类型的信号转换为对下游订阅者的 onNext、onComplete 和 onError 方法的调用。

凭借如此大范围的可能信号，Flux 是通用的无功类型。请注意，所有事件，甚至终止事件，都是可选的：没有 onNext 事件，但 onComplete 事件表示一个空的有限序列，但删除 onComplete 并且您有一个无限的空序列（不是特别有用，除了围绕取消的测试）。同样，无限序列不一定是空的。例如， Flux.interval(Duration) 生成无限的 Flux 并从时钟发出规则的滴答声。

## 二、Mono ##

具有基本 rx 运算符的 Reactive Streams Publisher 通过 onNext 信号最多发出一项，然后以 onComplete 信号终止（成功的 Mono，有或没有值），或者仅发出单个 onError 信号（失败的 Mono）。

下图显示了 Mono 如何转换项目：

![Mono 转换项目][Mono]

Mono是一种特殊的Publisher，通过onNext信号发出最多一个项目，然后通过onComplete信号终止（成功的Mono，有或没有值），或者只发出一个onError信号（失败的Mono）。  
大多数Mono实现在调用onNext后立即调用其Subscriber的onComplete。Mono.never()是一个例外：它不发出任何信号，在技术上并不禁止，但在测试之外没有太大用处。另一方面，明确禁止使用onNext和onError的组合。  
Mono只提供了Flux可用的操作符的子集，而某些操作符（特别是将Mono与另一个Publisher组合的操作符）会切换到Flux。例如，MonoconcatWith(Publisher)返回一个Flux，而Monothen(Mono)返回另一个Mono。  
请注意，您可以使用Mono来表示只有完成概念的无值异步过程（类似于Runnable）。要创建一个，您可以使用一个空的Mono。

## 三、代码示例 ##

> 创建一个Flux，发出一系列字符串元素并订阅打印出来：
> 
>     package top.emanjusaka; import reactor.core.publisher.Flux; public class Main { public static void main(String[] args) { Flux<String> flux = Flux.just("Hello", "emanjusaka", "!"); flux.subscribe(System.out::println); } } // 输出 Hello emanjusaka !

> 创建一个Mono，发出一个字符串元素并订阅打印出来：
> 
> `package top.emanjusaka; import reactor.core.publisher.Mono; public class Main { public static void main(String[] args) { Mono<String> mono = Mono.just("Hello"); mono.subscribe(System.out::println); } } // 输出 Hello`

> 使用Flux的操作符进行元素转换和过滤：
> 
>     package top.emanjusaka; import reactor.core.publisher.Flux; public class Main { public static void main(String[] args) { Flux<Integer> numbers = Flux.range(1, 10); numbers.map(num -> num * 2) .filter(num -> num % 3 == 0) .subscribe(System.out::println); } } // 输出 6 12 18

> 使用Mono的操作符进行元素转换和错误处理：
> 
> `package top.emanjusaka; import reactor.core.publisher.Mono; public class Main { public static void main(String[] args) { Mono<Integer> number = Mono.just(5); number.map(num -> num * 2) .doOnError(Throwable::printStackTrace) .subscribe(System.out::println); } } // 输出 10`

## 四、总结 ##

Flux 和 Mono 都是位于 reactor.core.publisher包下的类。

Reactor中的Flux和Mono是用于实现响应式编程的两种基本类型：

1.  Flux：表示一个异步序列，可以发出0到N个项目。它可以终止于完成信号或错误信号。Flux适用于处理多个项目的情况，可以使用各种操作符来处理和转换序列。
2.  Mono：表示一个异步序列，最多发出一个项目。它要么终止于完成信号（有或没有值），要么只发出一个错误信号。Mono适用于处理单个项目的情况，也可以使用一些操作符来处理和转换序列。

这两种类型都是Publisher的实现，遵循Reactive Streams规范，并可以与其他响应式库和框架进行互操作。

Flux和Mono都可以表示无限序列，也可以表示空序列。它们提供了丰富的操作符来处理和转换序列，例如映射、过滤、合并、扁平化等。此外，它们还支持异步和并发处理，可以与其他操作符和操作进行组合使用。

总的来说，Flux适用于处理多个项目的情况，而Mono适用于处理单个项目的情况。它们是Reactor中用于实现响应式编程的基本类型，提供了丰富的操作符和功能来处理和转换异步序列。

[Flux]: https://img-blog.csdnimg.cn/img_convert/b1e32c8ed223f5d80c5bb8669b43a34c.webp?x-oss-process=image/format,png
[Mono]: https://img-blog.csdnimg.cn/img_convert/026067e71eb1394120ae63ff5e00b4ee.webp?x-oss-process=image/format,png