Reactor 3指南

# 一、Reactor 3简介

Reactor 3 (opens new window) 是一个基于 Java 的响应式编程库，它是 Project Reactor 的核心实现，旨在为 Java 开发者提供一套高效、灵活的响应式编程工具。Reactor 3 的设计灵感来源于 Reactive Streams 规范，该规范定义了响应式编程的标准接口，使得不同的响应式库可以互相兼容。

Reactor 3 的主要特点包括：

Flux 和 Mono：Reactor 3 提供了两种核心数据类型：

Flux：表示一个包含 0 到 N 个元素的异步序列。
Mono：表示一个包含 0 或 1 个元素的异步序列。这两种类型为开发者提供了灵活的数据处理能力，适用于不同的场景。

丰富的操作符：Reactor 3 提供了大量的操作符（如 map、flatMap、filter 等），用于对数据流进行转换、过滤、合并等操作。这些操作符使得开发者可以轻松地构建复杂的数据处理管道。
背压支持：Reactor 3 完全支持 背压（Backpressure） 机制，能够有效地处理生产者和消费者之间的速率不匹配问题，避免系统资源的浪费。
与 Spring 生态的无缝集成：Reactor 3 是 Spring WebFlux 的底层实现，能够与 Spring 框架无缝集成，帮助开发者构建高性能的响应式 Web 应用程序。
高性能与低延迟：Reactor 3 在设计上注重性能优化，能够处理高并发场景下的数据流，同时保持低延迟。

# 二、安装与设置

# 1. 环境准备

# a. Java版本要求

Reactor 3 是基于 Java 的响应式编程库，因此需要确保开发环境中安装了合适的 Java 版本。Reactor 3 支持的最低 Java 版本为 Java 8，但建议使用 Java 11 或更高版本，以获得更好的性能和兼容性。

可以通过以下命令检查当前 Java 版本：

java -version

如果未安装 Java 或版本过低，可以从 Oracle JDK (opens new window) 或 OpenJDK (opens new window) 下载并安装合适的版本。

# b. 添加依赖项

Reactor 3 可以通过 Maven 或 Gradle 添加到项目中。以下是两种构建工具的依赖配置：

Maven 配置：

在 pom.xml 文件中添加以下依赖：

<dependencies>
    <!-- Reactor Core -->
    <dependency>
        <groupId>io.projectreactor</groupId>
        <artifactId>reactor-core</artifactId>
        <version>3.7.2</version>
    </dependency>
    <!-- Reactor Test (可选，用于测试) -->
    <dependency>
        <groupId>io.projectreactor</groupId>
        <artifactId>reactor-test</artifactId>
        <version>3.7.2</version>
        <scope>test</scope>
    </dependency>
</dependencies>

Gradle 配置：

在 build.gradle 文件中添加以下依赖：

dependencies {
    // Reactor Core
    implementation 'io.projectreactor:reactor-core:3.7.2'
    // Reactor Test (可选，用于测试)
    testImplementation 'io.projectreactor:reactor-test:3.7.2'
}

# 2. 第一个Reactor程序

# a. 创建简单的Flux和Mono序列

在 Reactor 3 中，Flux 和 Mono 是两种核心数据类型，分别用于处理多个元素和单个元素的异步数据流。以下是创建简单 Flux 和 Mono 序列的示例代码：

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

public class FirstReactorProgram {
    public static void main(String[] args) {
        // 创建一个包含多个元素的 Flux
        Flux<String> flux = Flux.just("Apple", "Banana", "Cherry");

        // 创建一个包含单个元素的 Mono
        Mono<String> mono = Mono.just("Hello Reactor");
    }
}

# b. 订阅和处理数据

创建 Flux 和 Mono 序列后，需要通过 subscribe 方法订阅数据流并处理数据。以下是订阅和处理数据的示例代码：

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

public class FirstReactorProgram {
    public static void main(String[] args) {
        // 创建一个包含多个元素的 Flux
        Flux<String> flux = Flux.just("Apple", "Banana", "Cherry");

        // 订阅并处理 Flux 中的数据
        flux.subscribe(
            fruit -> System.out.println("Received: " + fruit), // 处理每个元素
            error -> System.err.println("Error: " + error),    // 处理错误
            () -> System.out.println("Flux completed")         // 处理完成事件
        );

        // 创建一个包含单个元素的 Mono
        Mono<String> mono = Mono.just("Hello Reactor");

        // 订阅并处理 Mono 中的数据
        mono.subscribe(
            message -> System.out.println("Received: " + message), // 处理元素
            error -> System.err.println("Error: " + error),        // 处理错误
            () -> System.out.println("Mono completed")             // 处理完成事件
        );
    }
}

代码说明：

Flux 示例：

使用 Flux.just 创建一个包含三个元素的 Flux 序列。
通过 subscribe 方法订阅数据流，并分别处理元素、错误和完成事件。

Mono 示例：

使用 Mono.just 创建一个包含单个元素的 Mono 序列。
通过 subscribe 方法订阅数据流，并分别处理元素、错误和完成事件。

运行结果：

Received: Apple
Received: Banana
Received: Cherry
Flux completed
Received: Hello Reactor
Mono completed

通过以上步骤，您已经成功创建了第一个 Reactor 程序，并学会了如何订阅和处理数据流。接下来，可以进一步探索 Reactor 3 的核心概念和高级特性。

# 三、核心概念

# 1. Flux 和 Mono

在 Reactor 3 中，Flux 和 Mono 是两种核心数据类型，分别用于处理多个元素和单个元素的异步数据流。它们是响应式编程的基础，理解它们的特性和用法是掌握 Reactor 3 的关键。

# a. Flux 类型详解

Flux 是一个表示 0 到 N 个元素 的异步序列。它可以发出多个元素，并在序列结束时发出完成信号，或者在发生错误时发出错误信号。Flux 适用于处理多个数据项的场景，例如从数据库读取多条记录、处理事件流等。

Flux 的创建方式：

静态工厂方法：

Flux.just(T... data)：创建一个包含指定元素的 Flux。
Flux.fromIterable(Iterable<T> iterable)：从集合或迭代器创建 Flux。
Flux.range(int start, int count)：创建一个包含连续整数的 Flux。
Flux.interval(Duration period)：创建一个按固定时间间隔发出元素的 Flux。

动态生成：

使用 Flux.generate 或 Flux.create 动态生成数据流。

示例代码：

import reactor.core.publisher.Flux;

public class FluxExample {
    public static void main(String[] args) {
        // 创建一个包含多个元素的 Flux
        Flux<String> flux = Flux.just("Apple", "Banana", "Cherry");

        // 订阅并处理 Flux 中的数据
        flux.subscribe(
            fruit -> System.out.println("Received: " + fruit), // 处理每个元素
            error -> System.err.println("Error: " + error),    // 处理错误
            () -> System.out.println("Flux completed")         // 处理完成事件
        );

        // 创建一个按时间间隔发出元素的 Flux
        Flux<Long> intervalFlux = Flux.interval(java.time.Duration.ofSeconds(1))
                                      .take(5); // 只取前 5 个元素

        intervalFlux.subscribe(
            value -> System.out.println("Interval: " + value),
            error -> System.err.println("Error: " + error),
            () -> System.out.println("Interval Flux completed")
        );
    }
}

运行结果：

Received: Apple
Received: Banana
Received: Cherry
Flux completed
Interval: 0
Interval: 1
Interval: 2
Interval: 3
Interval: 4
Interval Flux completed

# b. Mono 类型详解

Mono 是一个表示 0 或 1 个元素 的异步序列。它通常用于处理单个数据项的场景，例如从数据库读取一条记录、处理单个 HTTP 请求等。

Mono 的创建方式：

静态工厂方法：

Mono.just(T data)：创建一个包含指定元素的 Mono。
Mono.empty()：创建一个不包含任何元素的 Mono。
Mono.error(Throwable error)：创建一个包含错误信号的 Mono。
Mono.fromCallable(Callable<T> callable)：从 Callable 创建 Mono。

动态生成：

使用 Mono.create 动态生成数据流。

示例代码：

import reactor.core.publisher.Mono;

public class MonoExample {
    public static void main(String[] args) {
        // 创建一个包含单个元素的 Mono
        Mono<String> mono = Mono.just("Hello Reactor");

        // 订阅并处理 Mono 中的数据
        mono.subscribe(
            message -> System.out.println("Received: " + message), // 处理元素
            error -> System.err.println("Error: " + error),        // 处理错误
            () -> System.out.println("Mono completed")             // 处理完成事件
        );

        // 创建一个空的 Mono
        Mono<String> emptyMono = Mono.empty();

        emptyMono.subscribe(
            message -> System.out.println("Received: " + message),
            error -> System.err.println("Error: " + error),
            () -> System.out.println("Empty Mono completed")
        );
    }
}

运行结果：

Received: Hello Reactor
Mono completed
Empty Mono completed

# 2. 操作符

Reactor 3 提供了丰富的操作符，用于对数据流进行转换、过滤、合并等操作。这些操作符是响应式编程的核心工具，能够帮助开发者轻松地构建复杂的数据处理管道。以下是三类常用的操作符：数据转换操作符、错误处理操作符以及合并与拆分操作符。

# a. 数据转换操作符

数据转换操作符用于对数据流中的元素进行转换或映射。以下是一些常用的数据转换操作符：

map：将元素转换为另一种类型。
flatMap：将元素转换为另一个 Flux 或 Mono，并将结果展平。
filter：根据条件过滤元素。
concatMap：类似于 flatMap，但保持顺序。
switchMap：将元素转换为另一个 Flux 或 Mono，并取消之前的订阅。

示例代码：

import reactor.core.publisher.Flux;

public class TransformationOperators {
    public static void main(String[] args) {
        // 使用 map 转换元素
        Flux<Integer> numbers = Flux.just(1, 2, 3, 4, 5);
        numbers.map(n -> n * 2)
               .subscribe(System.out::println); // 输出: 2, 4, 6, 8, 10

        // 使用 flatMap 转换并展平
        Flux<String> words = Flux.just("Hello", "World");
        words.flatMap(word -> Flux.fromArray(word.split("")))
             .subscribe(System.out::println); // 输出: H, e, l, l, o, W, o, r, l, d

        // 使用 filter 过滤元素
        Flux<Integer> evenNumbers = numbers.filter(n -> n % 2 == 0);
        evenNumbers.subscribe(System.out::println); // 输出: 2, 4
    }
}

# b. 错误处理操作符

错误处理操作符用于处理数据流中的错误信号。以下是一些常用的错误处理操作符：

onErrorReturn：在发生错误时返回一个默认值。
onErrorResume：在发生错误时切换到另一个 Flux 或 Mono。
onErrorMap：将错误转换为另一种类型的错误。
retry：在发生错误时重试订阅。

示例代码：

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

public class ErrorHandlingOperators {
    public static void main(String[] args) {
        // 使用 onErrorReturn 处理错误
        Flux<Integer> fluxWithError = Flux.just(1, 2, 3)
                                          .concatWith(Flux.error(new RuntimeException("Error occurred")));
        fluxWithError.onErrorReturn(-1)
                     .subscribe(System.out::println); // 输出: 1, 2, 3, -1

        // 使用 onErrorResume 切换到另一个 Flux
        fluxWithError.onErrorResume(e -> Flux.just(4, 5, 6))
                     .subscribe(System.out::println); // 输出: 1, 2, 3, 4, 5, 6

        // 使用 retry 重试订阅
        fluxWithError.retry(1)
                     .subscribe(System.out::println, 
                                error -> System.err.println("Error: " + error)); // 输出: 1, 2, 3, 1, 2, 3, Error: Error occurred
    }
}

# c. 合并与拆分操作符

合并与拆分操作符用于将多个数据流合并为一个，或将一个数据流拆分为多个。以下是一些常用的合并与拆分操作符：

merge：将多个 Flux 合并为一个。
concat：按顺序连接多个 Flux。
zip：将多个 Flux 的元素按索引组合。
groupBy：根据条件将 Flux 拆分为多个组。
window：将 Flux 拆分为多个窗口。

示例代码：

import reactor.core.publisher.Flux;

public class CombiningAndSplittingOperators {
    public static void main(String[] args) {
        // 使用 merge 合并多个 Flux
        Flux<Integer> flux1 = Flux.just(1, 2, 3);
        Flux<Integer> flux2 = Flux.just(4, 5, 6);
        Flux.merge(flux1, flux2)
            .subscribe(System.out::println); // 输出: 1, 2, 3, 4, 5, 6

        // 使用 concat 按顺序连接多个 Flux
        Flux.concat(flux1, flux2)
            .subscribe(System.out::println); // 输出: 1, 2, 3, 4, 5, 6

        // 使用 zip 组合多个 Flux
        Flux<String> names = Flux.just("Alice", "Bob", "Charlie");
        Flux<Integer> ages = Flux.just(25, 30, 35);
        Flux.zip(names, ages, (name, age) -> name + " is " + age + " years old")
            .subscribe(System.out::println); // 输出: Alice is 25 years old, Bob is 30 years old, Charlie is 35 years old

        // 使用 groupBy 分组
        Flux<Integer> numbers = Flux.just(1, 2, 3, 4, 5);
        numbers.groupBy(n -> n % 2 == 0 ? "Even" : "Odd")
               .subscribe(group -> group.collectList()
                                       .subscribe(list -> System.out.println(group.key() + ": " + list)));
        // 输出: Odd: [1, 3, 5], Even: [2, 4]
    }
}

# 3. 背压（Backpressure）

背压（Backpressure）是响应式编程中的一个重要概念，用于解决生产者和消费者之间速率不匹配的问题。当生产者生成数据的速度快于消费者处理数据的速度时，背压机制可以防止数据积压，避免系统资源耗尽。

# a. 背压原理

背压的核心思想是让消费者能够控制生产者的数据生成速率，从而避免数据积压。在 Reactor 3 中，背压是通过 响应式流规范（Reactive Streams Specification） 实现的，该规范定义了以下四个核心接口：

Publisher：数据生产者，负责生成数据。
Subscriber：数据消费者，负责处理数据。
Subscription：连接生产者和消费者的桥梁，用于控制数据流。
Processor：同时扮演生产者和消费者的角色。

背压的工作流程：

消费者通过 Subscription 向生产者请求一定数量的数据。
生产者根据消费者的请求生成数据，并将数据发送给消费者。
消费者处理完数据后，可以继续请求更多数据。

背压的优势：

避免数据积压，防止内存溢出。
提高系统的稳定性和可伸缩性。
支持动态调整数据生成速率。

# b. 实现背压策略

Reactor 3 提供了多种背压策略，开发者可以根据具体需求选择合适的策略。以下是几种常见的背压策略及其实现方式：

onBackpressureBuffer：

将未处理的数据缓存在内存中，直到消费者准备好处理。
适用于消费者处理速度较慢，但数据量不大的场景。

示例代码：

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

public class BackpressureBufferExample {
    public static void main(String[] args) {
        Flux.range(1, 1000)
            .onBackpressureBuffer(100) // 设置缓冲区大小为 100
            .publishOn(Schedulers.parallel())
            .subscribe(
                data -> {
                    try {
                        Thread.sleep(10); // 模拟慢速消费者
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println("Received: " + data);
                },
                error -> System.err.println("Error: " + error),
                () -> System.out.println("Completed")
            );
    }
}

onBackpressureDrop：

当消费者无法处理数据时，直接丢弃多余的数据。
适用于可以容忍数据丢失的场景。

示例代码：

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

public class BackpressureDropExample {
    public static void main(String[] args) {
        Flux.range(1, 1000)
            .onBackpressureDrop(dropped -> System.out.println("Dropped: " + dropped))
            .publishOn(Schedulers.parallel())
            .subscribe(
                data -> {
                    try {
                        Thread.sleep(10); // 模拟慢速消费者
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println("Received: " + data);
                },
                error -> System.err.println("Error: " + error),
                () -> System.out.println("Completed")
            );
    }
}

onBackpressureLatest：

当消费者无法处理数据时，保留最新的数据，丢弃旧数据。
适用于需要实时数据的场景。

示例代码：

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

public class BackpressureLatestExample {
    public static void main(String[] args) {
        Flux.range(1, 1000)
            .onBackpressureLatest()
            .publishOn(Schedulers.parallel())
            .subscribe(
                data -> {
                    try {
                        Thread.sleep(10); // 模拟慢速消费者
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println("Received: " + data);
                },
                error -> System.err.println("Error: " + error),
                () -> System.out.println("Completed")
            );
    }
}

limitRate：

限制生产者的数据生成速率，使其与消费者的处理速率匹配。
适用于需要动态调整速率的场景。

示例代码：

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

public class LimitRateExample {
    public static void main(String[] args) {
        Flux.range(1, 1000)
            .limitRate(10) // 每批处理 10 个元素
            .publishOn(Schedulers.parallel())
            .subscribe(
                data -> {
                    try {
                        Thread.sleep(10); // 模拟慢速消费者
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println("Received: " + data);
                },
                error -> System.err.println("Error: " + error),
                () -> System.out.println("Completed")
            );
    }
}

# 4. Sinks

在响应式编程中，Sinks 是 Reactor 提供的一个强大工具，用于手动控制数据流的生成和发布。它允许开发者以编程方式创建和管理数据流，适用于需要动态生成数据的场景。

# a. Sinks 的作用

手动控制数据流：通过 Sinks，开发者可以手动触发数据的发布、完成或错误处理。
支持多种数据流类型：Sinks 支持 Mono 和 Flux 两种数据流类型。
线程安全：Sinks 提供了线程安全的 API，适用于多线程环境。

# b. Sinks 的类型

Reactor 提供了多种 Sinks 类型，适用于不同的场景：

Sinks.One：

用于生成 Mono 数据流。
只能发布一个数据项或一个错误。
适用于单次数据发布的场景。

Sinks.Many：

用于生成 Flux 数据流。
可以发布多个数据项、完成信号或错误。
支持多种发布模式（如 unicast、multicast、replay）。

# c. Sinks 的使用示例

# 示例 1：使用 `Sinks.One` 创建 Mono

import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;

public class SinksOneExample {
    public static void main(String[] args) {
        // 创建 Sinks.One
        Sinks.One<String> sink = Sinks.one();

        // 获取 Mono
        Mono<String> mono = sink.asMono();

        // 订阅 Mono
        mono.subscribe(System.out::println);

        // 发布数据
        sink.tryEmitValue("Hello, Sinks.One!");
    }
}

输出：

Hello, Sinks.One!

# 示例 2：使用 `Sinks.Many` 创建 Flux

import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;

public class SinksManyExample {
    public static void main(String[] args) {
        // 创建 Sinks.Many（使用 multicast 模式）
        Sinks.Many<String> sink = Sinks.many().multicast().onBackpressureBuffer();

        // 获取 Flux
        Flux<String> flux = sink.asFlux();

        // 订阅 Flux
        flux.subscribe(System.out::println);

        // 发布多个数据项
        sink.tryEmitNext("Data 1");
        sink.tryEmitNext("Data 2");
        sink.tryEmitNext("Data 3");

        // 完成数据流
        sink.tryEmitComplete();
    }
}

输出：

Data 1
Data 2
Data 3

# 示例 3：处理错误

import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;

public class SinksErrorExample {
    public static void main(String[] args) {
        // 创建 Sinks.Many
        Sinks.Many<String> sink = Sinks.many().multicast().onBackpressureBuffer();

        // 获取 Flux
        Flux<String> flux = sink.asFlux();

        // 订阅 Flux 并处理错误
        flux.subscribe(
            System.out::println,
            error -> System.err.println("Error: " + error.getMessage())
        );

        // 发布数据
        sink.tryEmitNext("Data 1");

        // 发布错误
        sink.tryEmitError(new RuntimeException("Something went wrong!"));
    }
}

输出：

Data 1
Error: Something went wrong!

# d. Sinks 的发布模式

Sinks.Many 支持多种发布模式，适用于不同的场景。以下是每种发布模式的详细说明及代码示例：

# 1. Unicast 模式

特点：
- 只允许一个订阅者。
- 如果尝试添加第二个订阅者，会抛出 IllegalStateException。
- 适用于点对点的数据流。
示例：

import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;

public class SinksUnicastExample {
    public static void main(String[] args) {
        // 创建 Sinks.Many（使用 unicast 模式）
        Sinks.Many<String> sink = Sinks.many().unicast().onBackpressureBuffer();

        // 获取 Flux
        Flux<String> flux = sink.asFlux();

        // 订阅 Flux
        flux.subscribe(data -> System.out.println("Subscriber 1: " + data));

        // 发布数据
        sink.tryEmitNext("Data 1");
        sink.tryEmitNext("Data 2");

        // 尝试添加第二个订阅者（会抛出异常）
        try {
            flux.subscribe(data -> System.out.println("Subscriber 2: " + data));
        } catch (IllegalStateException e) {
            System.err.println("Error: " + e.getMessage());
        }

        // 完成数据流
        sink.tryEmitComplete();
    }
}

输出：

Subscriber 1: Data 1
Subscriber 1: Data 2
Error: UnicastProcessor allows only a single Subscriber

# 2. Multicast 模式

特点：
- 允许多个订阅者。
- 新订阅者只能接收到订阅之后发布的数据。
- 适用于广播场景。
示例：

import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;

public class SinksMulticastExample {
    public static void main(String[] args) {
        // 创建 Sinks.Many（使用 multicast 模式）
        Sinks.Many<String> sink = Sinks.many().multicast().onBackpressureBuffer();

        // 获取 Flux
        Flux<String> flux = sink.asFlux();

        // 订阅 Flux
        flux.subscribe(data -> System.out.println("Subscriber 1: " + data));

        // 发布数据
        sink.tryEmitNext("Data 1");
        sink.tryEmitNext("Data 2");

        // 添加第二个订阅者
        flux.subscribe(data -> System.out.println("Subscriber 2: " + data));

        // 发布更多数据
        sink.tryEmitNext("Data 3");

        // 完成数据流
        sink.tryEmitComplete();
    }
}

输出：

Subscriber 1: Data 1
Subscriber 1: Data 2
Subscriber 1: Data 3
Subscriber 2: Data 3

# 3. Replay 模式

特点：
- 缓存历史数据，新订阅者可以接收到之前发布的数据。
- 支持配置缓存大小或时间窗口。
- 适用于需要重放数据的场景。
示例：

import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;

public class SinksReplayExample {
    public static void main(String[] args) {
        // 创建 Sinks.Many（使用 replay 模式，缓存最近 2 个数据项）
        Sinks.Many<String> sink = Sinks.many().replay().limit(2);

        // 获取 Flux
        Flux<String> flux = sink.asFlux();

        // 发布数据
        sink.tryEmitNext("Data 1");
        sink.tryEmitNext("Data 2");
        sink.tryEmitNext("Data 3");

        // 订阅 Flux
        flux.subscribe(data -> System.out.println("Subscriber 1: " + data));

        // 添加第二个订阅者
        flux.subscribe(data -> System.out.println("Subscriber 2: " + data));

        // 完成数据流
        sink.tryEmitComplete();
    }
}

输出：

Subscriber 1: Data 2
Subscriber 1: Data 3
Subscriber 2: Data 2
Subscriber 2: Data 3

# 四、高级特性

在响应式编程中，并发与线程管理是至关重要的高级特性。Reactor 3 提供了强大的工具来管理线程和并发执行，确保数据流的高效处理。以下是关于并发与线程管理的两个核心主题：Scheduler 调度器和多线程执行。

# 1. 并发与线程管理

# a. Scheduler 调度器

Scheduler 是 Reactor 3 中用于控制任务执行线程的核心组件。它允许开发者将任务分配到不同的线程池中，从而实现并发执行。Reactor 提供了多种内置的 Scheduler，适用于不同的场景：

Schedulers.immediate()：

在当前线程中立即执行任务。
适用于不需要切换线程的场景。

Schedulers.single()：

使用单个线程执行所有任务。
适用于需要顺序执行的场景。

Schedulers.parallel()：

使用固定大小的线程池执行任务。
适用于 CPU 密集型任务。

Schedulers.elastic()：

使用可扩展的线程池执行任务。
适用于 I/O 密集型任务。

Schedulers.boundedElastic()：

使用有限制的弹性线程池执行任务。
适用于需要控制资源使用的 I/O 密集型任务。

示例代码：

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

public class SchedulerExample {
    public static void main(String[] args) {
        // 使用 parallel Scheduler 执行任务
        Flux.range(1, 10)
            .parallel()
            .runOn(Schedulers.parallel())
            .subscribe(i -> {
                System.out.println("Processing " + i + " on thread: " + Thread.currentThread().getName());
            });

        // 使用 elastic Scheduler 执行任务
        Flux.range(1, 10)
            .publishOn(Schedulers.elastic())
            .subscribe(i -> {
                System.out.println("Processing " + i + " on thread: " + Thread.currentThread().getName());
            });

        // 使用 boundedElastic Scheduler 执行任务
        Flux.range(1, 10)
            .publishOn(Schedulers.boundedElastic())
            .subscribe(i -> {
                System.out.println("Processing " + i + " on thread: " + Thread.currentThread().getName());
            });
    }
}

# b. 多线程执行

在响应式编程中，多线程执行可以通过以下方式实现：

publishOn：

将后续操作切换到指定的 Scheduler 线程中执行。
适用于需要将部分操作分配到不同线程的场景。

subscribeOn：

将整个订阅过程（包括数据生成和处理）切换到指定的 Scheduler 线程中执行。
适用于需要将整个数据流分配到不同线程的场景。

parallel：

将数据流并行化处理，使用多个线程同时处理数据。
适用于需要并行处理的场景。

示例代码：

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

public class MultiThreadExecutionExample {
    public static void main(String[] args) throws InterruptedException {
        // 使用 publishOn 切换线程
        Flux.range(1, 5)
            .map(i -> {
                System.out.println("Map 1 on thread: " + Thread.currentThread().getName());
                return i * 2;
            })
            .publishOn(Schedulers.parallel())
            .map(i -> {
                System.out.println("Map 2 on thread: " + Thread.currentThread().getName());
                return i + 1;
            })
            .subscribe(i -> System.out.println("Received: " + i + " on thread: " + Thread.currentThread().getName()));

        // 使用 subscribeOn 切换线程
        Flux.range(1, 5)
            .subscribeOn(Schedulers.elastic())
            .map(i -> {
                System.out.println("Map on thread: " + Thread.currentThread().getName());
                return i * 2;
            })
            .subscribe(i -> System.out.println("Received: " + i + " on thread: " + Thread.currentThread().getName()));

        // 使用 parallel 并行处理
        Flux.range(1, 10)
            .parallel()
            .runOn(Schedulers.parallel())
            .map(i -> {
                System.out.println("Processing " + i + " on thread: " + Thread.currentThread().getName());
                return i * 2;
            })
            .sequential()
            .subscribe(i -> System.out.println("Received: " + i + " on thread: " + Thread.currentThread().getName()));

        // 等待异步任务完成
        Thread.sleep(1000);
    }
}

# 2. 测试

在响应式编程中，测试是一个重要的环节。Reactor 提供了强大的测试工具，如 StepVerifier 和 TestPublisher，帮助开发者验证数据流的正确性和行为。以下是关于测试的两个核心主题：使用 StepVerifier 进行测试和 TestPublisher 的使用。

# a. 使用 StepVerifier 进行测试

StepVerifier 是 Reactor 提供的测试工具，用于验证数据流的行为。它可以模拟订阅过程，并逐步验证数据流中的每个元素、错误和完成信号。

StepVerifier 的主要功能：

验证数据流中的元素是否符合预期。
验证数据流是否按预期完成或抛出错误。
支持时间相关的测试（如 virtualTime）。

示例代码：

import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;

public class StepVerifierExample {
    public static void main(String[] args) {
        // 创建一个简单的 Flux
        Flux<Integer> flux = Flux.range(1, 5);

        // 使用 StepVerifier 测试 Flux
        StepVerifier.create(flux)
            .expectNext(1) // 验证第一个元素
            .expectNext(2) // 验证第二个元素
            .expectNext(3) // 验证第三个元素
            .expectNext(4) // 验证第四个元素
            .expectNext(5) // 验证第五个元素
            .expectComplete() // 验证数据流完成
            .verify(); // 启动测试

        // 测试包含错误的 Flux
        Flux<Integer> errorFlux = Flux.range(1, 5)
            .map(i -> {
                if (i == 3) throw new RuntimeException("Error at 3");
                return i;
            });

        StepVerifier.create(errorFlux)
            .expectNext(1) // 验证第一个元素
            .expectNext(2) // 验证第二个元素
            .expectError(RuntimeException.class) // 验证抛出异常
            .verify(); // 启动测试
    }
}

虚拟时间测试： 对于涉及时间操作的数据流（如 delayElements），可以使用 StepVerifier.withVirtualTime 来模拟时间流逝，避免实际等待。

示例代码：

import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;
import java.time.Duration;

public class VirtualTimeExample {
    public static void main(String[] args) {
        // 创建一个延迟的 Flux
        Flux<Integer> delayedFlux = Flux.range(1, 5)
            .delayElements(Duration.ofSeconds(1));

        // 使用虚拟时间测试
        StepVerifier.withVirtualTime(() -> delayedFlux)
            .expectSubscription() // 验证订阅
            .thenAwait(Duration.ofSeconds(5)) // 模拟等待 5 秒
            .expectNext(1, 2, 3, 4, 5) // 验证所有元素
            .expectComplete() // 验证完成
            .verify(); // 启动测试
    }
}

# b. TestPublisher 的使用

TestPublisher 是一个用于测试的工具类，允许开发者手动控制数据流的发布行为。它可以模拟各种场景，如正常数据发布、错误抛出和完成信号。

TestPublisher 的主要功能：

手动发布数据。
模拟错误和完成信号。
支持背压测试。

示例代码：

import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import reactor.test.publisher.TestPublisher;
import reactor.test.StepVerifier;

public class TestPublisherExample {
    @Test
    public void testTestPublisher() {
        // 创建一个 TestPublisher
        TestPublisher<Integer> testPublisher = TestPublisher.create();

        // 创建一个 Flux 并订阅
        Flux<Integer> flux = testPublisher.flux();
        StepVerifier.create(flux)
            .then(() -> testPublisher.next(1, 2, 3)) // 手动发布数据
            .expectNext(1, 2, 3) // 验证数据
            .then(() -> testPublisher.error(new RuntimeException("Test Error"))) // 手动抛出错误
            .expectError(RuntimeException.class) // 验证错误
            .verify(); // 启动测试
    }
}

背压测试： TestPublisher 还可以用于测试背压行为，验证消费者是否能够正确处理数据流。

示例代码：

import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import reactor.test.publisher.TestPublisher;
import reactor.test.StepVerifier;

public class BackpressureTestExample {
    @Test
    public void testBackpressure() {
        // 创建一个 TestPublisher
        TestPublisher<Integer> testPublisher = TestPublisher.create();

        // 创建一个 Flux 并订阅
        Flux<Integer> flux = testPublisher.flux();
        StepVerifier.create(flux, 1) // 设置初始请求量为 1
            .then(() -> testPublisher.next(1)) // 发布第一个元素
            .expectNext(1) // 验证第一个元素
            .thenRequest(2) // 请求两个元素
            .then(() -> testPublisher.next(2, 3)) // 发布两个元素
            .expectNext(2, 3) // 验证两个元素
            .thenCancel() // 取消订阅
            .verify(); // 启动测试
    }
}

# 3. 集成 Spring WebFlux

Spring WebFlux 是 Spring 5 引入的响应式 Web 框架，基于 Reactor 实现，支持非阻塞、异步的编程模型。通过集成 Spring WebFlux，开发者可以构建高性能的响应式 Web 应用程序。以下是关于集成 Spring WebFlux 的两个核心主题：构建响应式 Web 应用程序和 Spring Cloud Gateway 集成案例。

# a. 构建响应式 Web 应用程序

Spring WebFlux 的核心特性：

基于 Reactor 的响应式编程模型。
支持非阻塞 I/O，适用于高并发场景。
提供注解驱动的编程模型（如 @RestController）和函数式编程模型。

构建响应式 Web 应用程序的步骤：

添加 Spring WebFlux 依赖。
定义响应式控制器。
配置路由和处理器。
启动应用程序。

示例代码：

添加依赖（Maven）：

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-webflux</artifactId>
    </dependency>
</dependencies>

定义响应式控制器：

import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;

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

    @GetMapping("/numbers")
    public Flux<Integer> getNumbers() {
        return Flux.range(1, 10); // 返回 1 到 10 的数字流
    }
}

启动应用程序：

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class ReactiveWebApplication {
    public static void main(String[] args) {
        SpringApplication.run(ReactiveWebApplication.class, args);
    }
}

测试 API：

启动应用程序后，访问 http://localhost:8080/api/numbers，将返回 [1,2,3,4,5,6,7,8,9,10]。

# b. Spring Cloud Gateway 集成案例

Spring Cloud Gateway 是 Spring Cloud 提供的 API 网关，基于 Spring WebFlux 实现，支持响应式编程模型。它可以用于路由、负载均衡、限流、熔断等场景。

Spring Cloud Gateway 的核心特性：

基于路由的请求转发。
支持过滤器（Filter）机制。
集成 Hystrix 实现熔断和降级。

集成 Spring Cloud Gateway 的步骤：

添加 Spring Cloud Gateway 依赖。
配置路由规则。
启动网关服务。

示例代码：

添加依赖（Maven）：

<dependencies>
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-gateway</artifactId>
    </dependency>
</dependencies>

配置路由规则：

# application.yml
spring:
  cloud:
    gateway:
      routes:
        - id: service_route
          uri: http://localhost:8080 # 目标服务地址
          predicates:
            - Path=/api/** # 匹配路径
          filters:
            - AddRequestHeader=X-Request-Foo, Bar # 添加请求头

启动网关服务：

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class GatewayApplication {
    public static void main(String[] args) {
        SpringApplication.run(GatewayApplication.class, args);
    }
}

测试网关：

启动网关服务后，访问 http://localhost:8081/api/numbers（假设网关运行在 8081 端口），网关会将请求转发到 http://localhost:8080/api/numbers。

自定义过滤器： 可以通过实现 GatewayFilter 或 GlobalFilter 接口来自定义过滤器。

示例代码：

import org.springframework.cloud.gateway.filter.GatewayFilter;
import org.springframework.cloud.gateway.filter.factory.AbstractGatewayFilterFactory;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Mono;

@Component
public class CustomFilter extends AbstractGatewayFilterFactory<CustomFilter.Config> {

    public CustomFilter() {
        super(Config.class);
    }

    @Override
    public GatewayFilter apply(Config config) {
        return (exchange, chain) -> {
            System.out.println("Custom Filter: Request path - " + exchange.getRequest().getPath());
            return chain.filter(exchange).then(Mono.fromRunnable(() -> {
                System.out.println("Custom Filter: Response status - " + exchange.getResponse().getStatusCode());
            }));
        };
    }

    public static class Config {
        // 配置参数
    }
}

# 五、最佳实践

在响应式编程和 Spring WebFlux 的使用过程中，遵循最佳实践可以显著提升应用程序的性能和稳定性。以下是关于 性能优化建议 和 常见问题及解决方案 的详细内容。

# 1. 性能优化建议

响应式编程的性能优化主要集中在以下几个方面：

# a. 合理使用线程池

问题：响应式编程依赖于非阻塞 I/O，但某些阻塞操作（如数据库访问）可能会占用线程资源。
建议：
- 使用 Schedulers 提供的线程池（如 Schedulers.parallel()）来隔离阻塞操作。
- 避免在主线程中执行阻塞操作。

示例代码：

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

public class ThreadPoolExample {
    public static void main(String[] args) {
        Mono.fromCallable(() -> {
                // 模拟阻塞操作
                Thread.sleep(1000);
                return "Blocking Operation Completed";
            })
            .subscribeOn(Schedulers.boundedElastic()) // 使用弹性线程池
            .subscribe(System.out::println);

        // 防止主线程退出
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

# b. 减少不必要的订阅

问题：频繁订阅和取消订阅会增加开销。
建议：
- 尽量复用 Flux 或 Mono 实例。
- 使用 cache() 操作符缓存数据流。

示例代码：

import reactor.core.publisher.Flux;

public class CacheExample {
    public static void main(String[] args) {
        Flux<Integer> cachedFlux = Flux.range(1, 5)
            .cache(); // 缓存数据流

        cachedFlux.subscribe(System.out::println); // 第一次订阅
        cachedFlux.subscribe(System.out::println); // 第二次订阅，直接使用缓存
    }
}

# c. 优化背压处理

问题：数据生产速度过快可能导致消费者无法处理。
建议：
- 使用 onBackpressureBuffer、onBackpressureDrop 或 onBackpressureLatest 操作符处理背压。
- 根据业务需求选择合适的背压策略。

示例代码：

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

public class BackpressureExample {
    public static void main(String[] args) {
        Flux.range(1, 100)
            .onBackpressureBuffer(10) // 缓冲 10 个元素
            .subscribe(System.out::println);
    }
}

# d. 使用虚拟时间测试

问题：时间相关的测试（如 delayElements）可能导致测试时间过长。
建议：
- 使用 StepVerifier.withVirtualTime 模拟时间流逝。

示例代码：

import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;
import java.time.Duration;

public class VirtualTimeExample {
    public static void main(String[] args) {
        Flux<Integer> delayedFlux = Flux.range(1, 5)
            .delayElements(Duration.ofSeconds(1));

        StepVerifier.withVirtualTime(() -> delayedFlux)
            .expectSubscription()
            .thenAwait(Duration.ofSeconds(5)) // 模拟等待 5 秒
            .expectNext(1, 2, 3, 4, 5)
            .expectComplete()
            .verify();
    }
}

# 2. 常见问题及解决方案

# a. 数据流未触发

问题：数据流未按预期触发。
解决方案：
- 检查是否遗漏了 subscribe() 方法。
- 确保数据源正确初始化。

示例代码：

import reactor.core.publisher.Flux;

public class SubscriptionExample {
    public static void main(String[] args) {
        Flux<Integer> flux = Flux.range(1, 5);
        flux.subscribe(System.out::println); // 必须调用 subscribe
    }
}

# b. 内存泄漏

问题：未取消订阅导致内存泄漏。
解决方案：
- 使用 Disposable 管理订阅，并在不再需要时调用 dispose() 方法。

示例代码：

import reactor.core.publisher.Flux;
import reactor.core.Disposable;

public class DisposableExample {
    public static void main(String[] args) {
        Flux<Integer> flux = Flux.range(1, 100);
        Disposable disposable = flux.subscribe(System.out::println);

        // 取消订阅
        disposable.dispose();
    }
}

# c. 阻塞操作导致性能下降

问题：在响应式流中执行阻塞操作（如数据库访问）导致性能下降。
解决方案：
- 将阻塞操作放到单独的线程池中执行。

示例代码：

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

public class BlockingOperationExample {
    public static void main(String[] args) {
        Mono.fromCallable(() -> {
                // 模拟阻塞操作
                Thread.sleep(1000);
                return "Blocking Operation Completed";
            })
            .subscribeOn(Schedulers.boundedElastic()) // 使用弹性线程池
            .subscribe(System.out::println);

        // 防止主线程退出
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

# d. 背压处理不当

问题：数据生产速度过快，消费者无法处理。
解决方案：
- 使用背压操作符（如 onBackpressureBuffer）或调整生产者的速率。

示例代码：

import reactor.core.publisher.Flux;

public class BackpressureSolutionExample {
    public static void main(String[] args) {
        Flux.range(1, 100)
            .onBackpressureBuffer(10) // 缓冲 10 个元素
            .subscribe(System.out::println);
    }
}

# 六、总结

响应式编程和 Spring WebFlux 为构建高性能、高可用的现代应用程序提供了强大的支持。通过本文的学习，您已经掌握了响应式编程的核心概念、Spring WebFlux 的使用方法以及性能优化的最佳实践。希望这些知识能够帮助您在未来的项目中更好地应用响应式编程，构建更加高效和稳定的系统。

祝你变得更强!

编辑

#Reactor

上次更新: 2025/01/23

← Spring体系介绍 Spring容器初始化过程和Bean生命周期探究→

Reactor 3指南

# 一、Reactor 3简介

# 二、安装与设置

# 1. 环境准备

# a. Java版本要求

# b. 添加依赖项

# 2. 第一个Reactor程序

# a. 创建简单的Flux和Mono序列

# b. 订阅和处理数据

# 三、核心概念

# 1. Flux 和 Mono

# a. Flux 类型详解

# b. Mono 类型详解

# 2. 操作符

# a. 数据转换操作符

# b. 错误处理操作符

# c. 合并与拆分操作符

# 3. 背压（Backpressure）

# a. 背压原理

# b. 实现背压策略

# 4. Sinks

# a. Sinks 的作用

# b. Sinks 的类型

# c. Sinks 的使用示例

# 示例 1：使用 Sinks.One 创建 Mono

# 示例 2：使用 Sinks.Many 创建 Flux

# 示例 3：处理错误

# d. Sinks 的发布模式

# 1. Unicast 模式

# 2. Multicast 模式

# 3. Replay 模式

# 四、高级特性

# 1. 并发与线程管理

# a. Scheduler 调度器

# b. 多线程执行

# 2. 测试

# a. 使用 StepVerifier 进行测试

# b. TestPublisher 的使用

# 3. 集成 Spring WebFlux

# a. 构建响应式 Web 应用程序

# b. Spring Cloud Gateway 集成案例

# 五、最佳实践

# 1. 性能优化建议

# a. 合理使用线程池

# b. 减少不必要的订阅

# c. 优化背压处理

# d. 使用虚拟时间测试

# 2. 常见问题及解决方案

# a. 数据流未触发

# b. 内存泄漏

# c. 阻塞操作导致性能下降

# d. 背压处理不当

# 六、总结

# 示例 1：使用 `Sinks.One` 创建 Mono

# 示例 2：使用 `Sinks.Many` 创建 Flux