Reactor 3 里操作符多到让人眼花缭乱但如果让我挑一个“语义最直观、细节最多坑”的我会第一个想到zip也就是标题里说的数据压缩。它的本质并不是把字节流压成小体积文件而是把多个数据流按位置配对像拉链一样一齿对一齿地合并成一个新的数据流而它的小弟zipWith则是专门处理两个流之间合并的快捷入口。这篇博客不打算走教科书路线我会从一个真实项目的角度把Reactor 3的zip/zipWith的合并语义、与merge的区别、背压表现、空流处理和实战案例一次讲透。适合正在用 Reactor 改造异步代码的同学也适合刚接触响应式编程、想搞明白“到底什么时候该用 zip 而不是 merge”的新手。1. 先搞懂 zip 是什么从“拉链”的比喻说起1.1 什么叫“按位置配对”zip这个名字来自函数式编程里的Zip组合子最早可以追溯到列表处理时代的zip函数。它的基本语义非常简单从每个上游数据流里各取一个元素把它们组成一个配对然后交给你提供的组合函数去生成下游元素。我用生活化的例子解释一下。假设你有两个队列左边是[A, B, C]右边是[1, 2, 3]。zip会把它们按顺序一一对应A配1B配2C配3。配对成功后你可以用(char, number) - char number这样的组合函数得到A1, B2, C3。整个过程就像把两条拉链的齿对齐咬合推进的步数取决于较短的那一条。这里最关键的两个字是“位置”。zip不看时间戳不关心数据到底什么时候到达它只关心“这是第几个元素”。第一个元素只跟第一个元素配对第十个元素只跟第十个元素配对。这一点是整个操作符的基石也是很多人踩坑的根源以为 zip 像merge那样“谁先来就先合并谁”实际上它严格遵循序号对齐。我们看一个最基础的代码示例FluxString names Flux.just(Alice, Bob, Cathy); FluxInteger scores Flux.just(90, 85, 95); Flux.zip(names, scores, (name, score) - name : score) .subscribe(System.out::println); // 输出 // Alice:90 // Bob:85 // Cathy:95如果其中一个流的元素更多多余的会直接被忽略。比如names有 4 个元素而scores只有 3 个那么最终只会发出 3 个结果第 4 个名字不会参与任何配对。这段行为是由“拉链”语义天然决定的不是 bug但没搞清楚的话就容易在产品里出问题。1.2 zip 与 merge、concat 的本质区别Reactor 里负责“合并”的操作符远不止zip一个。最容易被拉出来对比的是merge和concat。初学者经常把这三个放在一起问“有什么区别”我干脆用一个表格把它们的核心差异列出来操作符合并方式决定性因素典型场景concat先订阅第一个流等它 complete 后再订阅第二个流流的先后顺序按顺序拼接多个数据源merge同时订阅所有流元素按到达时间交错合并时间谁先到谁先发实时事件流合并zip同时订阅所有流元素按序号一一配对序号第几个对第几个多路结果按索引聚合concat是最容易理解的它本质上是在“接力”第一个源跑完才轮到第二个源。它不追求并发也不追求配对只是把多个数据段首尾相接。merge则是把多个源看成多条管道谁先吐出一个元素下游就先收到谁最终顺序完全由时间决定大概率是乱序的。zip跟它们有本质不同。它虽然也是并发订阅所有上游但下游看到的顺序不是“到达时间”而是“元素序号”。所有上游都必须发出第N个元素zip才会发出第N个结果如果某个上游迟迟不发其它已经准备好的元素只能在缓冲队列里等待。我举个例子假设Publisher1发出1, 2, 3Publisher2发出10, 20, 30它们到达的绝对时间完全错开。merge的结果可能是1, 10, 2, 20, 3, 30也可能是10, 1, 2, 20, 30, 3完全看调度。但zip的结果一定是110, 220, 330只要组合函数是简单的相加。1.3 什么场景下非 zip 不可理解了语义之后自然要回答一个问题到底什么场景必须用zip我的经验是当你有“多路独立计算的结果需要按下游数据的序号或索引对齐后合并”时zip是唯一自然的选择。举几个实际例子。第一个是多个远程接口并行调用用户详情、订单列表、优惠券信息三路 RPC 互不依赖但页面必须三份数据全部到齐才能渲染。这种场景用Flux.zip或Mono.zip能天然表达“并行请求 全部完成后聚合”。第二个是批处理任务里的多文件合并每个文件按行号对齐第一行对上第一行第二行对上第二行。第三个是传感器数据融合两台设备按采样序号输出数值需要在序号层级做笛卡尔式的配对计算。这些场景如果用merge去做顺序会乱用concat去做则必须等待前一个数据源完全结束既慢又不符合“多路并行”的需求。zip把“并发订阅”和“按序号对齐”两个能力合二为一这也是它在 Reactor 家族里占据独特位置的真正原因。2. zip 与 zipWith两个入口的细微差别2.1 方法签名与参数说明Reactor 3给我们提供了两个入口静态方法Flux.zip(...)和实例方法Flux#zipWith(...)。它们的核心能力一模一样但使用姿势不同。先看签名。zip的常用重载static T1, T2, O FluxO zip( Publisher? extends T1 source1, Publisher? extends T2 source2, BiFunction? super T1, ? super T2, ? extends O combinator )zipWith的签名是final T2, O FluxO zipWith( Publisher? extends T2 source2, BiFunction? super T, ? super T2, ? extends O combinator )区别看起来只是“静态方法 vs 实例方法”但实际使用体验差别很大。zipWith是挂在已有Flux上的实例方法符合响应式链路的“流式”写法读起来像自然语言fluxA.zipWith(fluxB, combinator)一眼就能看出主从关系。而Flux.zip(fluxA, fluxB, combinator)更强调“两个源是对等的”。除了两个参数的版本还有不带组合函数、返回Tuple2的重载FluxTuple2String, Integer zipped Flux.zip(names, scores); // zipped 的每个元素都是 Tuples.tuple2(Alice, 90) 这样的二元组这个重载在只需要“把数据绑在一起、不做额外业务处理”的场景下非常方便。你可以先 zip 成二元组下游再用.map()或者flatMap()去拆包处理。我个人的习惯是如果组合逻辑很短直接用zip的BiFunction如果组合逻辑很长先返回Tuple2再在下一个操作符里写清晰的处理逻辑可读性会好很多。2.2 多源合并与 Tuples 类型zipWith只能处理两个流而Flux.zip可以处理两个以上的流。Reactor 为zip提供了从 2 个到 8 个参数的系列重载每个参数都是一个Publisher组合函数则接收对应数量的入参。比如三个流的合并Flux.zip( Flux.just(A, B), Flux.just(C, D), Flux.just(E, F), (a, b, c) - a b c ) .subscribe(System.out::println); // 输出ACE、BDF当源的数量超过 8 个或者源的数量在运行时才能确定时还有基于数组和Iterable的重载。它们的组合函数入参是一个Object[]需要自己按索引取元素类型安全会差一些。FluxObject[] sources new Flux[] { flux1, flux2, flux3, flux4 }; Flux.zip(sources, (Object[] tuple) - { int a (int) tuple[0]; String b (String) tuple[1]; // ... return new Result(a, b); });不带组合函数的多源zip则返回FluxTupleN比如三个源返回FluxTuple3T1, T2, T3。Tuples工具类提供T2到T8的二元组到八元组取元素的方法是getT1(),getT2(),getT3()按序号取不要跟下标从 0 搞混。用多源zip的时候我提醒一句源的数量越多整体节奏就越取决于最慢的那一个。六路合并意味着六路里随便一路慢其余五路都得原地等着。所以多源合并前最好确认每个源的延迟量级在可控范围内否则整个响应式链路的吞吐会被拉低。2.3 zipWith 与 zip 的关系本质是同一个东西从源码实现角度看zipWith是一个很“薄”的包装。它内部最终会构造出一个zip操作只是把你当前的这个Flux作为第一个源传入的Publisher作为第二个源组合函数原样透传。换句话说fluxA.zipWith(fluxB, combinator)和Flux.zip(fluxA, fluxB, combinator)的行为几乎没有差别。那为什么两个都要提供我个人理解是 API 设计上的便利性考量。zip作为静态方法适合从零开始组装多个流zipWith作为实例方法适合在一条已经存在的链路里“顺便”引入第二个数据源。举个例子service.getBaseInfo(userId) // FluxBaseInfo .flatMap(info - info.getItems()) // FluxItem .zipWith(service.getPromotions(userId), // 另一个独立 FluxPromo (item, promo) - assemble(item, promo)) .subscribe(...);这种写法里zipWith不需要在方法开头就准备好第二个Publisher而是可以在链路的某个具体位置动态决定“此刻我要合并谁”代码的语义边界更清晰。还有一点容易忽略zipWith也有带prefetch参数的重载。prefetch控制每个上游队列的预取数量默认值是 32。这个参数在两端速率差距很大时会起作用后面我单独用一节展开讲。.zipWith(source2, combinator)不传prefetch时内部会用默认值这在大部分同步场景没有感知但在高吞吐异步场景可能成为瓶颈。2.4 更多 zip 变体Iterable 兜底与动态源合并除了固定元数的重载Reactor 3还提供了几个不太常用但关键时刻能救命的变体。第一个是接收Iterable? extends Publisher?的版本第二个是接收Publisher? extends Publisher?的版本。Iterable版本适合“源列表在循环里生成”的场景。比如我从配置中心加载十个数据源把它们放进ListFluxInteger再统一合并ListFluxInteger sourceList loadSources(); // 运行时确定的多个源 Flux.zip(sourceList, (Object[] tuple) - { int sum 0; for (Object value : tuple) { sum (int) value; } return sum; }).subscribe(...);Publisher? extends Publisher?版本则更进一步它允许上游动态地产生新的Publisher。这意味着你不需要预先知道一共有多少路数据流随着时间推移新的源会被动态纳入 zip 的配对范围。这个能力很强但它也引入了更多不确定性动态加入的源怎么和已有元素序号对齐源的数量会无限增长吗使用时建议对源的数量做上限控制防止队列无限扩张。3. 实操三个完整案例带你上手3.1 案例一并行接口调用聚合返回我在某个项目里写过这样的逻辑前端请求用户详情页后端需要同时调用用户信息服务、订单服务和优惠券服务等三个结果全部返回后组装成一个页面模型。最开始用回调写法串行调用耗时累加体验很糟糕后来改造成 Reactor 之后核心代码只要一个zip就能搞定。public MonoUserDetailPage getUserDetailPage(String userId) { MonoUserBaseInfo baseInfoMono userService.getBaseInfo(userId); MonoListOrder orderListMono orderService.getOrders(userId); MonoListCoupon couponListMono couponService.getCoupons(userId); return Flux.zip(baseInfoMono, orderListMono, couponListMono) .map(tuple - { UserBaseInfo baseInfo tuple.getT1(); ListOrder orders tuple.getT2(); ListCoupon coupons tuple.getT3(); return new UserDetailPage(baseInfo, orders, coupons); }) .next(); }这里我故意用了Flux.zip而不是Mono.zip是因为三个Mono本身也是PublisherFlux.zip完全适用。实际项目里更推荐用Mono.zip语义更精确返回Mono也更符合“最终只有一个结果”的预期。三个Mono在被订阅时会同时发起请求三个远程调用并发执行整体耗时约等于最慢的那一路这就是zip最强的价值不是它合并了数据而是它让互不依赖的请求真正并行了起来。不要小看“并行”这三个字。如果串行执行假设每路 200ms总耗时 600ms并行之后只花 200ms页面响应速度提升三倍。这类优化在响应式编程里是被zip天然支持的不需要额外引入线程池和CountDownLatch。3.2 案例二多路传感器数据按采样序号合并第二个案例来自我做过的一个数据采集相关的小工具。两台传感器设备各自输出数值流采样序号从 1 开始自增我需要把同一序号的读数放在一起做差值计算。FluxLong deviceA Flux.range(1, 100) .delayElements(Duration.ofMillis(20)) .map(i - readDeviceA(i)); FluxLong deviceB Flux.range(1, 100) .delayElements(Duration.ofMillis(30)) .map(i - readDeviceB(i)); deviceA.zipWith(deviceB, (a, b) - Math.abs(a - b)) .subscribe(diff - log.info(sample diff: {}, diff));这个案例最值得体会的是设备 A 每 20ms 产出一个读数设备 B 每 30ms 产出一个读数但zip输出的顺序完全按序号对齐设备 A 的第 5 个读数只会跟设备 B 的第 5 个读数做差值不会被时间差打乱。如果设备 A 比设备 B 快设备 A 产生的早到元素会先进入 zip 的缓冲区等待设备 B 对应的序号到达。设备 B 一旦到达两者立刻配对并发射。整个过程不需要我自己维护任何“对齐状态”zip把跨流索引对齐这件事完全封装掉了。这也是在我看来zip最“省心”的地方逻辑上的对应关系由框架保证我只需要关心组合函数怎么写。3.3 案例三数据长度不一致时的处理第三个案例是在实际开发里最容易遇到边界情况两个流长度不一致。zip的默认行为我已经说过以较短的那个流为准。看下面的运行结果Flux.zip( Flux.just(1, 2, 3, 4), Flux.just(a, b), (num, str) - num str ) .subscribe(System.out::println); // 输出 // 1a // 2b // 结束3 和 4 被忽略如果你希望反过来以较长的流为准把短的那一侧补上默认值zip本身不提供这个能力。标准做法是先通过map把短流填充成相同长度或者用defaultIfEmpty/concatWith补位。FluxString padded Flux.just(a, b) .concatWith(Flux.just(default, default)); Flux.zip(Flux.just(1, 2, 3, 4), padded, (num, str) - num str) .subscribe(System.out::println); // 输出 // 1a // 2b // 3default // 4default这种“补齐”方案虽然可行但要注意补位数据的业务含义。如果你把缺失值补成“默认”或“0”下游在计算统计指标时可能会混入假数据。我建议在补位之前先评估业务是否允许近似值如果不允许宁可让zip以短流为准提前结束也不要硬补。4. 最容易踩的坑背压、空流与异常传播4.1 最慢的源决定整体节奏zip最大的特性也是最大的坑整体节奏永远被最慢的那个源卡住。上游 A 一秒发射 100 个元素上游 B 一秒只发射 1 个那么zip下游一秒也只能收到 1 个结果。A 的 99 个元素都得排队等 B。这个行为从拉链语义上讲没毛病但如果你没预估到数据源的速率差异很容易在生产环境发现“明明每个单源都很快合起来却慢得离谱”。我做过一次性能排查最终定位到一路从数据库分页拉取的流在特定情况下会退化成慢速流导致另一路毫秒级的实时数据全被堵在缓冲区里。应对方式分两种。第一种是接受现实zip的结果本来就应该以最慢的源为准如果业务上能接受这种等待是合理的。第二种是优化源头给慢的那一路增加并行度、加缓存、或者使用「尽快返回」策略比如combineLatest。combineLatest不再严格按序号配对只要任意一路有新值就立刻用“其他路最近一次的值”参与计算。它牺牲了对齐的精确性换来了实时性。我个人的选型经验是需要强一致的对齐关系用zip只要“最新值”参与计算用combineLatest。两者不存在谁替代谁纯粹是语义适配的问题。4.2 空数据源会让 zip 直接沉默这个坑非常隐蔽而且一旦踩中结果基本是“线上静默故障”。假设你zip了两路数据其中一路查询结果为空也就是返回一个空的Flux或Mono.empty()那么整个zip的结果也会为空不会发出任何元素。为什么会这样因为zip需要每个源都发出一个元素才能配对空源永远不会发出元素所以永远凑不齐一个完整配对。于是无论另一个源有多少数据最终结果都是complete连一个元素都不给。你可能会想“这不是很合理吗”合理归合理但业务上往往不希望“某一路没数据整个页面都空掉”。比如用户没有优惠券但用户基本信息和订单列表是有的你不能因为第三路为空就把整个详情页数据吞掉。解决这个问题的两个常见姿势第一个是用defaultIfEmpty给空流兜底MonoListCoupon coupons couponService.getCoupons(userId) .defaultIfEmpty(Collections.emptyList());第二个是把空流转成Mono.error让错误处理器统一处理“缺数据”的情况由上层决定是重试还是降级展示。两种思路对应两个业务取向我一般优先选择defaultIfEmpty因为详情页这种场景“降级为空集合”比“直接失败”更符合体验预期。4.3 任一源出错整个链路快速失败zip的错误传播策略也是初学者容易忽略的。当任一上游发出error信号时zip会立即把错误传给下游并取消其他上游的订阅。这意味着一路失败整体失败不会继续等待其他路完成。这种“快速失败”策略在大多数场景是正确的。比如并行请求里用户信息服务超时了即使订单和优惠券都拿到了页面仍然需要展示错误页或做降级没有必要继续等。但要注意取消行为带来的资源问题。某个上游出错后其他上游还在处理中zip内部会主动取消它们的订阅底层如果在发 RPC取消并不一定能真正中断连接可能在服务端仍然消耗资源。所以我建议在写多路合并逻辑时给每路都加上超时控制比如.timeout(Duration.ofSeconds(3))宁可让某一路超时报错走降级也不要让它在后台悬空。4.4 prefetch 参数调优与内存占用prefetch是 zip 系操作符里最容易被无视的参数。默认值是 32含义是每个内部队列最多预取 32 个元素。当一个源的生产速度远高于另一个源且两者差距超过 32 时快的那一路上游会被zip暂停拉取直到队列里腾出空位。你可以把内部队列想象成一个小仓库prefetch就是仓库的最大容量。容量太小快源频繁被暂停可能增加上下文切换容量太大内存占用上升如果某一路永远不发射那 32 个或 100 个元素就会一直堆积在内存里。我对prefetch的经验是默认 32 够用但如果你明确知道某个源会产生短时间的突发数据可以适当调大。比如zipWith(source, 128, combinator)减少上游被暂停的次数。反过来如果合并的源本身就是低频的比如每秒才几个元素默认值完全足够不要无脑调大否则纯属浪费内存。prefetch并不是越大越好它只影响预取水位不影响合并语义。把精力花在确认“两侧速率是否匹配”上产出比往往比调参数更高。5. 从源码视角看 zip 的内部机制5.1 每个上游对应一个独立队列zip的实现让我觉得最精巧的部分是它的多路队列机制。当一个zip操作被订阅时它会同时订阅所有上游 Publisher并为每个上游维护一个独立的元素队列。每次上游发出一个元素元素就进入对应队列每次要向下游发射结果时zip会轮询所有队列只有每个队列都不为空才会各自取队首元素组成配对应用到组合函数然后发射结果。这套机制解释了三个现象。第一为什么快的源会被缓存因为它的元素进入了队列等待慢的源补齐。第二为什么空源会导致整体为空因为空源的队列永远是空的zip永远等不齐。第三为什么prefetch等于队列容量上限队列里最多缓存的元素数量天然受预取数约束超过这个数上游会被zip反向停止拉取。如果你写过响应式框架的底层代码会发现zip的队列管理逻辑本质上跟combineLatest一脉相承只是发射条件不同。zip要所有队列同时非空combineLatest则只要有一个队列有新元素就发射其他队列用“最近值”。这个视角能帮你把两个操作符放在同一个坐标系里理解记忆成本会低很多。5.2 订阅、取消与异常传导zip的订阅机制是“一次订阅内部辐射”。你订阅的是zip之后的下游流zip在你订阅时立即向所有上游请求订阅关系这与concat那种“先订阅第一个结束后才订阅第二个”的懒加载策略不同。换句话说zip从订阅那一刻开始就同时对每个上游发起了数据请求。下游取消订阅时zip会把自己收到的取消信号透传给所有上游。这个传播动作很重要它保证了不会因为下游不消费了上游还在拼命生产导致内存泄漏。在长生命周期组件里这种取消传播能力尤其关键能避免潜在的无界事件循环。异常传导方面任何一个上游的error信号都会立即中断zip主流程同时触发对其它上游的取消。这个行为在设计上是刻意的它的理论基础是“失败应当快速传播”不让错误被拖在时间轴上慢慢扩散。实际使用中我习惯在 zip 链路的末端放一个onErrorResume兜底负责把错误映射成用户可读的提示同时记录完整日志方便定位是哪一路的哪个阶段出了问题。5.3 zip 与 flatMap 并行模式的配合最后聊一个工程组合技巧zip配合flatMap做并行任务分发。flatMap的并发重载可以让一个Flux的每个元素都派生出一个新的Publisher这些Publisher并发执行最后把结果平铺回同一个流。但如果每个派生任务都需要和“另一路数据”进行按序配对直接用flatMap是做不到的这时候再叠加zipWith就很自然。我写过一段类似这样的代码FluxItem items itemService.getItems(); FluxPrice prices priceService.getPrices(); items.flatMap(item - itemService.resolveSku(item)) .zipWith(prices, (sku, price) - new SkuPrice(sku, price)) .subscribe(...);这个组合的精髓在于flatMap负责把每个元素异步展开成更细粒度的任务zipWith负责把展开后的结果与另一路流按序对齐。两条流水线各司其职链路保持单一数据流却实现了真正的并行计算和配对合并。如果只依赖flatMap你需要自己引入Map缓存去维护对应关系代码会立刻变得脏起来。不过要提醒一句flatMap改造元素顺序的能力是超出你预期的它默认的并发处理和内部队列可能会导致元素顺序与原始流不一致。如果你在flatMap之后使用zipWith要确认顺序是否依旧满足对齐要求。如果需要严格保序可以让flatMap带上concurrency 1或者改用concatMap用一点并发度换取顺序确定性。写在最后的一个小经验zip和zipWith用熟了之后我在项目里的选型标准其实慢慢沉淀成了一句自己的判断如果我需要的是“第几个元素去对应第几个元素”选zip如果我只是想让多个流的数据尽快汇合不在乎顺序merge更合适如果我需要“最新值参与运算”combineLatest才是正确的选择。这三者经常被放在一起讨论但它们的语义边界其实非常清楚多数让人困惑的问题回头看都是没有先想明白“对齐”这两个字。实际操作中我还养成了一个小习惯每次写完包含zip的链路都会刻意检查一遍“如果其中一个源是空的会怎样”。这个检查成本极低但能避免大半线上静默问题。如果你也在用 Reactor 做多路数据合并建议把zip的队列机制和空流行为刻在脑子里这是我在生产环境里踩过几次坑之后最想强调的东西。