doOnNext() vs flatMap()：区别与适用场景
在 Reactor（Flux / Mono）中，doOnNext() 和 flatMap() 都可以用来处理流中的元素，但它们有不同的作用和适用场景。
1. doOnNext()
✅ 作用
- 用于执行副作用（side effects），但不会改变数据流。
- 适用于 日志记录、计数、打印信息、同步操作 等。
📌 特点
- 不会改变数据流，只是对流中的元素 执行额外操作（如日志、统计等）。
- 只适用于 同步操作，不会等待异步任务完成。
- 返回 原始的
Flux或Mono，数据不会被修改。
🚀 示例
Flux<String> flux = Flux.just("A", "B", "C")
.doOnNext(item -> System.out.println("Processing: " + item)) // 仅执行副作用
.map(String::toLowerCase); // 原始流不变
flux.subscribe(System.out::println);
📝 输出
Processing: A
a
Processing: B
b
Processing: C
c
✅ doOnNext() 只是执行副作用（打印日志），但不会修改流中的数据。
2. flatMap()
✅ 作用
- 用于异步转换，可将一个元素映射为一个新的
Mono或Flux（异步执行）。 - 适用于 调用异步服务、数据库查询、Web 请求等。
📌 特点
- 返回
Publisher（Mono或Flux），会订阅并合并新的异步流。 - 适用于 异步操作，如数据库查询、Web API 调用等。
- 可以改变数据流（
A → Mono<B>）。
🚀 示例
Flux<String> flux = Flux.just("A", "B", "C")
.flatMap(item -> Mono.fromCallable(() -> {
System.out.println("Processing: " + item);
return item.toLowerCase(); // 返回 Mono<String>
}));
flux.subscribe(System.out::println);
📝 输出
Processing: A
a
Processing: B
b
Processing: C
c
✅ flatMap() 会执行异步任务，并等待 Mono<String> 结果再继续。
3. doOnNext() vs flatMap() 的核心区别
| 对比点 | doOnNext() | flatMap() |
|---|---|---|
| 作用 | 执行副作用（side effect），但不会改变数据流 | 转换数据流，可返回新的 Mono 或 Flux |
| 适用于 | 记录日志、计数、打印信息、同步操作 | 异步操作（数据库查询、Web 请求等） |
| 是否会改变流 | ❌ 不改变流中的数据 | ✅ 可以映射到新的数据流 |
| 是否支持异步 | ❌ 只适用于同步操作 | ✅ 适用于异步任务 |
是否订阅新的 Mono 或 Flux | ❌ 不会订阅新的 Publisher | ✅ 会订阅新的 Publisher |
4. 什么时候用 doOnNext()，什么时候用 flatMap()？
✅ doOnNext() 适用于：
- 只是想 观察数据流，但不改变数据：
Flux.just("A", "B", "C") .doOnNext(item -> log.info("Received: {}", item)) .subscribe(); - 记录日志、打印信息、计数等 副作用操作。
✅ flatMap() 适用于：
- 异步任务（数据库查询、HTTP 请求等）：
Flux.just("user1", "user2") .flatMap(userId -> getUserInfo(userId)) // 假设 getUserInfo 返回 Mono<User> .subscribe(user -> log.info("User: {}", user)); - 需要等待异步任务的结果并修改数据流。
5. 代码示例：WebSocket 处理
❌ 错误示例（doOnNext() 不能正确处理异步任务）
session.receive()
.doOnNext(message -> handleIncomingMessage(sessionId, message, missedPongs)) // handleIncomingMessage() 可能是异步的
.then();
🚨 问题：
handleIncomingMessage()可能是一个异步任务（例如存数据库）。doOnNext()不会等待Mono任务完成，可能导致数据处理不完整。
✅ 正确示例（使用 flatMap() 处理异步任务）
session.receive()
.flatMap(message -> handleIncomingMessage(sessionId, message, missedPongs)) // 让 Reactor 订阅 `Mono`
.then();
🔹 这样 handleIncomingMessage() 返回的 Mono<Void> 会被正确订阅，确保它的逻辑执行完成后才继续。
6. 结论
🚀 使用 doOnNext()：
- 适用于 日志、计数、打印等同步操作，不会改变数据流。
- 不会订阅异步
Mono或Flux。
🚀 使用 flatMap()：
- 适用于 异步任务（数据库查询、API 调用等），确保 Reactor 正确订阅
Mono/Flux。 - 会等待异步任务完成，然后继续数据流。
📌 记住：如果 handleIncomingMessage() 可能是异步的（返回 Mono<Void>），你应该用 flatMap()，而不是 doOnNext()！ 🚀