关于后端:响应式编程初识-Flux-和-Mono

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

前言

Reactor 是一个响应式编程的根底类库，其中有两个很要害的类：Flux 和 Mono。把握这两个类和相干概念有助于咱们学习响应式编程。

Flux 和 Mono 都是数据流的发布者，应用 Flux 和 Mono 都能够收回三种数据信号：元素值，谬误信号，实现信号；谬误信号和实现信号都代表终止信号，终止信号用于通知订阅者数据流完结了，谬误信号终止数据流同时把错误信息传递给订阅者。

一、Flux

具备 rx 运算符的响应式流公布器，收回 0 到 N 个元素，而后实现（胜利或有谬误）。

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

​​

Flux<T> 是一个规范的 Publisher<T>，示意一个异步的 0 到 N 个收回的我的项目序列，可抉择终止于实现信号或谬误信号。依据 Reactive Streams 标准，这三种类型的信号转换为对上游 Subscriber 的 onNext、onComplete 和 onError 办法的调用。

因为可能呈现的信号范畴很大，Flux 是通用的响应式类型。请留神，所有事件，包含终止事件，都是可选的：没有 onNext 事件但有 onComplete 事件示意一个空的无限序列，但如果去掉 onComplete，则失去一个有限的空序列（除了用于勾销测试之外，不是特地有用）。同样，有限序列不肯定为空。例如，Flux.interval(Duration) 会生成一个有限的 Flux<Long>，从时钟收回定期的滴答声。Flux<T> 是规范的 Publisher<T>，它示意 0 到 N 个收回项的异步序列，能够抉择由实现信号或谬误终止。与 Reactive Streams 标准中一样，这三种类型的信号转换为对上游订阅者的 onNext、onComplete 和 onError 办法的调用。

凭借如此大范畴的可能信号，Flux 是通用的无功类型。请留神，所有事件，甚至终止事件，都是可选的：没有 onNext 事件，但 onComplete 事件示意一个空的无限序列，但删除 onComplete 并且您有一个有限的空序列（不是特地有用，除了围绕勾销的测试）。同样，有限序列不肯定是空的。例如，Flux.interval(Duration) 生成有限的 Flux<Long> 并从时钟收回规定的滴答声。

二、Mono

具备根本 rx 运算符的 Reactive Streams Publisher 通过 onNext 信号最多收回一项，而后以 onComplete 信号终止（胜利的 Mono，有或没有值），或者仅收回单个 onError 信号（失败的 Mono）。

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

​​

Mono<T> 是一种非凡的 Publisher<T>，通过 onNext 信号收回最多一个我的项目，而后通过 onComplete 信号终止（胜利的 Mono，有或没有值），或者只收回一个 onError 信号（失败的 Mono）。

大多数 Mono 实现在调用 onNext 后立刻调用其 Subscriber 的 onComplete。Mono.never() 是一个例外：它不收回任何信号，在技术上并不禁止，但在测试之外没有太大用处。另一方面，明确禁止应用 onNext 和 onError 的组合。

Mono 只提供了 Flux 可用的操作符的子集，而某些操作符（特地是将 Mono 与另一个 Publisher 组合的操作符）会切换到 Flux。例如，MonoconcatWith(Publisher) 返回一个 Flux，而 Mono​​then(Mono) 返回另一个 Mono。

请留神，您能够应用 Mono 来示意只有实现概念的无值异步过程（相似于 Runnable）。要创立一个，您能够应用一个空的 Mono<Void>。

三、代码示例

1. 创立一个 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
   !

2. 创立一个 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

3. 应用 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

4. 应用 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 中用于实现响应式编程的根本类型，提供了丰盛的操作符和性能来解决和转换异步序列。

五、参考文献

1. 《Reactor》参考文档

本文原创，满腹经纶，如有纰漏，欢送斧正。尊贵的敌人，如果本文对您有所帮忙，欢送点赞，并期待您的反馈，以便于一直优化。

原文地址：https://www.emanjusaka.top/archives/4/br
微信公众号：emanjusaka 的编程栈