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

2023-09-07 09:14 由 emanjusaka 发表于 #后端开发

by emanjusaka from  https://www.emanjusaka.top/archives/4 彼岸花开可奈何
本文欢迎分享与聚合，全文转载请留下原文地址。

前言

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

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

一、Flux

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

下图显示了 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是一种特殊的Publisher，通过onNext信号发出最多一个项目，然后通过onComplete信号终止（成功的Mono，有或没有值），或者只发出一个onError信号（失败的Mono）。
大多数Mono实现在调用onNext后立即调用其Subscriber的onComplete。Mono.never()是一个例外：它不发出任何信号，在技术上并不禁止，但在测试之外没有太大用处。另一方面，明确禁止使用onNext和onError的组合。
Mono只提供了Flux可用的操作符的子集，而某些操作符（特别是将Mono与另一个Publisher组合的操作符）会切换到Flux。例如，Mono#concatWith(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是用于实现响应式编程的两种基本类型：

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

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

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

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

五、参考文献

《Reactor》参考文档

本文原创，才疏学浅，如有纰漏，欢迎指正。尊贵的朋友，如果本文对您有所帮助，欢迎点赞，并期待您的反馈，以便于不断优化。

原文地址： https://www.emanjusaka.top/archives/4

微信公众号：emanjusaka的编程栈