<返回更多

新一代WebFlux框架核心技术Reactor响应式编程基本用法

2023-11-27  微信公众号  Spring全家桶实战案例源码
加入收藏

环境:projectreactor2020.0.14

 

1. 前言

在响应式编程中,Project Reactor提供了两个核心的概念:Mono和Flux。Mono和Flux都是Reactor中的Publisher,它们可以产生并发布数据,然后可以被订阅和消费。这两个概念在WebFlux中有着广泛的应用,帮助我们实现异步和非阻塞的编程模型。

在这个主题中,我们将深入探讨Mono和Flux的基本使用。我们将了解它们如何被创建,如何订阅它们的事件,以及如何处理错误和完成通知。通过学习这些内容,你将能够更好地理解WebFlux的响应式编程模型,并能够在你的项目中有效地使用Mono和Flux。

让我们开始吧!

2. 环境依赖

<dependency>
  <groupId>io.projectreactor</groupId>
  <artifactId>reactor-core</artifactId>
</dependency>
<dependencyManagement>
  <dependencies>
    <dependency>
      <groupId>io.projectreactor</groupId>
      <artifactId>reactor-bom</artifactId>
      <version>${reactor.version}</version>
      <type>pom</type>
      <scope>import</scope>
    </dependency>
  </dependencies>
</dependencyManagement>

3. Mono & Flux介绍

Flux

Flux表示了0到N个元素序列,下图展示了Flux如何转换元素

新一代WebFlux框架核心技术Reactor响应式编程基本用法Flux

一个Flux<T>是一个标准的Publisher<T>,它表示一个由0到N个发射项目组成的异步序列,可选地由一个完成信号或一个错误终止。在响应式流规范中,这三种类型的信号转换为对下游订阅者的onNext、onComplete和onError方法的调用。

由于可能信号的范围很大,Flux是通用的反应式类型。请注意,所有事件,甚至是终止事件,都是可选的:只有onComplete事件才能表示一个空的有限序列,但删除onComplete事件就会得到一个无限的空序列(没什么用处,除了关于取消的测试)。类似地,无限序列不一定是空的。例如,Flux.interval(Duration)产生一个无限长的Flux<Long>,并从时钟发出规则的时标。

Mono

Mono表示了0个或1个元素序列,下图展示了Mono如何转换元素

新一代WebFlux框架核心技术Reactor响应式编程基本用法

Mono

 

Mono<T>是一个专门的发布者<T>,它通过onNext信号发出最多一个项目,然后以onComplete信号终止(Mono成功,有或没有值),或只发出一个onError信号(Mono失败)。

大多数Mono实现都希望在调用onNext之后立即对其订阅者调用onComplete。Mono.never()是一个异常值:它不会发出任何信号,这在技术上并没有被禁止,但在测试之外并不是特别有用。另一方面,onNext和onError的组合是明确禁止的。

Mono只提供了可用于` Flux `的操作符子集,有些操作符(特别是那些将Mono与另一个`Publisher`结合的操作符)会切换到`Flux`。例如,Mono#concatWith(Publisher)返回一个Flux,而Mono#then(Mono)返回另一个Mono。

注意,你可以使用Mono来表示只有完成概念的无值异步进程(类似于Runnable)。要创建一个,可以使用一个空的Mono<Void>。

4. Mono & Flux常用操作

Mono常用操作

Mono.just(T value)方法:创建一个包含指定值的Mono对象。

Mono.just(10).subscribe(System.out::println) ;

Mono.empty()方法:创建一个空的Mono对象,即不包含任何元素。

Mono.justOrEmpty(T value)方法:如果指定值不为null,则创建一个包含该值的Mono对象;否则创建一个空的Mono对象。

// 输出10
Mono.justOrEmpty(10).subscribe(System.out::println) ;
// 如果值为null,没有任何输出
Mono.justOrEmpty(null).subscribe(System.out::println) ;

新一代WebFlux框架核心技术Reactor响应式编程基本用法图片

Mono.fromCallable(Callable<? extends T> supplier)方法:创建一个Mono对象,该对象包含通过调用给定Callable对象的call()方法得到的返回值。

// 通过Callable方式,我们可以在内部执行其它一些动作
Mono.fromCallable(() -> 666).subscribe(System.out::println) ;

新一代WebFlux框架核心技术Reactor响应式编程基本用法

Mono.fromSupplier(Supplier<? extends T> supplier)方法:创建一个Mono对象,该对象包含通过调用给定Supplier对象的get()方法得到的返回值。

Mono.fromSupplier(() -> 666).subscribe(System.out::println) ;

新一代WebFlux框架核心技术Reactor响应式编程基本用法图片

Mono.fromFuture(CompletableFuture<? extends Integer> future)方法:创建一个Mono对象,该对象包含通过调用给定CompletableFuture对象

Mono.fromFuture(CompletableFuture.supplyAsync(() -> 666)).subscribe(System.out::println) ;

新一代WebFlux框架核心技术Reactor响应式编程基本用法

下面这个示例完整的展示了当发生异常后的处理

public static Mono<Users> invoke(Mono<Users> user) {
  return user.flatMap(u -> {
    if ("admin".equals(u.getName())) {
      return Mono.error(new RuntimeException("越权")) ;
    }
    u.setName(u.getName() + " - ");
    return Mono.just(u) ;
  });
}


public static void mAIn(String[] args) {
  invoke(Mono.just(new Users("admin")))
    .doOnNext(System.out::println)
    .doOnError(e -> {
      System.out.println(e.getMessage()) ;
    })
    // .onErrorResume(e -> Mono.just(new Users(e.getMessage() + " - fallback"))) // 功能更强,可以对捕获的异常进行响应的处理,然后再返回一个值
    .onErrorReturn(new Users("return")) // 捕获异常,简单粗暴直接返回一个静态值
    .doOnNext(System.out::println)
    .subscribe(); 


}
越权
Users [name=return]

新一代WebFlux框架核心技术Reactor响应式编程基本用法

新一代WebFlux框架核心技术Reactor响应式编程基本用法

将该Mono的发射与提供的发布者连接(不交错)。

Mono.just(10).concatWith(Mono.just(20)).subscribe(System.out::println) ;

新一代WebFlux框架核心技术Reactor响应式编程基本用法图片

该操作符是在当前Mono执行完成后切换到另外一个Mono。

Mono.just(10).doOnNext(System.out::println)
  .then(Mono.just(666)) // 切换到另外一个Mono通道, 忽略之前的Mono元素
  .doOnNext(System.out::println)
  .subscribe();

新一代WebFlux框架核心技术Reactor响应式编程基本用法

Flux常用操作

just():直接使用元素创建Flux,即在创建Flux时拿到数据,之后有谁订阅它,就重新发送数据给订阅者。

Flux.just(1, 2, 3...)

新一代WebFlux框架核心技术Reactor响应式编程基本用法图片

fromArray()、fromIterable()和fromStream():可以从数组、Iterable对象或Stream对象中创建Flux对象。

Flux.fromArray(new String[]{"1","2","3"});
Flux.fromIterable(List.of("a","b","c"));
Flux.fromStream(List.of("a","b","c").stream());

fromArray

新一代WebFlux框架核心技术Reactor响应式编程基本用法

fromIterable

新一代WebFlux框架核心技术Reactor响应式编程基本用法

fromStream

新一代WebFlux框架核心技术Reactor响应式编程基本用法

empty():创建一个不包含任何元素,只发布结束消息的序列。

新一代WebFlux框架核心技术Reactor响应式编程基本用法

range(int start, int count):创建包含从start起始的count个数量的Integer对象的序列。

Flux.range(1, 10) ;

新一代WebFlux框架核心技术Reactor响应式编程基本用法图片

Flux.error(new RuntimeException("错误")).onErrorResume(ex -> Mono.just("发生异常:" + ex.getMessage())).subscribe(System.out::println) ;

error操作符

新一代WebFlux框架核心技术Reactor响应式编程基本用法

onErrorResume操作符

新一代WebFlux框架核心技术Reactor响应式编程基本用法

Flux.just(1, 3, 6).flatMap(id -> {
  Mono<Integer> query = Mono.fromSupplier(() -> {
    System.out.println("查询数据...") ;
    return id * 10 ;
  }).delayElement(Duration.ofSeconds(2)) ;
  Mono<String> save = Mono.fromSupplier(() -> {
    System.out.println("保存数据...") ;
    return "success - " + id ;
  }) ;
  return Mono.when(query, save) ;
}).doOnComplete(() -> {
  System.out.println("执行完成...") ;
}).subscribe() ;

新一代WebFlux框架核心技术Reactor响应式编程基本用法图片

Flux.just(1,2,3,4,5,6).concatMap(item -> Mono.just(item).filterWhen(r -> {
  return Mono.just(r % 2 == 0) ;
})).subscribe(System.out::println) ;

新一代WebFlux框架核心技术Reactor响应式编程基本用法

总之,Reactor中的Flux和Mono是响应式编程的核心组件,它们提供了丰富的操作符和方法来处理异步数据流。因此,对于使用WebFlux的开发者来说,掌握Reactor的使用是非常重要的。

完毕!!!

关键词:框架      点击(12)
声明:本站部分内容来自互联网,如有版权侵犯或其他问题请与我们联系,我们将立即删除或处理。
▍相关推荐
更多框架相关>>>