之前在做一个报表导出功能时我第一次真正被扁平化操作符难住。业务本身不复杂拿到一批订单ID每个ID要异步去查询订单明细明细又是多条记录最终把所有明细合并成一个数据流写进Excel。我一开始用 map 去写结果拿回来的不是订单明细流而是流里面套流FluxFlux 类型看着就难受。换成 flatMap 之后数据是对了但输出顺序全乱了再换 concatMap顺序是保住了可速度直线下降。也就是从那天起我对 flatMap 和 concatMap 的态度从会用变成了必须从底层搞懂。这篇就把 Reactor 3 里数据扁平处理的完整链路讲清楚为什么响应式编程里总绕不开 flatMap 和 concatMap它们底层到底怎么处理事件流什么时候用哪个以及我在真实项目里踩过的一些坑。适合已经掌握 map、filter 等基础操作符、但一碰到流里面套流就犯迷糊的读者。1. 从 map 到 flatMap一次不得不做的思维转换1.1 map 的边界只能做 1:1 的同步变换很多刚接触 Reactor 的同学会把 flatMap 想得很神秘其实它和 map 之间的关系非常天然。map 做的事情很单纯上游来一个元素我同步地把它变成另一个元素一对一不改变流的维度。Flux.just(o-1, o-2, o-3) .map(orderId - orderId.toUpperCase()) .subscribe(System.out::println);这段代码里每个字符串进来出去还是字符串流的形状没有变。这种 1:1 变换在同步场景下非常舒服你不需要考虑异步、不需要考虑顺序因为 map 本身就是阻塞的。但实际业务里一个元素映射出一个结果的情况远远不够。你往往碰到的是一个元素对应着一个异步操作而异步操作的结果又是一个容器。比如订单ID要查询出明细列表再比如要访问某个服务去拿配置数据返回的是一个集合。这时候如果你继续用 map写出来的代码会变成什么样Flux 里的每个元素被映射成一个 Publisher整个流就变成了嵌套流。Reactor 的泛型会把这种结构完整地暴露出来编译器会直接告诉你这个嵌套关系存在但你的下游订阅者并不关心这件事它只想拿到扁平的订单明细。这就是扁平化操作符存在的理由把每个上游元素 - 一个内部流这种结构重新压平成一个对外统一发射元素的下游流。map 管不了这件事因为它结构上就没有展开这一层。1.2 当每个元素自己开出一个流用一个最常见的场景来感受一下这种流里套流。假设上游是一串用户ID下游是用户详情对象每个用户ID都需要去远程接口拉一次数据。远程接口的函数签名是这样的FluxUserDetail fetchDetail(String userId);如果你坚持用 mapFlux.just(u-1, u-2, u-3) .map(this::fetchDetail) // 返回 FluxFluxUserDetail静态类型已经清楚地告诉你map 只是把每个字符串换成了另一个流对象这些流对象彼此独立并没有合并到你想要的同一个下游流里。你把这样的流交出去订阅者要面对的是一层嵌套结构处理起来极其别扭。用 flatMap 就不是这个效果了。同样是这个映射函数flatMap 会自动订阅每个由用户ID产生出来的内部流并把它们发出的元素重新汇聚到一个外层的公共流中Flux.just(u-1, u-2, u-3) .flatMap(this::fetchDetail) .subscribe(detail - System.out.println(detail.getName()));从订阅者的视角看它拿到的就是一个不断的 UserDetail 流至于这个元素是 u-1 来的还是 u-3 来的它根本不关心。这也是扁平化这个词最直观的含义把二维的流结构拉平成一维。打个比方map 就像一个人递给你一张纸flatMap 则像他递给你一叠纸同时帮你把纸摊开铺到桌面上。铺开之后你看不到原来的一叠这个层级只看到一张张独立的纸张。1.3 扁平化操作符家族不止 flatMap 和 concatMapReactor 3 里和扁平化相关的操作符其实有一族除了标题里的 flatMap、concatMap还有 flatMapSequential、switchMap 这些。它们的共性都是上游一个元素映射为一个内部 Publisher区别在于合并内部流的方式以及对顺序和并发的取舍。我在实际项目里真正高频使用的也就三四个flatMap 解决并发展开问题concatMap 解决顺序强制问题flatMapSequential 在两者之间找平衡switchMap 则适合最新结果覆盖旧结果的场景。这篇先把 flatMap 和 concatMap 讲透因为它们代表了扁平化里两个最核心的极端完全并发无序和完全串行有序。理解了这两个极端flatMapSequential 和 switchMap 你自然而然就能推出来。2. flatMap 核心机制并发展开与无序合并2.1 一样的代码不一样的输出顺序flatMap 最迷惑人的地方在于它表面上看起来和 concatMap 的写法几乎一样就是方法名差一个前缀但行为差异巨大。先用一个简单的延迟模拟来观察Flux.just(u-1, u-2, u-3) .flatMap(this::fetchName) .subscribe(System.out::println);对应的映射函数给不同 ID 制造不同延迟private MonoString fetchName(String id) { return Mono.delay(Duration.ofMillis(u-1.equals(id) ? 80 : 20)) .map(tick - id -detail); }我先卖个关子你猜一下输出顺序是什么。直觉可能会告诉你既然上游发射顺序是 u-1、u-2、u-3那输出应该也是这个顺序。但 flatMap 的答案恰恰反直觉u-1 虽然先进入流但它延迟了 80msu-2 和 u-3 的延迟只有 20ms所以先输出的反而是后面的元素。这就是 flatMap 的核心特征它会立刻订阅所有内部流多个内部流之间的发射顺序是交错的谁先有结果谁先往公共流里发。如果你依赖先查询的先输出这种假设用 flatMap 就会踩到顺序坑。2.2 flatMap 内部做了什么多路订阅与合并从源码层面看flatMap 做的事情可以简化成三步。第一步上游每到达一个元素就通过映射函数生成一个对应的内部 Publisher第二步框架立即订阅这个内部 Publisher并且这个订阅是并发的不是等上一个内部流结束再订阅下一个第三步所有并发运行中的内部流谁发射了元素谁就把元素丢进同一个合并缓冲区再传给下游。这个合并缓冲区就是无序的根源。它本质上是一种 merge 行为多个生产者在同时生产生产者之间没有顺序协议谁先生产完谁先绕过缓冲区抵达下游。所以在实际运行中你可能看到u-2 的结果先出来u-3 的结果跟上u-1 的结果最后才到。这不是偶发而是 flatMap 的常态化表现。2.3 背压、并发度与合并缓冲flatMap 还有一个容易被人忽略的重载参数并发度。默认情况下flatMap 最多同时订阅 256 个内部流也就是说上游来元素的速度再快它也只会优先处理 256 个其余的进入排队等待。flatMap(this::fetchDetail, 16); // 限制最多16个内部流并发为什么要限制并发度我见过不少新人在并发量小的测试里跑得挺好一上生产就把下游数据库连接池打爆。上游可能一瞬间来一万个IDflatMap 默认并发 256加上每个 ID 的查询要占用一条数据库连接等于瞬时需要 256 个连接很多服务根本扛不住。这时候用第二参数把并发度调到 16、32既保留并发加速又保护下游资源。如果你还需要更精细的控制flatMap 还有第三个参数 prefetch控制每个内部流预取多少元素。日常业务里先调整 concurrency 就够了prefetch 一般用默认值即可。2.4 flatMap 最适用的场景和明显的反例我个人的经验flatMap 适合下面这几类场景每个元素映射出的内部流耗时不一致你希望更快的结果先出来不在乎顺序。下游处理逻辑不依赖上游发射顺序比如把数据写入消息队列消息之间互相独立。你想利用并发加速把串行等待变成并行等待。反过来如果存在以下信号就别用 flatMap下游对处理顺序有硬性要求比如必须按输入ID顺序写Excel。内部流之间互相依赖第二个元素的计算要等第一个元素的结果。前端需要展示先来先显示的列表但带上了明确的顺序预期。顺序要求这种东西刚开始写响应式代码时容易被忽略因为小数据量下 flatMap 的乱序可能完全看不出来顺序碰巧和输入一致。一旦数据量上来或者某个服务偶发慢查询乱序问题就会突然暴露而且这种 bug 非常难定位。3. concatMap 的顺序保证从哪里来串行订阅与它的成本3.1 concatMap 的慢是很诚实的concatMap 同样是把上游元素映射成内部流但它的合并方式完全不同。它不会并发订阅所有内部流而是严格地一个接一个先订阅第一个元素生成的内部流等这个内部流完完整整地发出最后一个元素并结束再订阅第二个元素生成的内部流。所以你看到的输出序列和上游输入序列永远是严格对应的。还是刚才那个延迟模拟Flux.just(u-1, u-2, u-3) .concatMap(this::fetchName) .subscribe(System.out::println);不管 u-1 内部的查询有多慢u-2 的结果都只能在 u-1 的流完全结束后才开始处理。这就是 wait-for-complete 的行为。它保证了顺序代价是彻底放弃了并行。用生活类比来说flatMap 是超市开了多个收银台大家谁排完队谁先走concatMap 是只有一个收银台第一个人结完账走了第二个人才能开始把商品放上收银台。3.2 concatMap 的预取机制为了不断流如果你读过 concatMap 的实现会发现它没有想象中那么死板它内部其实有一个预取机制默认预取数量是 2。也就是说concatMap 当前正在处理第一个内部流时上游可以提前把下一个元素准备好这样一旦第一个流结束第二个流立刻就能接着订阅不用等上游重新发射数据。这个设计本质上是为了解决串行场景下的空转问题。它让 concatMap 在保序的同时尽量保持流式节奏但千万不要误解成 concatMap 允许并行它只是做了流水线式的缓冲执行仍然是串行。concatMap(this::fetchName, 4); // 增大预取稍微减小切换开销调大预取量在内部流很短、切换频繁的场景下有点作用但作用有限核心瓶颈还是被串行执行拖住的。3.3 做一个简单的延迟账本为了直观理解 concatMap 的成本我们算一笔账。假设上游有 10 个元素每个元素对应的内部操作平均耗时 200ms。用 flatMap 并发执行理论上总耗时接近最慢的那个元素大约 200 到 300ms 就能全部完成。用 concatMap 串行执行总耗时是 10 乘以 200ms也就是约 2000ms。这笔延迟账在元素多、单个操作慢的场景下会被放得很大。所以我会在代码评审时特别留意如果有人把 flatMap 直接改成 concatMap 来修乱序 bug那他其实是在用性能换顺序这种取舍不是不行但要想清楚代价。3.4 concatMap 真正不可替代的场景既然 concatMap 这么慢那什么时候必须用它我的经验是凡是内部流存在时间相关性、强依赖性的场景都逃不开它。比如我做过一个设备配置下发功能一批配置项之间是有先后约束的先写网络参数再重启设备最后校验状态。如果并发执行第二步可能在第一步还没完成时就启动整个流程都会出错。这种场景下concatMap 的顺序保证不是性能问题而是正确性问题。再比如金融业务里的流水处理依赖上一笔状态变更完成后才能处理下一笔的规则这种场景你不可能用 flatMap 去赌运行时的顺序。有一类更隐蔽的场景也适合 concatMap内部流本身有副作用并且副作用之间存在上下文关联。比如依次给用户发提醒短信同一个用户的两条短信顺序颠倒了后端不报错但前端客服会看到历史顺序混乱。这种逻辑层面的关联靠测试不一定能覆盖但它真实存在于业务语义里。4. 实战选择我在代码里按这三条规则挑操作符4.1 规则一先回答顺序重不重要这是我选型时问自己的第一句话。判断标准很简单把输入顺序和输出结果顺序打乱下游能感知到变化吗能感知就必须放弃 flatMap转而考虑 concatMap 或 flatMapSequential感知不到就大胆用 flatMap 享受并发优势。有些场景的顺序要求藏在细节里比如你要把处理结果回写到一个数组数组索引按输入顺序排。这种情况下即使业务看起来不敏感写入位置一变就会错位本质还是顺序敏感。下面这个表是我在评审时经常用来和同事对齐的场景特征保序需求并发需求首选操作符数据上报、日志清洗无高flatMap依赖严格的流水处理高低concatMap结果展示、报表导出高高flatMapSequential秒级超时的实时查询低极高flatMap 配合限流4.2 规则二看内部流的类型是 Mono 还是 Flux这里有一个很多教程没强调的细节扁平化操作符的映射函数返回的是 Publisher具体是 Mono 还是 Flux会影响你对并发和顺序的预期。内部是 Mono 时每个内部流只会发出一个元素就结束flatMap 和 concatMap 的顺序差异主要体现在第一个结果何时出现而不会出现一个内部流的多个元素被另一个流打断的情况。内部是 Flux 时就不一样了一个内部流可能连续发出多个元素flatMap 会把不同流的元素交错合并顺序问题会被放大。我之前排查过一个诡异的现象上游数据量很小flatMap 的输出顺序和输入顺序一致同事困惑为什么会产生乱序 bug。最后发现真正的差异在内部映射函数某个分支返回了 Flux另一个分支返回 Mono不同分支的时序行为不一致才导致线上偶发乱序。所以写映射函数时明确返回值是 Mono 还是 Flux对你的心智负担影响很大。4.3 规则三用并发度参数保护下游资源不管选择 flatMap 还是 flatMapSequential我都建议显式设置并发度别依赖 256 这个默认值。256 这个数字对大部分内部操作来说都太大了尤其是你的内部操作要访问数据库、调用第三方HTTP接口时瞬间 256 路并发打出去没有几个外部服务能稳稳接住。我之前有一个数据同步任务上游是几万条变更记录每条变更要去查一个老系统的接口。最开始没做并发限制上线第一天就把老系统的连接池打崩了对方直接拒绝后续请求。后来改成 flatMap(..., 32)并加了重试和退避整个任务从频繁超时变成稳定跑完。限制并发的本质是控制你对别人的冲击程度这和响应式本身没关系但操作符给了你这个旋钮不用就是浪费。另外如果并发度设得太低比如 4而每个内部流又要做网络交互整体吞吐可能还不如串行优化得好。并发度设置不是一个固定值要根据下游服务的响应时间、连接池大小、单次任务的体量综合估算。4.4 flatMapSequential既要并发又要保序很多人会在 flatMap 和 concatMap 之间纠结其实 Reactor 还给了第三个选项flatMapSequential。它结合了两边的优点内部流是并发执行的但最终发射到下游的元素会按照上游输入的顺序重新排列。用法很简单Flux.just(u-1, u-2, u-3) .flatMapSequential(this::fetchDetail)它的内部实现会为每个元素的结果保留一个槽位哪个内部流先完成先放进属于自己的槽位但发射动作必须按顺序来第一个槽位没有结果后面的结果就算先到也得等待。flatMapSequential 适合报表导出这类场景既希望大量查询并发跑又必须保证输出顺序与输入顺序一致。我自己的经验是只要并发安全和顺序要求同时存在无脑选 flatMapSequential 大概率比 concatMap 合适因为它没有完全丢掉并发收益。但它也有代价每个内部流的生命周期变长因为先完成的结果要缓存在槽位里等前序结果完成才能发出去内存占用比 flatMap 和 concatMap 高一些。数据量极大时要注意内存水位。5. 我踩过的几个扁平化相关的坑5.1 在 flatMap 里用了阻塞调用背压和调度全废了刚开始写响应式时最容易犯的错就是在 flatMap 的映射函数里直接写阻塞代码.flatMap(id - { String result jdbcTemplate.queryForObject(...); // 阻塞调用 return Mono.just(result); })从语法上说代码能跑但语义上完全跑偏了。flatMap 的映射函数可能在事件循环线程上被回调一个线程池就那么几个线程你在里面 sleep 或做 JDBC 同步查询等于把整个并发度卡死。更麻烦的是背压机制也被绕过因为你没有通过异步信号告知下游当前处理的进度。正确做法是把阻塞调用包装成异步比如使用专门的阻塞适配器或者提前把数据查好再进入响应式流。如果实在绕不开至少要换一个足够大的弹性线程池避免占住事件循环线程。5.2 内部流为空时flatMap 会静默跳过有一个很容易忽略的边界情况映射函数返回的内部流可能一个元素都不发直接就结束了。这时候 flatMap 的行为是默默跳过这个元素继续处理其他内部流。它不会报错也不会中断。这种设计本身没问题但容易掩盖逻辑 bug。有一次我做一个用户存档迁移上游是用户ID内部流去查该用户的历史订单新用户没订单内部流自然是空的。结果下游汇总时发现用户数少了排查半天才发现是空流被静默过滤掉了。如果你确实需要感知某些元素没有结果可以在内部流里加上默认值.flatMap(id - fetchOrders(id) .defaultIfEmpty(new EmptyOrder(id)))这样每个用户至少会产出一个元素下游更容易判断哪些是真正的空结果。5.3 内部流异常会直接拖垮整个流程flatMap 的另一个默认行为是内部流发出 error 信号后外层流会立刻终止。你在一个内部流里的下游服务超时可能导致整批数据处理中断而不是只跳过那一条。这个行为和 Java 8 Stream 里一个元素抛异常导致整个流失败很像。响应式编程本身也是这个哲学错误用信号表达而非吞掉。所以如果你的业务要求单条失败不影响整体必须在内部流里先做错误处理例如用 onErrorResume 把异常转成一个结果而不是把错误抛到外层。我一般会在扁平化操作之前做一个错误兜底策略.flatMap(id - fetchDetail(id) .onErrorResume(e - Mono.just(defaultDetail(id))))这样单个查询失败只会出来一个默认结果外层流能继续跑完。5.4 concatMap 在上游快、下游慢时也会内存膨胀concatMap 看起来是串行的好像不会堆积太多数据但其实它也有预取队列。上游发射速度远大于处理速度时上游还没来得及发完的上万个元素会源源不断进入缓冲区。等待处理的数据量越大内存占用就越高。我遇到过一次 OOM场景是从数据库一次性查了五万条记录下游对每一条做一次远程调用用了 concatMap 保序。远程调用平均要几百毫秒五万条全部处理完成需要很长时间而数据库端的数据还在往上游流里灌背压没能把内存压住最终堆内存被打爆。遇到这种场景我的建议是先把数据分批控制上游发射速度或者在 concatMap 前后搭配 limitRate 进行限速。另一个思路是改用 flatMapSequential 配合合适的并发度至少内部流并发执行能更快消化队列降低堆积风险。5.5 调试顺序问题用 log 观察没那么玄如果你遇到了疑似乱序的 bug最有效的工具其实是 log 操作符。它会在每个事件发生时打印 onNext、onComplete、onSubscribe 等信号配合元素内容你能直观看到发射顺序和耗时。Flux.just(u-1, u-2, u-3) .flatMap(this::fetchName) .log(flat-map-demo) .subscribe();跑一次之后把日志时间线拉出来flatMap 和 concatMap 的差异一目了然。我个人在写复杂响应式链路时习惯先在每个操作符之间临时加 log等流程稳定后再删掉。这个习惯帮我定位了很多看起来玄学的顺序怪问题。6. 一句话口诀和两个想推荐的练习如果看完这篇只记住一件事那我希望是这句顺序重要用 concatMap性能优先用 flatMap既要顺序又要并发用 flatMapSequential。这句话我每次评审响应式代码时都会在脑子里过一遍基本能解决绝大多数选型问题。最后给你两个可以动手验证的小练习练完对扁平化的理解会比读十篇博客都有用。第一个练习模拟一个用户列表转用户详情列表的任务给不同用户设置不同的延迟时间分别用 flatMap、concatMap、flatMapSequential 跑三遍打印输出顺序。你会在控制台直观看到三种操作符的时序差异。第二个练习在你自己的项目里找一个目前用 map 处理集合数据的响应式接口试着把它改成 flatMap 并加好并发度参数观察下游压力和耗时变化。只要不做顺序敏感的业务大概率能发现明显吞吐提升。我在实际使用中的一个体会是flatMap 和 concatMap 之所以容易让人困惑不是因为 API 复杂而是它们迫使你直面顺序和并发不可兼得这个真实约束。在熟悉这两个极端之后再去理解 flatMapSequential、switchMap 这些高阶变体整个知识体系就串起来了。也建议你在下次遇到乱序 bug 时先别急着把 flatMap 换成 concatMap仔细想想业务是否真的完全串行也许 flatMapSequential 这个中间选项才是更划算的那一个。