两个阶段
一条 Reactor 链有两个时间点：
- 装配时：调用
Flux.range(...)、.map(...)、.filter(...)时。只是构造 Publisher 对象，没有数据流动。 - 订阅时：调用
.subscribe(...)时。创建 Subscriber 链，向上游请求，开始发信号。
Flux<Integer> flux = Flux.range(1, 5)
.map(i -> {
System.out.println("map: " + i);
return i * 2;
});
System.out.println("Assembly done");
flux.subscribe(System.out::println);
实测：
Assembly done
map: 1
2
map: 2
4
map: 3
6
map: 4
8
map: 5
10
map 里的打印没有在装配时出现。装配阶段只是把 FluxRange 包进一个 FluxMap，仅此而已。
冷流：每次订阅都重新执行
Flux<Integer> flux = Flux.range(1, 3);
flux.subscribe(i -> System.out.println("A: " + i));
flux.subscribe(i -> System.out.println("B: " + i));
输出：
A: 1
A: 2
A: 3
B: 1
B: 2
B: 3
同步源的第一次订阅会完整跑完，才轮到第二次。两次订阅是两个独立的 Subscriber 链，互不共享状态。需要共享执行结果时用 cache() 或 replay()，不要指望「同一条链订阅两次只执行一次」。
装配阶段到底生成了什么
Reactor 有 Scannable 接口，可以查看一条链的实际步骤：
Flux<Integer> flux = Flux.range(1, 3)
.map(i -> i + 1)
.map(i -> i * 2);
Scannable.from(flux).steps().forEach(System.out::println);
实测输出：
source(FluxRange)
map
map
两个 map 是两个独立步骤，没有被合并成一个 lambda。Reactor 不会把相邻的 map 合成 i -> (i + 1) * 2。这一点在优化代码时很重要：如果你以为框架会自动合并，就可能写出很多层 map 并期待零成本，实际并不会。
那 Fusion 是什么
Reactor 的 fusion 不是合并 lambda，而是让支持 Fuseable 的源与操作符在订阅时用更直接的路径传递元素，减少队列和边界开销。比如：
Flux.range(1, 3)
.map(i -> i + 1)
.map(i -> i * 2)
.log("two-maps")
.subscribe();
实测日志：
[main] INFO two-maps - | onSubscribe([Fuseable] FluxMapFuseable.MapFuseableSubscriber)
[main] INFO two-maps - | request(unbounded)
[main] INFO two-maps - | onNext(4)
[main] INFO two-maps - | onNext(6)
[main] INFO two-maps - | onNext(8)
[main] INFO two-maps - | onComplete()
[Fuseable] 说明下游拿到了一个可融合的 Subscriber，但 Scannable.steps() 仍然显示两个 map。融合发生在订阅路径上，不改变装配出来的结构，也不合并两个函数。
装配时的副作用：真的会提前执行
map 里的代码不是装配时执行，但构造 Publisher 时的参数表达式是。下面这个例子在装配阶段就查了一次数据库：
static int load(AtomicInteger counter) {
return counter.incrementAndGet();
}
AtomicInteger calls = new AtomicInteger();
Flux<Integer> eager = Flux.just(load(calls)); // 立即执行
Flux<Integer> lazy = Flux.defer(() -> Flux.just(load(calls))); // 订阅时才执行
System.out.println("after assembly: " + calls.get());
eager.subscribe(v -> System.out.println("eager got " + v));
System.out.println("after eager subscribe: " + calls.get());
lazy.subscribe(v -> System.out.println("lazy got " + v));
System.out.println("after first lazy subscribe: " + calls.get());
lazy.subscribe(v -> System.out.println("lazy got " + v));
System.out.println("after second lazy subscribe: " + calls.get());
实测：
after assembly: 1
eager got 1
after eager subscribe: 1
lazy got 2
after first lazy subscribe: 2
lazy got 3
after second lazy subscribe: 3
结论：
Flux.just(load(calls))在装配时调用一次load，订阅时直接复用这个值。Flux.defer(() -> Flux.just(load(calls)))在每次订阅时调用 supplier，所以两次订阅得到两个新值。- 单次订阅的
defer只执行一次，不会因为下游有多个元素而重复执行。
同类工具还有 Mono.fromSupplier(() -> load(calls))：supplier 在订阅时执行；Mono.fromCallable 也一样，区别是它可以抛受检异常。
常见误判：把 map 的副作用当成装配时执行
下面这段代码常被误认为「装配时插入了 3 次，订阅时又插入 3 次」：
Flux.range(1, 3)
.map(i -> {
database.insert(i);
return i;
})
.subscribe();
实际上 database.insert(i) 只在订阅后、每个元素经过 map 时执行一次，总共 3 次。装配阶段不会执行 lambda。如果连续订阅两次，才会变成 6 次——原因是两个 Subscriber 各自跑了一遍，不是装配时预先跑过。
真正需要在「订阅时」而不是「装配时」执行的初始化，用 defer、fromSupplier，或者 doOnSubscribe 做副作用。注意 doOnSubscribe 每次订阅都会触发，多订阅就会多次执行。
log() 的位置与方向
Flux.range(1, 3)
.log("before-map")
.map(i -> i * 2)
.log("after-map")
.subscribe();
实测：
[main] INFO after-map - | onSubscribe([Fuseable] FluxMapFuseable.MapFuseableSubscriber)
[main] INFO after-map - | request(unbounded)
[main] INFO before-map - | onSubscribe([Synchronous Fuseable] FluxRange.RangeSubscription)
[main] INFO before-map - | request(unbounded)
[main] INFO before-map - | onNext(1)
[main] INFO after-map - | onNext(2)
[main] INFO before-map - | onNext(2)
[main] INFO after-map - | onNext(4)
[main] INFO before-map - | onNext(3)
[main] INFO after-map - | onNext(6)
[main] INFO before-map - | onComplete()
[main] INFO after-map - | onComplete()
两个 log() 都是在订阅时输出，装配时不打印任何东西。顺序上有两个值得注意的点：
- 订阅信号从上往下：
after-map的onSubscribe先出现，然后才到before-map。 - 请求信号从下往上：
after-map的request先发出，再传到before-map；数据信号则从before-map流向after-map。
所以日志里的 before-map 看到的是原始值 1、2、3，after-map 看到的是翻倍后的 2、4、6。把 log() 放在链中的位置不同，看到的值就不同。
装配与订阅分离的意义
| 能力 | 说明 |
|---|---|
| 复用 | 同一个 Flux 可以被多次订阅，每次都是新的执行 |
| 组合 | 可以在没有订阅者的情况下拼装子流，再交给上层组合 |
| 延迟 | 用 defer/fromSupplier 把资源创建推迟到订阅时 |
| 可观测 | Scannable 可以在装配后检查链路结构 |
不要指望装配阶段做「函数合并」优化；也不要因为「链已经建好」就认为副作用已经执行。把这两件事分开，能避免很多「为什么数据库在启动时就连接了」和「为什么订阅两次执行了两遍」的问题。
小结
| 问题 | 结论 |
|---|---|
| 装配时执行 lambda 吗 | 不执行，只构造 Publisher |
| 两个相邻 map 会合并吗 | 不会，Scannable.steps() 显示两个 map |
Flux.just(load()) 何时执行 | 装配时，参数表达式先求值 |
Flux.defer(...) 何时执行 | 每次订阅时 |
map 里的副作用执行几次 | 每个元素一次，每次订阅各算一遍 |
log() 何时打印 | 订阅时，装配时不打印 |
阅读导航
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。