JAVA 如何用 Reactor 实现非阻塞式响应式微服务？ - 智猿学院-前后端，数据库，人工智能，云计算等领域前沿技术讲座

Reactor 构建非阻塞响应式微服务：从理论到实践

大家好，今天我们来深入探讨如何使用 Project Reactor 构建非阻塞响应式微服务。Reactor 作为一个强大的响应式编程框架，可以帮助我们构建高性能、弹性的微服务，更好地应对高并发和复杂的业务场景。

1. 响应式编程与非阻塞 I/O 的必要性

在传统的阻塞式 I/O 模型中，每个请求都需要一个线程来处理。在高并发的情况下，大量的线程会导致 CPU 资源消耗过大，上下文切换频繁，最终导致系统性能下降。

响应式编程和非阻塞 I/O 的结合可以有效解决这个问题。

响应式编程 (Reactive Programming): 是一种基于异步数据流和变化传播的编程范式。它强调数据流的连续性和变化的处理，而不是传统的请求-响应模式。
非阻塞 I/O (Non-blocking I/O): 允许线程在等待 I/O 操作完成时，不被阻塞，而是可以继续处理其他任务。当 I/O 操作完成时，系统会通知线程。

Reactor 正是提供了这样的能力，它基于 Reactive Streams 规范，并提供了丰富的操作符，可以帮助我们轻松地构建响应式应用。

2. Reactor 核心概念

在使用 Reactor 之前，我们需要了解其核心概念：

Flux: 表示一个包含 0 到 N 个元素的异步序列。可以理解为一个“流”，它可以发出多个数据项。
Mono: 表示一个包含 0 或 1 个元素的异步序列。类似于 Flux，但只发出最多一个数据项。
Publisher: 是 Reactive Streams 规范中的一个接口，用于发布数据。Flux 和 Mono 都实现了 Publisher 接口。
Subscriber: 是 Reactive Streams 规范中的另一个接口，用于订阅 Publisher 并接收数据。
Subscription: 表示 Publisher 和 Subscriber 之间的连接。它允许 Subscriber 请求数据或取消订阅。
Scheduler: Reactor 使用 Scheduler 来控制任务的执行线程。不同的 Scheduler 适用于不同的场景，例如 Schedulers.immediate() (当前线程), Schedulers.single() (单线程), Schedulers.elastic() (弹性线程池), Schedulers.parallel() (并行线程池) 等。

3. Reactor 的依赖引入

首先，需要在项目中引入 Reactor 的依赖。如果使用 Maven，可以在 pom.xml 文件中添加以下依赖：

<dependency>
    <groupId>io.projectreactor</groupId>
    <artifactId>reactor-core</artifactId>
    <version>3.6.0</version> <!-- 使用最新版本 -->
</dependency>

如果使用 Gradle，可以在 build.gradle 文件中添加以下依赖：

implementation 'io.projectreactor:reactor-core:3.6.0' // 使用最新版本

4. 创建 Flux 和 Mono

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

Flux.just(): 从已有的数据创建 Flux。

Flux<String> flux = Flux.just("Hello", "Reactor", "World");
flux.subscribe(System.out::println); // 输出：Hello Reactor World

Mono.just(): 从已有的数据创建 Mono。

Mono<String> mono = Mono.just("Hello");
mono.subscribe(System.out::println); // 输出：Hello

Flux.fromIterable(): 从集合创建 Flux。

List<String> list = Arrays.asList("A", "B", "C");
Flux<String> flux = Flux.fromIterable(list);
flux.subscribe(System.out::println); // 输出：A B C

Flux.range(): 创建一个包含指定范围整数的 Flux。

Flux<Integer> flux = Flux.range(1, 5);
flux.subscribe(System.out::println); // 输出：1 2 3 4 5

Flux.empty(): 创建一个空的 Flux。

Flux<String> flux = Flux.empty();
flux.subscribe(System.out::println); // 不输出任何内容

Mono.empty(): 创建一个空的 Mono。

Mono<String> mono = Mono.empty();
mono.subscribe(System.out::println); // 不输出任何内容

Flux.error() 和 Mono.error(): 创建一个包含错误的 Flux 或 Mono。

Flux<String> flux = Flux.error(new RuntimeException("Something went wrong"));
flux.subscribe(
        System.out::println,
        Throwable::printStackTrace // 输出异常信息
);

Mono<String> mono = Mono.error(new IllegalArgumentException("Invalid argument"));
mono.subscribe(
        System.out::println,
        Throwable::printStackTrace
);

Flux.create() 和 Mono.create(): 允许手动创建 Flux 和 Mono，可以更灵活地控制数据的发布。

Flux<String> flux = Flux.create(emitter -> {
    emitter.next("Data 1");
    emitter.next("Data 2");
    emitter.complete(); // 表示数据流完成
});
flux.subscribe(System.out::println); // 输出：Data 1 Data 2

Mono<String> mono = Mono.create(sink -> {
    sink.success("Single Data"); // 发送数据并完成
});
mono.subscribe(System.out::println); // 输出：Single Data

5. Reactor 常用操作符

Reactor 提供了丰富的操作符，可以对 Flux 和 Mono 进行各种转换和处理：

操作符	描述	示例代码
`map()`	将 `Flux` 或 `Mono` 中的每个元素转换为另一个元素。	`Flux.range(1, 5).map(i -> "Number: " + i).subscribe(System.out::println); // 输出：Number: 1 Number: 2 Number: 3 Number: 4 Number: 5`

filter(): 过滤掉不符合条件的元素。

Flux<Integer> flux = Flux.range(1, 10).filter(i -> i % 2 == 0); // 只保留偶数
flux.subscribe(System.out::println); // 输出：2 4 6 8 10

flatMap(): 将每个元素转换为一个 Flux 或 Mono，然后将所有 Flux 或 Mono 合并为一个 Flux。适用于异步操作。

Flux<Integer> flux = Flux.just(1, 2, 3);
Flux<String> stringFlux = flux.flatMap(i -> Mono.just("Value: " + i).delayElement(Duration.ofMillis(100))); // 模拟异步操作
stringFlux.subscribe(System.out::println); // 输出：Value: 1 Value: 2 Value: 3 (延迟 100ms)

concatMap(): 类似于 flatMap()，但保持元素的顺序。

Flux<Integer> flux = Flux.just(1, 2, 3);
Flux<String> stringFlux = flux.concatMap(i -> Mono.just("Value: " + i).delayElement(Duration.ofMillis(100))); // 模拟异步操作
stringFlux.subscribe(System.out::println); // 输出：Value: 1 Value: 2 Value: 3 (延迟 100ms，保持顺序)

zip(): 将多个 Publisher 的元素组合成一个新元素。

Flux<String> flux1 = Flux.just("A", "B", "C");
Flux<Integer> flux2 = Flux.just(1, 2, 3);

Flux.zip(flux1, flux2, (s, i) -> s + i)
        .subscribe(System.out::println); // 输出：A1 B2 C3

reduce(): 将 Flux 中的所有元素聚合成一个单一的值。

Flux<Integer> flux = Flux.range(1, 5);
Mono<Integer> sum = flux.reduce(0, Integer::sum); // 计算总和
sum.subscribe(System.out::println); // 输出：15

onErrorResume(): 当发生错误时，使用备用 Publisher。

Flux<Integer> flux = Flux.just(1, 2, 0)
        .map(i -> 10 / i) // 可能抛出 ArithmeticException
        .onErrorResume(e -> Flux.just(-1)); // 发生错误时，返回 -1
flux.subscribe(System.out::println); // 输出：10 5 -1

onErrorReturn(): 当发生错误时，返回一个默认值。

Flux<Integer> flux = Flux.just(1, 2, 0)
        .map(i -> 10 / i) // 可能抛出 ArithmeticException
        .onErrorReturn(-1); // 发生错误时，返回 -1
flux.subscribe(System.out::println); // 输出：10 5 -1

retry(): 当发生错误时，重试操作。

Flux<Integer> flux = Flux.just(1, 2, 0)
        .map(i -> 10 / i) // 可能抛出 ArithmeticException
        .retry(2); // 重试 2 次
flux.subscribe(
        System.out::println,
        Throwable::printStackTrace // 输出异常信息
);

delayElements(): 延迟每个元素的发布。

Flux<String> flux = Flux.just("A", "B", "C").delayElements(Duration.ofMillis(500));
flux.subscribe(System.out::println); // A, B, C 依次输出，每次间隔 500ms

6. 使用 Reactor 实现非阻塞 HTTP 请求

Spring WebFlux 提供了基于 Reactor 的非阻塞 HTTP 客户端 WebClient，可以用来发起 HTTP 请求。

import org.springframework.web.reactive.function.client.WebClient;

public class WebClientExample {

    public static void main(String[] args) {
        WebClient client = WebClient.create("https://jsonplaceholder.typicode.com"); // 创建 WebClient 实例

        Mono<String> result = client.get()
                .uri("/todos/1") // 请求的 URI
                .retrieve()
                .bodyToMono(String.class); // 将响应体转换为 String 类型

        result.subscribe(
                System.out::println, // 成功时的处理
                Throwable::printStackTrace // 失败时的处理
        );

        // 阻塞主线程，等待请求完成 (仅用于演示，实际应用中避免阻塞)
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

在这个例子中，WebClient 发起一个 GET 请求到 https://jsonplaceholder.typicode.com/todos/1，并将响应体转换为 String 类型。bodyToMono(String.class) 方法返回一个 Mono<String>，表示异步的响应结果。

7. 使用 Reactor 实现响应式 REST API

Spring WebFlux 也提供了构建响应式 REST API 的能力。可以使用 @RestController 和 @RequestMapping 注解来定义 API 端点，并使用 Flux 和 Mono 作为返回值。

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.util.Arrays;
import java.util.List;

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

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

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

    @GetMapping("/users")
    public Flux<User> getAllUsers() {
        // 模拟从数据库获取用户列表
        List<User> users = Arrays.asList(
                new User(1L, "User 1"),
                new User(2L, "User 2"),
                new User(3L, "User 3")
        );
        return Flux.fromIterable(users);
    }

    // 内部类，用于表示用户信息
    static class User {
        private Long id;
        private String name;

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

        public Long getId() {
            return id;
        }

        public String getName() {
            return name;
        }
    }
}

在这个例子中，/api/hello 端点返回一个 Mono<String>，/api/users/{id} 端点返回一个 Mono<User>，/api/users 端点返回一个 Flux<User>。这些 API 都是非阻塞的，可以高效地

Reactor 构建非阻塞响应式微服务：从理论到实践

发表回复 取消回复

发表回复取消回复