三种转换操作符的语义
Reactor 提供了三种核心的「转换」操作符，它们的区别是很多工程师在使用时的第一个困惑点：
| 操作符 | 输入:输出 | 同步/异步 | 顺序保证 | 常用场景 |
|---|---|---|---|---|
map | 1:1 | 同步 | 保持 | 数据格式转换 |
flatMap | 1:N | 异步 | 不保证 | 并发查询、并行调用 |
concatMap | 1:N | 异步 | 保证 | 有序依赖调用 |
map：同步一对一
map 是最简单的转换——每个输入元素产生一个输出元素，转换逻辑是同步的：
Flux.range(1, 5)
.map(i -> i * i)
.subscribe(System.out::println);
// 输出: 1, 4, 9, 16, 25
底层实现
map 内部创建了一个 FluxMap，它在 onNext 中同步调用用户函数，然后同步调用下游的 onNext：
// 伪代码（简化自 reactor-core）
@Override
public void onNext(T t) {
V v = mapper.apply(t); // 同步转换
downstream.onNext(v); // 同步发射
}
异常处理
如果 map 中的函数抛出异常，异常会直接传播为 onError：
Flux.range(1, 5)
.map(i -> {
if (i == 3) throw new RuntimeException("boom at 3");
return i * 2;
})
.subscribe(
System.out::println,
e -> System.out.println("Error: " + e.getMessage()),
() -> System.out.println("Complete")
);
// 输出: 2, 4, Error: boom at 3
注意：map 出错后，流终止。不会继续处理 4 和 5。
flatMap：异步一对多，无序
flatMap 是解决「每个元素触发一个异步操作」场景的核心操作符。
基本用法
Flux.range(1, 3)
.flatMap(i -> fetchUser(i)) // fetchUser 返回 Mono<User>
.subscribe(System.out::println);
这里 fetchUser(1)、fetchUser(2)、fetchUser(3) 是并发执行的。哪个先返回，User 就先出现在输出中。
无序性证明
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.time.Duration;
public class FlatMapOrderDemo {
public static void main(String[] args) throws Exception {
Flux.range(1, 5)
.flatMap(i -> Mono.just(i)
.delayElement(Duration.ofMillis((6 - i) * 50)) // 元素越大延迟越短
)
.subscribe(i -> System.out.print(i + " "));
Thread.sleep(500);
System.out.println();
}
}
输出（示意）：
5 4 3 2 1
解释：
i=1延迟 250msi=2延迟 200msi=5延迟 50ms- 所以 5 最先到达，1 最后到达。
并发度控制
flatMap 有一个关键参数——concurrency，控制同时展开的异步流数量：
Flux.range(1, 100)
.flatMap(i -> fetchUser(i), 10) // 最多同时 10 个并发请求
.subscribe();
如果不指定，默认是 256（Reactor 的默认值）。这意味着如果有 1000 个元素，会有 256 个请求并发发出，其余等待。
背压不是「无限缓冲」
flatMap 不会把上游元素无限制地堆进队列。它按 concurrency 向上游请求：默认 concurrency = 256，每个内层流 prefetch = 32。内存上界大致是 concurrency × prefetch 加上内层流自身的缓冲，而不是「上游有多少就存多少」。
真正的风险有两个：
- 内层流太慢：256 个内层同时挂起，每个都占着连接、线程或内存。这时不是队列无界，而是并发占用的资源总量过大。
- 内层流不尊重背压：例如
Flux.create在onRequest之外持续next()，或者自己实现了无界缓冲，问题出在内层而不是 flatMap。
控制方式是把两个参数显式调小：
Flux.range(1, 1000)
.flatMap(i -> slowOperation(i), 10, 2) // concurrency=10, prefetch=2
.subscribe();
把 onBackpressureBuffer(100) 放在 flatMap 之前是常见的错误位置：它会对上游采取无界请求再缓冲，反而把 flatMap 的并发限制架空了。背压应该由 flatMap 的 concurrency/prefetch 和下游的消费速度来约束。
concatMap：异步一对多，有序
concatMap 和 flatMap 的区别只有一个：顺序保证。
Flux.range(1, 5)
.concatMap(i -> Mono.just(i)
.delayElement(Duration.ofMillis((6 - i) * 50))
)
.subscribe(i -> System.out.print(i + " "));
Thread.sleep(500);
输出：
1 2 3 4 5
即使 i=5 的延迟最短，它也必须等 i=4 完成后才能发射。
代价
顺序保证的代价是吞吐量下降：
| 场景 | flatMap | concatMap |
|---|---|---|
| 5 个请求，每个 100ms | ~100ms 总耗时（并发） | ~500ms 总耗时（串行） |
| 内存占用 | 高（需要队列缓冲） | 低（一次只处理一个） |
| CPU 利用率 | 高（多个内层并发） | 低（同一时刻只推进一个内层） |
什么时候用 concatMap
- 数据库查询有依赖关系（查询 A 的结果用于查询 B）。
- 文件处理必须按顺序（日志文件不能乱序）。
- 用户操作的命令序列（撤销/重做）。
map vs flatMap vs concatMap：决策树
每个元素需要转换？
├── 同步转换 → map
└── 异步转换 →
├── 结果顺序不重要 → flatMap
└── 结果必须有序 → concatMap
flatMap 的 merge 深度
flatMap 内部使用 merge 语义——把多个异步流的输出合并到一个流中。merge 有一个参数 prefetch，控制从每个子流预取的元素数：
Flux.range(1, 10)
.flatMap(
i -> Flux.range(i * 10, 3), // 每个元素展开成 3 个
5, // concurrency = 5
2 // prefetch = 2
)
.subscribe(System.out::println);
prefetch 影响的是背压行为：如果子流发射很快，但下游消费慢，prefetch 限制了每个子流的缓冲量。
实跑对比
public class MapVsFlatMapVsConcatMap {
public static void main(String[] args) throws Exception {
System.out.println("=== map ===");
long start = System.nanoTime();
Flux.range(1, 5)
.map(i -> i * 2)
.blockLast();
System.out.println("耗时: " + (System.nanoTime() - start) / 1_000_000 + "ms");
System.out.println("\n=== flatMap （无序）===");
start = System.nanoTime();
Flux.range(1, 5)
.flatMap(i -> Mono.just(i * 2).delayElement(Duration.ofMillis(10)))
.blockLast();
System.out.println("耗时: " + (System.nanoTime() - start) / 1_000_000 + "ms");
System.out.println("\n=== concatMap （有序）===");
start = System.nanoTime();
Flux.range(1, 5)
.concatMap(i -> Mono.just(i * 2).delayElement(Duration.ofMillis(10)))
.blockLast();
System.out.println("耗时: " + (System.nanoTime() - start) / 1_000_000 + "ms");
}
}
预期输出（近似）：
=== map ===
耗时: 0ms
=== flatMap （无序）===
耗时: 15ms
=== concatMap （有序）===
耗时: 55ms
map同步执行，几乎无延迟。flatMap并发执行，总耗时 ≈ 单个延迟。concatMap串行执行，总耗时 ≈ 单个延迟 × 元素数。
常见错误
错误 1：在 flatMap 里做同步阻塞操作
// 错误！阻塞了 flatMap 的线程
Flux.range(1, 100)
.flatMap(i -> {
Thread.sleep(100); // 阻塞！
return Mono.just(i);
})
.subscribe();
flatMap 的 mapper 在订阅线程上同步执行，Thread.sleep 会把整条订阅线程卡住，内层流根本没有机会并发。正确做法是把阻塞调用交给弹性调度器：
Flux.range(1, 100)
.flatMap(i -> Mono.fromCallable(() -> {
Thread.sleep(100);
return i;
}).subscribeOn(Schedulers.boundedElastic())
)
.subscribe();
错误 2：以为 flatMap 会等所有子流完成
// 错误理解：以为会先打印所有 onNext，再打印 "done"
Flux.range(1, 3)
.flatMap(i -> fetchUser(i))
.doOnComplete(() -> System.out.println("done"))
.subscribe();
实际上 doOnComplete 在所有子流都完成后才触发，但 subscribe 的 onNext 是交错到达的。
错误 3：用 flatMap 做有序操作
// 错误：需要顺序，但用了 flatMap
Flux.just(order1, order2, order3)
.flatMap(this::processOrder) // 可能乱序！
.subscribe();
// 正确：用 concatMap
Flux.just(order1, order2, order3)
.concatMap(this::processOrder) // 保证顺序
.subscribe();
小结
| 操作符 | 核心语义 | 吞吐量 | 顺序 | 风险 |
|---|---|---|---|---|
map | 同步 1:1 变换 | 最高 | 保持 | 函数抛异常终止流 |
flatMap | 异步 1:N 展开，无序 | 高 | 不保证 | 内部队列 OOM、并发耗尽 |
concatMap | 异步 1:N 展开，有序 | 低 | 严格保证 | 串行吞吐量低 |
选择建议：
- 能用
map就不用flatMap——同步操作更简单、更可预测。 - 需要并发且顺序不重要时用
flatMap，记得设置concurrency。 - 需要顺序时用
concatMap，接受吞吐量代价。
下一节，我们讲解组合操作符：zip、merge 和 combineLatest——它们把多个流合并成一个。
阅读导航
上一篇：4.1 Mono 与 Flux：零个、一个与多个元素
下一篇：4.3 组合操作符：zip、merge、combineLatest
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。