响应式编程框架Reactor【9】

一、Reactor核心目标

在高并发、低延迟的现代系统中，阻塞式编程（如 synchronized, Thread.sleep()）会导致线程资源耗尽，系统吞吐量急剧下降。

Reactor 基于 Reactive Streams 规范，提供了一套非阻塞、异步、支持背压的响应式编程模型，是构建高性能、高并发系统的利器。

✅ 核心目标：用更少的线程处理更多的请求。

二、Reactor 核心知识点

2.1 响应式编程范式

响应式编程是一种基于数据流和变化传播的编程范式。一切皆为流（Stream），你可以对流进行声明式操作（map, filter, flatMap 等）。

// 声明式：我关心“做什么”，而不是“怎么做”
Flux.just(1, 2, 3).map(x -> x * 2).filter(x -> x > 3).subscribe(System.out::println);
// 输出：4, 6

2.2 核心类型：Mono与Flux

类型	元素数量	用途
Mono<T>	0 或 1 个	单个结果（如 HTTP 请求、数据库查询）
Flux<T>	0 到 N 个	多个结果（如列表、事件流）

// ✅ Mono 示例
Mono<String> user = Mono.just("Alice");
Mono<Void> save = userRepository.save(userEntity); // 无返回值// ✅ Flux 示例
Flux<String> users = Flux.just("Alice", "Bob", "Charlie");
Flux<Integer> numbers = Flux.range(1, 5); // 1,2,3,4,5

2.3 操作符

Reactor 提供了丰富的操作符，所有操作符都是冷流（Cold Stream），即订阅时才执行。

import reactor.core.publisher.Flux;
import java.time.Duration;public class OperatorsExample {public static void main(String[] args) throws InterruptedException {Flux.range(1, 10).filter(n -> n % 2 == 0)                    // 过滤偶数.map(n -> "Item-" + n)                      // 转换为字符串.delayElements(Duration.ofMillis(100))      // 每 100ms 发一个.doOnNext(System.out::println)              // 副作用：打印.take(3)                                    // 只取前 3 个.subscribe();Thread.sleep(500); // 等待输出}
}
// 输出：Item-2, Item-4, Item-6

🔍 delayElements 证明了非阻塞特性：不会阻塞主线程

2.4 背压（Backpressure）

问题：生产者太快，消费者太慢，导致内存溢出。

解决方案：背压 —— 消费者主动控制请求的数据量。

Flux.range(1, 1000).onBackpressureBuffer(100) // 缓冲区最多 100 个.onBackpressureDrop(item -> System.out.println("丢弃: " + item)) // 超出则丢弃.subscribe(new BaseSubscriber<Integer>() {@Overrideprotected void hookOnSubscribe(Subscription subscription) {request(10); // 初始请求 10 个}@Overrideprotected void hookOnNext(Integer value) {System.out.println("处理: " + value);request(1); // 处理完一个，再要一个}});

2.5 调度与线程模型（核心！）

subscribeOn vs publishOn

Flux.just("A", "B").map(data -> {System.out.println("上游线程: " + Thread.currentThread().getName());return data + "-1";}).subscribeOn(Schedulers.boundedElastic()) // 影响上游执行线程.map(data -> {System.out.println("下游线程: " + Thread.currentThread().getName());return data + "-2";}).publishOn(Schedulers.parallel())         // 切换下游执行线程.subscribe(result -> System.out.println("订阅线程: " + Thread.currentThread().getName() + " => " + result));

🎯 关键区别：

• subscribeOn：影响整个上游（从源头到当前位置）
• publishOn：只影响其后的下游操作

2.6 错误处理

Flux.just(1, 2, 3).map(n -> {if (n == 2) throw new RuntimeException("出错了");return "Result-" + n;}).onErrorResume(e -> {System.err.println("捕获错误: " + e.getMessage());return Flux.just("Fallback-1", "Fallback-2");}).retry(2) // 重试 2 次.subscribe(System.out::println);

三、底层实现原理

3.1 响应式流协议（Reactive Streams）

Reactor 实现了 Publisher, Subscriber, Subscription, Processor 四大接口。

public interface Publisher<T> {void subscribe(Subscriber<? super T> s);
}public interface Subscriber<T> {void onSubscribe(Subscription s);void onNext(T t);void onError(Throwable t);void onComplete();
}

3.2 操作符的链式结构

🔍 订阅时，链表逆向建立订阅关系，从 subscribe 回溯到源头。

3.3 异步与线程切换原理

• Schedulers 封装了 ExecutorService。
• publishOn 内部使用 Queue + Worker 实现线程切换。
• subscribeOn 在源头就切换执行线程。

四、典型应用场景

4.1 WebFlux 微服务（Spring 生态）

@RestController
public class UserController {@GetMapping("/users/{id}")public Mono<User> getUser(@PathVariable String id) {return userService.findById(id); // 非阻塞返回}@GetMapping("/users")public Flux<User> getAllUsers() {return userService.findAll(); // 流式返回}
}

✅ 优势：高并发下内存占用低，吞吐量高。

4.2 异步数据库访问（R2DBC）

@Repository
public class UserRepository {@Autowiredprivate DatabaseClient client;public Mono<User> findById(String id) {return client.sql("SELECT * FROM users WHERE id = $1").bind(0, id).map(row -> User.from(row)).one();}
}

✅ 替代 JDBC，实现真正的非阻塞数据库访问。

4.3 实时事件推送（WebSocket / SSE）

@GetMapping(value = "/events", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<String> streamEvents() {return eventBus.getEventFlux() // 事件流.map(event -> "Event: " + event.getData());
}

4.4 微服务间通信

WebClient client = WebClient.create("https://api.example.com");Mono<User> user = client.get().uri("/users/123").retrieve().bodyToMono(User.class); // 非阻塞调用

✅ 替代 RestTemplate，异步非阻塞。

四、实践经验与最佳实战

✅ 正确实践

• I/O 操作 → Schedulers.boundedElastic()

• CPU 计算 → Schedulers.parallel()

• 避免在 map 中阻塞

• 使用 StepVerifier 测试响应式流

❌ 常见陷阱

优点	缺点
✅ 高并发、低延迟	❌ 学习曲线陡峭
✅ 资源利用率高（线程少）	❌ 调试困难（异步栈）
✅ 支持背压，防止 OOM	❌ 与阻塞库集成复杂
✅ 与 Spring 深度集成	❌ 团队技能要求高

五、总结

• Reactor 是 Java 响应式编程的事实标准，其核心价值在于：

• 非阻塞异步：提升系统吞吐量

  背压机制：保障系统稳定性

  声明式编程：代码更简洁、可读

• 与 Spring 生态无缝集成

🚀 适用场景：高并发 Web 服务、实时系统、微服务、事件驱动架构。

⚠️ 不适用场景：简单 CRUD、低并发、团队无响应式经验。

掌握 Reactor，意味着你掌握了构建现代高性能系统的“核武器”。但切记：不要为了响应式而响应式，选择合适的工具解决合适的问题