四种模式对比
| 模式 | 何时连接源 | 何时断开 | 后订阅者看到什么 | 内存代价 |
|---|---|---|---|---|
refCount(n) | 累计 n 个订阅者 | 订阅者数量归零 | 订阅之后的数据 | 低 |
autoConnect(n) | 累计 n 个订阅者 | 不会因订阅者归零而断开 | 订阅之后的数据 | 低，但源可能空转 |
replay(n).autoConnect() | 累计 n 个订阅者（默认 1） | 不会因订阅者归零而断开 | 最近 n 个历史 + 新数据 | 中，受 n 限制 |
replay().autoConnect() | 累计 1 个订阅者 | 不会因订阅者归零而断开 | 全部历史 + 新数据 | 高，无界 |
cache() / cache(n) | 第一个订阅者 | 源完成后保留缓存 | 全部 / 最近 n 个缓存 | 高 / 有界 |
这张表里最容易记错的是两点：autoConnect(n) 的 n 只控制连接，不控制断开；refCount(n) 的 n 也只控制连接，断开阈值始终是 0。
refCount(n)：订阅归零就断开
先看一个同步源，确认「源完成后新订阅者会重新触发订阅」：
AtomicInteger starts = new AtomicInteger();
Flux<Integer> hot = Flux.range(1, 3)
.doOnSubscribe(s -> starts.incrementAndGet())
.publish()
.refCount(1);
List<Integer> a = new ArrayList<>();
List<Integer> b = new ArrayList<>();
hot.subscribe(a::add);
hot.subscribe(b::add);
System.out.println("A=" + a + " B=" + b + " starts=" + starts.get());
实测输出：
A=[1, 2, 3] B=[1, 2, 3] starts=2
A 订阅时连接，range 同步发射完并终止；B 订阅时连接已经结束，refCount 重新订阅源，所以 B 又拿到完整的 1、2、3。refCount 不是「只订阅一次」的缓存，源终止后它会把下一次订阅当作新一轮。
对无限源，可以看到归零断开：
AtomicInteger starts = new AtomicInteger();
Flux<Long> hot = Flux.interval(Duration.ofMillis(100))
.doOnSubscribe(s -> {
starts.incrementAndGet();
System.out.println("source subscribed");
})
.doOnCancel(() -> System.out.println("source cancelled"))
.publish()
.refCount(1);
List<Long> a = new ArrayList<>();
Disposable d1 = hot.subscribe(v -> { a.add(v); System.out.println("A: " + v); });
Thread.sleep(250);
d1.dispose();
Thread.sleep(150);
List<Long> b = new ArrayList<>();
Disposable d2 = hot.subscribe(v -> { b.add(v); System.out.println("B: " + v); });
Thread.sleep(250);
d2.dispose();
某次实测（100ms 间隔，250ms 窗口）：
source subscribed
A: 0
A: 1
source cancelled
source subscribed
B: 0
B: 1
source cancelled
refCount(2) 的连接条件是 2 个订阅者，但断开条件仍然是 0：从 2 个降到 1 个不会断开，只有最后一个订阅者取消才取消源。这个设计避免了「订阅者数在 n 附近抖动导致反复连接」的问题。
autoConnect(n)：连上就不放手
autoConnect(n) 同样等到 n 个订阅者才连接，但之后不跟踪订阅者数量：
AtomicInteger starts = new AtomicInteger();
Flux<Integer> hot = Flux.range(1, 3)
.doOnSubscribe(s -> starts.incrementAndGet())
.publish()
.autoConnect(2);
List<Integer> a = new ArrayList<>();
List<Integer> b = new ArrayList<>();
hot.subscribe(a::add); // 只有 1 个，等待
hot.subscribe(b::add); // 第 2 个到达，连接
System.out.println("A=" + a + " B=" + b + " starts=" + starts.get());
实测输出：
A=[1, 2, 3] B=[1, 2, 3] starts=1
连接后即使订阅者全部取消，源也不会停。用 interval 可以观察到这一点：
AtomicInteger emitted = new AtomicInteger();
Flux<Long> hot = Flux.interval(Duration.ofMillis(50))
.doOnNext(v -> emitted.incrementAndGet())
.publish()
.autoConnect(1);
Disposable d = hot.subscribe(v -> { });
Thread.sleep(220);
int before = emitted.get();
d.dispose();
Thread.sleep(300);
int after = emitted.get();
实测：
before dispose=4 after dispose=10
最后一个订阅者已经取消，interval 仍在继续发射，emitted 从 4 涨到 10。这就是 autoConnect 的资源泄漏风险：源会一直运行到进程结束或源自身终止。只有确定「源应该常驻」时才用它，例如进程级的共享事件总线。
replay()：给后订阅者补历史
replay() 返回 ConnectableFlux，连接后把历史元素缓存下来给后订阅者。无参版本缓存全部历史：
ConnectableFlux<Integer> replay = source.replay();
replay.connect();
replay(int n) 只保留最近 n 个：
AtomicReference<FluxSink<Integer>> ref = new AtomicReference<>();
ConnectableFlux<Integer> replay = Flux.<Integer>create(ref::set).replay(2);
replay.connect();
ref.get().next(1);
ref.get().next(2);
ref.get().next(3);
List<Integer> late = new ArrayList<>();
replay.subscribe(late::add);
ref.get().next(4);
ref.get().complete();
System.out.println("late=" + late);
实测输出：
late=[2, 3, 4]
前 3 个元素发出后，replay(2) 只保留最近 2 个（2、3）；后订阅者先收到这 2 个历史，再收到连接期间的 4。它不会把 1 补给它。
重载与语义：
| 重载 | 保留策略 |
|---|---|
replay() | 全部历史，等价于 replay(Integer.MAX_VALUE) |
replay(int n) | 最近 n 个 |
replay(Duration ttl) | ttl 内的全部元素 |
replay(int n, Duration ttl) | 同时受数量与时间限制 |
replay(0) | 不保留历史，退化为 publish() |
replay() 是有状态的：历史在内存里，和订阅者是否在线无关。对无限流使用 replay() 会让内存持续增长；replay(n) 给出数量上界，但不能限制速率，元素体积大时仍要按内存预算选 n。
replay().autoConnect() 与 replay().connect() 的区别只在连接时机：前者等第一个订阅者到达后连接，后者由你显式调用。两者都不会在订阅者归零时断开；需要自动停止时，用 replay().refCount(n)。
cache()：冷结果只算一次
cache() 不是 ConnectableFlux，没有 connect()。它在第一个订阅者出现时订阅源，并把结果缓存下来：
Flux<Integer> full = Flux.range(1, 5).cache();
Flux<Integer> lastTwo = Flux.range(1, 5).cache(2);
List<Integer> a = new ArrayList<>();
full.subscribe(a::add);
List<Integer> b = new ArrayList<>();
full.subscribe(b::add);
List<Integer> c = new ArrayList<>();
lastTwo.subscribe(c::add);
List<Integer> d = new ArrayList<>();
lastTwo.subscribe(d::add);
System.out.println("cache(): A=" + a + " B=" + b);
System.out.println("cache(2): A=" + c + " B=" + d);
实测输出：
cache(): A=[1, 2, 3, 4, 5] B=[1, 2, 3, 4, 5]
cache(2): A=[1, 2, 3, 4, 5] B=[4, 5]
两个要点：
cache(n)限制的是给后订阅者的历史条数，正在消费的第一个订阅者仍然拿到全部元素；- 缓存独立于订阅者存在。所有订阅者都取消后，
cache()仍保留已缓存的数据，直到cached引用本身被回收。
这与 replay().autoConnect() 的差别在连接语义：cache() 由第一个订阅者触发，源完成后缓存永久保留；replay().autoConnect() 在连接后即使没人订阅也继续运行，并把新元素加入历史。
Mono.cache 与失效
Mono.cache() 缓存的是终止信号（值或错误），不是「当前值」。失败也会被缓存：
A error: java.lang.IllegalStateException: boom
B error: java.lang.IllegalStateException: boom
attempts=1
需要基于时间或条件失效时，Mono 提供了额外的重载：
Mono<Config> config = loadConfig()
.cache(Duration.ofMinutes(10)); // TTL
Mono<User> user = fetchUser(id)
.cacheInvalidateIf(u -> u.isExpired()); // 值满足条件就失效
Mono<Price> price = fetchPrice(sku)
.cacheInvalidateWhen(p -> priceFeed.nextChange(sku)); // 外部信号触发失效
注意 cacheInvalidateIf / cacheInvalidateWhen 只在 Mono 上，Flux 没有对应方法；Flux 只有 cache()、cache(int)、cache(Duration)、cache(int, Duration) 及其带 Scheduler 的重载。
选择指南
| 需求 | 选择 | 原因 |
|---|---|---|
| 自动启停，省资源 | refCount(1) | 最后一个订阅者取消时断开 |
| 源应常驻，订阅者来去自由 | autoConnect(1) | 连接后不跟踪订阅者 |
| 后订阅者需要完整历史 | replay() / cache() | 但有内存上界问题 |
| 后订阅者只需要最近 n 个 | replay(n) / cache(n) | 内存有界，语义明确 |
| 冷结果只计算一次 | cache() | 第一个订阅者触发，后续复用 |
| 需要 TTL 或条件失效 | Mono.cache(ttl) / cacheInvalidate* | Flux 没有等价的失效 API |
决策树
需要把同一份数据共享给多个订阅者？
├── 否 → Cold Publisher
└── 是
├── 后订阅者需要历史？
│ ├── 是 → 历史有界？
│ │ ├── 是 → replay(n).autoConnect() / cache(n)
│ │ └── 否 → replay().autoConnect() / cache()（评估内存）
│ └── 否
│ ├── 没有订阅者时要停？ → refCount(n)
│ └── 源应一直运行？ → autoConnect(n)
└── 只需要复用一次计算结果？ → cache()
小结
| 模式 | 一句话 |
|---|---|
refCount(n) | n 个订阅者连接，归零断开，源终止后新订阅会重新连接 |
autoConnect(n) | n 个订阅者连接，之后永不因订阅者归零而断开 |
replay(n) | 保留最近 n 个历史，后订阅者补发 |
cache() | 第一个订阅者触发一次执行，结果长期保留 |
cache(n) | 只给后订阅者回放最近 n 个 |
选择之前先回答两个问题：后订阅者是否必须看到历史，以及没有订阅者时源是否应该继续跑。这两个答案基本决定了用哪一组操作符；内存只是随后要核算的代价。
阅读导航
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。