为什么需要组合操作符
实际系统中，数据往往来自多个源：
- 用户服务返回用户信息，订单服务返回订单列表，需要合并展示。
- 两个传感器分别上报温度和湿度，需要按时间戳对齐。
- 前端同时订阅 WebSocket 行情和 REST 快照，需要合并成统一数据结构。
组合操作符解决的是多流合一的问题。Reactor 提供了三种核心组合方式，它们的差异是很多工程师容易混淆的点。
zip：配对等待
zip 像拉链一样——等所有参与的流都产生一个元素后，把它们组合成一个元组发射。
Flux<Integer> numbers = Flux.just(1, 2, 3);
Flux<String> letters = Flux.just("a", "b", "c");
Flux.zip(numbers, letters)
.subscribe(tuple -> System.out.println(tuple.getT1() + " -> " + tuple.getT2()));
// 输出:
// 1 -> a
// 2 -> b
// 3 -> c
木桶效应
zip 的输出长度等于最短的输入流：
Flux<Integer> numbers = Flux.just(1, 2, 3, 4, 5); // 5 个
Flux<String> letters = Flux.just("a", "b"); // 2 个
Flux.zip(numbers, letters)
.subscribe(tuple -> System.out.println(tuple.getT1() + tuple.getT2()));
// 只输出 2 个：1a, 2b
// 3, 4, 5 被丢弃
这是 zip 的最大特点也是最大陷阱。若两个流速率不一致，快的流会被背压限制：zip 对每个源只预取有限数量（默认 prefetch = 32），队列不会无限增长。真正的代价是慢流决定整体吞吐——快流大部分时间在等，而不是在缓冲。
错误传播
Flux<Integer> a = Flux.just(1, 2, 3);
Flux<String> b = Flux.<String>error(new RuntimeException("b failed"));
Flux.zip(a, b)
.subscribe(
System.out::println,
e -> System.out.println("Error: " + e.getMessage())
);
// 输出: Error: b failed
任何一个流发出 onError，zip 立刻终止，其他流被 cancel。
merge：先 emit 先赢
merge 把多个流的数据交错发射——哪个流的元素先到，就先发射哪个。
Flux<Integer> fast = Flux.interval(Duration.ofMillis(50)).take(3).map(Long::intValue);
Flux<Integer> slow = Flux.interval(Duration.ofMillis(100)).take(3).map(Long::intValue).map(i -> i + 10);
Flux.merge(fast, slow)
.subscribe(System.out::println);
// 输出（近似）:
// 0 (fast @ 50ms)
// 10 (slow @ 100ms)
// 1 (fast @ 100ms)
// 2 (fast @ 150ms)
// 11 (slow @ 200ms)
// 12 (slow @ 300ms)
merge 的背压行为
merge 会向每个源请求 prefetch 个元素（默认 32），下游消费慢时不再继续请求，因此普通冷流不会被无限缓冲。总内存上界约为 源数量 × prefetch，不是「所有元素都堆进来」。
// 下游每秒处理 1 个，merge 会把每个源限制在 32 个未消费元素以内
Flux.merge(fastStream1, fastStream2, fastStream3)
.subscribe(this::slowProcess);
如果源是热流（如 Sinks.many().multicast()），它不会因为 merge 不请求就停止发射，这时才需要显式的溢出策略：
Flux.merge(
hotStream1.onBackpressureDrop(),
hotStream2.onBackpressureDrop(),
hotStream3.onBackpressureDrop()
).subscribe(this::slowProcess);
对普通冷流加 onBackpressureDrop 只会引入不必要的丢数据，merge 自身已经按下游需求回压。
merge 与 mergeOrdered
mergeOrdered 在 merge 的基础上按比较器排序：
Flux.mergeOrdered(
Comparator.naturalOrder(),
Flux.just(3, 1, 4),
Flux.just(2, 5)
).subscribe(System.out::println);
// 输出: 1, 2, 3, 4, 5
代价：需要内部缓冲和排序，内存占用比 merge 高。
combineLatest：最新值组合
combineLatest 的行为和前两者都不同：
- 等待所有流都产生至少一个元素。
- 任何一个流产生新元素时，用该流的新值 + 其他流的最新值组合发射。
Flux<String> prices = Flux.interval(Duration.ofMillis(100))
.map(i -> "price=" + (100 + i));
Flux<String> volumes = Flux.interval(Duration.ofMillis(150))
.map(i -> "volume=" + (1000 + i));
Flux.combineLatest(prices, volumes, (p, v) -> p + ", " + v)
.take(5)
.subscribe(System.out::println);
输出（示意）：
price=100, volume=1000 // 两个流都有了第一个值
price=101, volume=1000 // prices 更新，volumes 用最新值
price=101, volume=1001 // volumes 更新，prices 用最新值
price=102, volume=1001 // prices 更新
price=103, volume=1001 // prices 更新
初始等待陷阱
combineLatest 在所有流都产生第一个元素之前，不会发射任何数据：
Flux<String> fast = Flux.interval(Duration.ofMillis(50)).map(i -> "fast=" + i);
Flux<String> slow = Flux.interval(Duration.ofMillis(500)).map(i -> "slow=" + i);
Flux.combineLatest(fast, slow, (f, s) -> f + ", " + s)
.take(3)
.subscribe(System.out::println);
// 前 500ms 没有任何输出！
// 500ms 后才出现第一条
如果你的业务需要「只要有数据就立即输出」，combineLatest 不适合。考虑用 merge 或给缺失流加默认值。
combineLatest 的默认值处理
Flux<String> withDefault = Flux.concat(
Mono.just("default"),
realStream
);
Flux.combineLatest(withDefault, otherStream, (a, b) -> a + ", " + b)
.subscribe(System.out::println);
这样 withDefault 立即产生第一个值，不会阻塞 combineLatest。
三种操作符的对照实验
public class CombineOperatorsDemo {
public static void main(String[] args) throws Exception {
Flux<Integer> a = Flux.interval(Duration.ofMillis(100)).take(3).map(Long::intValue);
Flux<Integer> b = Flux.interval(Duration.ofMillis(150)).take(3).map(Long::intValue).map(i -> i + 10);
System.out.println("=== zip ===");
Flux.zip(a, b)
.subscribe(t -> System.out.println("zip: " + t.getT1() + ", " + t.getT2()));
Thread.sleep(600);
System.out.println("\n=== merge ===");
Flux.merge(a, b)
.subscribe(v -> System.out.println("merge: " + v));
Thread.sleep(600);
System.out.println("\n=== combineLatest ===");
Flux.combineLatest(a, b, (x, y) -> x + ", " + y)
.take(5)
.subscribe(v -> System.out.println("combineLatest: " + v));
Thread.sleep(600);
}
}
输出（近似）：
=== zip ===
zip: 0, 10
zip: 1, 11
zip: 2, 12
=== merge ===
merge: 0
merge: 10
merge: 1
merge: 11
merge: 2
merge: 12
=== combineLatest ===
combineLatest: 0, 10
combineLatest: 1, 10
combineLatest: 1, 11
combineLatest: 2, 11
combineLatest: 2, 12
注意：
zip输出 3 条（最短流的数量）。merge输出 6 条（所有元素交错），具体交错顺序取决于调度时序。combineLatest输出 5 条（take(5)），用最新值组合。a只发出 0、1、2，所以组合里最后一对是2, 12（b的 2 被映射成 12），不会再出现3。- 300ms 附近
a = 2与b = 1（映射后为 11）几乎同时到达，谁先进入combineLatest由调度时序决定：b先到会得到上面的1,11 → 2,11，a先到则得到2,10 → 2,11，两种顺序都合法。
选择指南
| 场景 | 推荐 | 原因 |
|---|---|---|
| 需要配对（如 join） | zip | 严格一对一配对 |
| 流数量不对等 | merge | 不丢弃任何元素 |
| 实时仪表盘 | combineLatest | 任一更新就刷新 |
| 需要排序 | mergeOrdered | 按比较器排序 |
| 背压敏感 | 先背压再组合 | merge/zip 内部会缓冲 |
背压策略
组合操作符的背压行为：
| 操作符 | 背压行为 |
|---|---|
zip | 向每个上游请求 prefetch 个（默认 32），配对消费后补请求 |
merge | 向每个上游请求 prefetch 个（默认 32），下游消费后补请求 |
combineLatest | 向每个上游请求 prefetch 个，只保留每个源的最新值 |
对于 zip，慢流会限制整体吞吐，但内部队列有界。给快流加 onBackpressureBuffer 是把缓冲放错了层——zip 已经按 prefetch 控制请求量。只有当某个源是热流、无法被回压时，才需要在源侧选择丢弃或有限缓冲。
小结
| 操作符 | 输出时机 | 输出长度 | 背压风险 |
|---|---|---|---|
zip | 所有流都有元素 | 最短流长度 | 快流队列积压 |
merge | 任一元素到达 | 所有元素之和 | 内部队列积压 |
combineLatest | 任一元素到达 | 无限（需 take） | 最新值缓冲 |
理解三种操作符的差异后，你可以根据业务需求正确选择：
- zip：严格配对，如数据库 join、多字段校验。
- merge：简单聚合，如日志合并、多源数据汇总。
- combineLatest：实时刷新，如仪表盘、配置热更新。
下一章，我们讲错误处理——当流中的某个操作失败时，如何恢复、重试、超时和取消。
阅读导航
上一篇：4.2 转换操作符：map、flatMap、concatMap
下一篇：5.1 错误处理：onErrorReturn、onErrorResume、retry
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。