Spring Boot 3.x 响应式编程（WebFlux）：Flux/Mono流处理与背压机制 - 智猿学院-前后端，数据库，人工智能，云计算等领域前沿技术讲座

好的，我们开始今天的讲座。

Spring Boot 3.x 响应式编程（WebFlux）：Flux/Mono流处理与背压机制

大家好，今天我们来深入探讨Spring Boot 3.x中的响应式编程，特别是WebFlux框架下Flux/Mono流处理以及背压机制。响应式编程是一种面向数据流和变化传播的声明式编程范式，它能够帮助我们构建更具弹性、响应更快、扩展性更强的应用程序。

一、响应式编程的基石：Reactive Streams规范

理解Flux和Mono之前，必须先了解Reactive Streams规范。Reactive Streams规范定义了一组用于异步处理元素序列的标准接口，旨在解决异步数据流处理中的背压问题。主要包含以下四个接口：

Publisher: 发布者，产生数据并发送给订阅者。
Subscriber: 订阅者，接收发布者发送的数据。
Subscription: 连接发布者和订阅者的桥梁，负责控制数据流的速率。订阅者通过Subscription请求数据，发布者根据请求发送数据。
Processor<T, R>: 既是Publisher又是Subscriber，可以对数据流进行转换和处理。

二、Flux和Mono：响应式数据类型的核心

Spring WebFlux基于Reactor库，提供了Flux和Mono两种响应式数据类型，它们是构建响应式应用程序的基础。

Flux: 表示一个包含0到N个元素的异步序列。可以理解为一个异步的数据流，它可以发出多个数据项，也可以在完成时发出完成信号，或者在发生错误时发出错误信号。
Mono: 表示一个包含0或1个元素的异步序列。相当于一个特殊的Flux，只能发出0个或者1个元素。适用于处理单个结果的场景，例如数据库查询、HTTP请求等。

2.1 创建Flux和Mono

Reactor 提供了多种方式来创建Flux和Mono。

创建Flux:

方法名	描述	示例
`just()`	创建一个包含指定元素的Flux。	`Flux.just("A", "B", "C");`
`fromIterable()`	从一个Iterable对象（如List、Set）创建一个Flux。	`Flux.fromIterable(Arrays.asList("A", "B", "C"));`
`fromArray()`	从一个数组创建一个Flux。	`Flux.fromArray(new String[]{"A", "B", "C"});`
`range()`	创建一个包含指定范围内整数的Flux。	`Flux.range(1, 5);` // 生成 1, 2, 3, 4, 5
`interval()`	创建一个以固定时间间隔发出Long型元素的Flux。	`Flux.interval(Duration.ofSeconds(1));` // 每隔1秒发出一个Long型元素 (0, 1, 2, …)
`empty()`	创建一个空的Flux。	`Flux.empty();`
`error()`	创建一个发出错误信号的Flux。	`Flux.error(new RuntimeException("Something went wrong!"));`
`generate()`	使用同步方式逐个生成元素的Flux。	`Flux.generate(sink -> { sink.next("Hello"); sink.complete(); });`
`create()`	使用异步方式生成元素的Flux，允许更灵活地控制数据流的产生和完成。	`Flux.create(sink -> { new Thread(() -> { sink.next("Async Hello"); sink.complete(); }).start(); });`

创建Mono:

方法名	描述	示例
`just()`	创建一个包含指定元素的Mono。	`Mono.just("A");`
`empty()`	创建一个空的Mono。	`Mono.empty();`
`error()`	创建一个发出错误信号的Mono。	`Mono.error(new RuntimeException("Something went wrong!"));`
`fromCallable()`	从一个Callable对象创建一个Mono，Callable在订阅时执行。	`Mono.fromCallable(() -> "Result");`
`fromFuture()`	从一个Future对象创建一个Mono。	`Mono.fromFuture(CompletableFuture.completedFuture("Result"));`
`justOrEmpty()`	如果传入的值为null，则创建Empty的Mono，否则创建包含该值的Mono。	`Mono.justOrEmpty(nullableValue);`
`delay()`	创建一个在指定延迟后发出元素的Mono。	`Mono.delay(Duration.ofSeconds(2)).thenReturn("Delayed Result");` // 2秒后发出 "Delayed Result"

示例代码：

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.util.Arrays;
import java.util.concurrent.CompletableFuture;

public class ReactiveExamples {

    public static void main(String[] args) {
        // 创建Flux
        Flux<String> flux1 = Flux.just("A", "B", "C");
        Flux<String> flux2 = Flux.fromIterable(Arrays.asList("D", "E", "F"));
        Flux<Integer> flux3 = Flux.range(1, 5);
        Flux<Long> flux4 = Flux.interval(Duration.ofSeconds(1)).take(3); //限制只取前三个

        // 创建Mono
        Mono<String> mono1 = Mono.just("G");
        Mono<String> mono2 = Mono.fromCallable(() -> "H");
        Mono<String> mono3 = Mono.fromFuture(CompletableFuture.completedFuture("I"));
        Mono<String> mono4 = Mono.delay(Duration.ofSeconds(2)).thenReturn("J");

        // 订阅并打印Flux元素
        System.out.println("Flux Examples:");
        flux1.subscribe(item -> System.out.println("flux1: " + item));
        flux2.subscribe(item -> System.out.println("flux2: " + item));
        flux3.subscribe(item -> System.out.println("flux3: " + item));
        flux4.subscribe(item -> System.out.println("flux4: " + item));

        // 订阅并打印Mono元素
        System.out.println("nMono Examples:");
        mono1.subscribe(item -> System.out.println("mono1: " + item));
        mono2.subscribe(item -> System.out.println("mono2: " + item));
        mono3.subscribe(item -> System.out.println("mono3: " + item));
        mono4.subscribe(item -> System.out.println("mono4: " + item));

        // 为了观察delay() 和 interval() 的效果，需要阻塞主线程一段时间
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

2.2 操作符：转换和组合数据流

Flux和Mono提供了丰富的操作符，用于对数据流进行转换、过滤、组合等操作。

常用的Flux操作符：

操作符	描述	示例
`map()`	将每个元素转换为另一个元素。	`Flux.just("A", "B").map(String::toLowerCase);` // 输出 a, b
`flatMap()`	将每个元素转换为一个Flux，并将所有Flux合并为一个Flux。	`Flux.just("A", "B").flatMap(s -> Flux.just(s.toLowerCase(), s.toUpperCase()));` // 输出 a, A, b, B
`filter()`	过滤元素，只保留满足条件的元素。	`Flux.range(1, 10).filter(i -> i % 2 == 0);` // 输出 2, 4, 6, 8, 10
`zip()`	将多个Flux的元素按照顺序两两合并。	`Flux.just("A", "B").zipWith(Flux.just("1", "2"));` // 输出 (A, 1), (B, 2)
`concatWith()`	将两个Flux顺序连接起来。	`Flux.just("A", "B").concatWith(Flux.just("C", "D"));` // 输出 A, B, C, D
`mergeWith()`	将两个Flux无序合并，哪个Flux先发出元素就先处理哪个。	`Flux.just("A", "B").mergeWith(Flux.just("C", "D"));` // 输出 A, B, C, D (顺序不确定)
`take()`	从Flux中获取指定数量的元素。	`Flux.range(1, 10).take(3);` // 输出 1, 2, 3
`skip()`	跳过Flux中指定数量的元素。	`Flux.range(1, 10).skip(3);` // 输出 4, 5, 6, 7, 8, 9, 10
`distinct()`	移除重复的元素。	`Flux.just("A", "B", "A", "C").distinct();` // 输出 A, B, C
`buffer()`	将元素收集到List中，达到指定数量或者时间间隔后，将List作为一个元素发出。	`Flux.range(1, 5).buffer(2);` // 输出 [1, 2], [3, 4], [5]
`window()`	将元素收集到Flux中，达到指定数量或者时间间隔后，将Flux作为一个元素发出（Flux<Flux>）。	`Flux.range(1, 5).window(2);` // 输出 Flux([1, 2]), Flux([3, 4]), Flux([5])
`onErrorReturn()`	当发生错误时，返回一个默认值。	`Flux.just(1, 2).concatWith(Flux.error(new RuntimeException())).onErrorReturn(0);` // 输出 1, 2, 0
`onErrorResume()`	当发生错误时，执行一个备用Flux。	`Flux.just(1, 2).concatWith(Flux.error(new RuntimeException())).onErrorResume(e -> Flux.just(3, 4));` // 输出 1, 2, 3, 4
`retry()`	当发生错误时，重试指定次数。	`Flux.just(1).concatWith(Flux.error(new RuntimeException())).retry(2);` // 最多重试2次

常用的Mono操作符：

操作符	描述	示例
`map()`	将元素转换为另一个元素。	`Mono.just("A").map(String::toLowerCase);` // 输出 a
`flatMap()`	将元素转换为一个Mono，并返回这个Mono。	`Mono.just("A").flatMap(s -> Mono.just(s.toLowerCase()));` // 输出 a
`filter()`	过滤元素，只保留满足条件的元素。	`Mono.just(1).filter(i -> i % 2 == 0);` // 输出 Empty Mono (因为1不满足条件)
`zipWith()`	将两个Mono的元素合并为一个Tuple。	`Mono.just("A").zipWith(Mono.just("1"));` // 输出 (A, 1)
`then()`	在Mono完成后执行另一个Mono，忽略前一个Mono的结果。	`Mono.just("A").then(Mono.just("B"));` // 输出 B
`thenReturn()`	在Mono完成后返回一个指定的值，忽略前一个Mono的结果。	`Mono.just("A").thenReturn("B");` // 输出 B
`onErrorReturn()`	当发生错误时，返回一个默认值。	`Mono.error(new RuntimeException()).onErrorReturn("Default");` // 输出 Default
`onErrorResume()`	当发生错误时，执行一个备用Mono。	`Mono.error(new RuntimeException()).onErrorResume(e -> Mono.just("Fallback"));` // 输出 Fallback
`retry()`	当发生错误时，重试指定次数。	`Mono.just(1).concatWith(Mono.error(new RuntimeException())).retry(2);` // 最多重试2次

示例代码：

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.Duration;

public class OperatorExamples {

    public static void main(String[] args) {
        // Flux 操作符示例
        Flux.range(1, 5)
                .map(i -> "Number " + i)
                .filter(s -> s.contains("2"))
                .subscribe(System.out::println);  // 输出 Number 2

        Flux.just("A", "B")
                .flatMap(s -> Flux.just(s.toLowerCase(), s.toUpperCase()))
                .subscribe(System.out::println); // 输出 a, A, b, B

        Flux.interval(Duration.ofMillis(200))
                .take(5)
                .subscribe(i -> System.out.println("Interval: " + i));

        // Mono 操作符示例
        Mono.just("Hello")
                .map(String::toUpperCase)
                .subscribe(System.out::println); // 输出 HELLO

        Mono.just(1)
                .filter(i -> i > 0)
                .subscribe(i -> System.out.println("Mono Filtered: " + i)); // 输出 Mono Filtered: 1

        Mono.delay(Duration.ofSeconds(1))
                .thenReturn("Delayed Value")
                .subscribe(value -> System.out.println("Mono Delayed: " + value));

        try {
            Thread.sleep(2000); // 为了观察 delay 的效果
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

三、背压机制：流量控制的关键

背压（Backpressure）是指在响应式流中，当订阅者的处理速度跟不上发布者的发布速度时，订阅者通知发布者降低发送速率的一种机制。避免订阅者被发布者发送的大量数据淹没，导致资源耗尽甚至崩溃。

3.1 背压策略

Reactor提供了多种背压策略，允许开发者根据实际情况选择合适的策略。

背压策略	描述
`BUFFER`	缓存所有数据，直到订阅者可以处理。如果发布者速度远大于订阅者，可能导致内存溢出。
`DROP`	当订阅者无法处理时，直接丢弃新发布的数据。会导致数据丢失，适用于可以容忍少量数据丢失的场景。
`LATEST`	只保留最新的数据，丢弃旧的数据。适用于只关心最新状态的场景。
`ERROR`	当订阅者无法处理时，发出一个错误信号。适用于需要保证数据完整性的场景，一旦出现背压情况，立即停止数据流。
`IGNORE`	忽略背压，不进行任何处理。不建议使用，容易导致资源耗尽。
`onBackpressureBuffer()`	使用更细粒度的缓存控制，可以设置缓存大小和溢出策略（如丢弃最旧的或最新的）。可以结合`BUFFER`、`DROP` 和 `LATEST` 的优点，提供更灵活的背压控制。
`onBackpressureDrop()`	丢弃无法处理的元素。可以自定义丢弃的逻辑。
`onBackpressureLatest()`	保留最新的元素。与`LATEST`类似，但可以应用于更复杂的数据流场景。

3.2 使用背压策略

可以使用onBackpressureXXX()操作符来应用背压策略。

示例代码：

import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;

import java.time.Duration;

public class BackpressureExamples {

    public static void main(String[] args) throws InterruptedException {

        //  BUFFER 策略
        Flux.interval(Duration.ofMillis(1))
                .onBackpressureBuffer() // 使用buffer策略
                .publishOn(Schedulers.boundedElastic())
                .subscribe(data -> {
                    try {
                        Thread.sleep(10); // 模拟慢速消费者
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println("Received: " + data);
                });

        // DROP 策略
        Flux.interval(Duration.ofMillis(1))
                .onBackpressureDrop(data -> System.out.println("Dropped: " + data)) // 使用drop策略
                .publishOn(Schedulers.boundedElastic())
                .subscribe(data -> {
                    try {
                        Thread.sleep(10); // 模拟慢速消费者
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println("Received: " + data);
                });

        // LATEST 策略
        Flux.interval(Duration.ofMillis(1))
                .onBackpressureLatest() // 使用latest策略
                .publishOn(Schedulers.boundedElastic())
                .subscribe(data -> {
                    try {
                        Thread.sleep(10); // 模拟慢速消费者
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println("Received: " + data);
                });

        Thread.sleep(5000); // 让程序运行一段时间
    }
}

在这个例子中，发布者以非常快的速度生成数据，而订阅者模拟一个慢速的消费者。通过应用不同的背压策略，可以观察到不同的行为。

四、WebFlux中的响应式编程

Spring WebFlux 是一个基于响应式编程模型的Web框架，它利用Reactor库提供的Flux和Mono来处理HTTP请求和响应。

4.1 响应式Controller

在WebFlux中，Controller可以返回Flux或Mono，WebFlux会自动将这些响应式数据类型转换为HTTP响应。

import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.util.Arrays;
import java.util.List;

@RestController
@RequestMapping("/reactive")
public class ReactiveController {

    @GetMapping("/hello")
    public Mono<String> hello() {
        return Mono.just("Hello Reactive World!");
    }

    @GetMapping("/numbers")
    public Flux<Integer> numbers() {
        return Flux.range(1, 10);
    }

    @GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<Long> stream() {
        return Flux.interval(Duration.ofSeconds(1));
    }

    @GetMapping("/user/{id}")
    public Mono<User> getUser(@PathVariable String id) {
        // 模拟从数据库获取用户
        User user = new User(id, "User " + id, "user" + id + "@example.com");
        return Mono.just(user);
    }

    @GetMapping("/users")
    public Flux<User> getUsers() {
        List<User> users = Arrays.asList(
                new User("1", "User 1", "[email protected]"),
                new User("2", "User 2", "[email protected]"),
                new User("3", "User 3", "[email protected]")
        );
        return Flux.fromIterable(users);
    }

    static class User {
        private String id;
        private String name;
        private String email;

        public User(String id, String name, String email) {
            this.id = id;
            this.name = name;
            this.email = email;
        }

        public String getId() {
            return id;
        }

        public String getName() {
            return name;
        }

        public String getEmail() {
            return email;
        }
    }
}

在这个例子中，/hello 返回一个包含字符串的Mono，/numbers 返回一个包含整数序列的Flux，/stream 返回一个无限的Long型数据流（Server-Sent Events）。 /user/{id}返回一个包含User对象的Mono, /users返回一个包含多个User对象的Flux。

4.2 响应式WebClient

WebClient 是 Spring WebFlux 提供的非阻塞式、响应式的 HTTP 客户端。可以用来调用其他的REST API.

import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;

public class WebClientExample {

    public static void main(String[] args) {
        WebClient client = WebClient.create("http://localhost:8080"); // 替换为你的服务器地址

        Mono<String> result = client.get()
                .uri("/reactive/hello")
                .retrieve()
                .bodyToMono(String.class);

        result.subscribe(System.out::println);

        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

这段代码使用WebClient调用了/reactive/hello接口，并将返回的结果转换为String类型的Mono。

4.3 响应式数据访问

Spring Data 提供了响应式的数据访问支持，可以使用响应式的Repository来操作数据库。例如，Spring Data MongoDB 提供了ReactiveMongoRepository，可以用来进行响应式的MongoDB操作。

五、实际应用场景

高并发API: 处理大量并发请求，例如实时数据推送、在线游戏等。
微服务架构: 构建弹性、可扩展的微服务系统。
事件驱动应用: 处理事件流，例如日志分析、金融数据处理等。
IO密集型应用: 优化IO操作，例如文件上传下载、网络请求等。

六、选择合适的背压策略和一些建议

在选择背压策略时，需要根据具体的应用场景和需求进行权衡。

如果可以容忍少量数据丢失，可以选择 DROP 或 LATEST 策略。
如果需要保证数据完整性，可以选择 ERROR 策略，或者使用 BUFFER 策略并设置合理的缓存大小。
使用 onBackpressureBuffer() 可以提供更细粒度的缓存控制。
监控数据流的速率和消费者的处理能力，根据实际情况调整背压策略。
合理使用线程池和调度器，避免阻塞操作。
充分利用响应式编程的优势，构建更具弹性和响应性的应用程序。

总结

响应式编程是一种强大的编程范式，可以帮助我们构建更具弹性和响应性的应用程序。理解Flux和Mono的特性，掌握常用的操作符，并合理应用背压机制，是构建成功的响应式应用的关键。通过Spring WebFlux，我们可以轻松地构建响应式的Web应用，提高系统的性能和可扩展性。

发表回复 取消回复

发表回复取消回复