Cold Publisher 的定义
Cold Publisher：数据流与订阅绑定。每次 subscribe() 都会启动一个新的独立序列，数据从源头重新产生。
Flux<Integer> cold = Flux.range(1, 3);
cold.subscribe(i -> System.out.println("A: " + i));
cold.subscribe(i -> System.out.println("B: " + i));
实测输出：
A: 1
A: 2
A: 3
B: 1
B: 2
B: 3
两次订阅独立运行，各自消费完整的 1、2、3 序列。第二次订阅不是「从上次的位置继续」，而是从头再来。
特征
| 特征 | 说明 |
|---|---|
| 订阅驱动 | 不 subscribe 就不启动 |
| 独立序列 | 每次订阅有独立的订阅状态与计数器 |
| 可重放 | 同一数据源可以被重复消费 |
| 源头耦合 | 每次订阅重新访问数据源（查询、读文件、发请求） |
每次订阅都会重新执行副作用
Cold 的关键不是「值相同」，而是「生成值的动作会重新执行」：
AtomicInteger calls = new AtomicInteger();
Mono<Integer> mono = Mono.fromCallable(() -> {
System.out.println("fetch " + calls.incrementAndGet());
return 42;
});
mono.subscribe(v -> System.out.println("A: " + v));
mono.subscribe(v -> System.out.println("B: " + v));
实测输出：
fetch 1
A: 42
fetch 2
B: 42
重点是 fetch 出现了两次。如果这里换成 HTTP 调用或数据库查询，就是两次真实的远程请求。
这里有一个常被混淆的对照：
Mono<Integer> eager = Mono.just(expensiveCall()); // 装配时就执行
Mono<Integer> lazy = Mono.fromCallable(() -> expensiveCall()); // 每次订阅执行
Mono.just(...) 的参数在调用方法时就已经求值，代价发生在装配阶段；fromCallable 把调用包进 Supplier，订阅时才执行。想利用 Cold 的「每次订阅重新获取」，必须用后者。
典型 Cold 操作符
| 操作符 | 数据源 | 每次订阅的行为 |
|---|---|---|
Flux.just / Mono.just | 内存常量 | 重放相同的值（值本身已在装配时确定） |
Flux.range | 计数器 | 重新计数 |
Flux.fromIterable | 集合 | 重新遍历 |
Flux.create | 自定义 | 重新执行 create 的 consumer |
Mono.fromCallable | 调用 | 重新调用 |
Mono.defer | 延迟创建 | 重新执行 Supplier |
Flux.using | 资源 | 每次订阅创建并释放资源 |
Mono.defer 和 fromCallable 的区别在于延迟的层次：defer 延迟的是「创建哪个 Publisher」，适合每次订阅都要重新决定数据源的场景；fromCallable 延迟的是「执行哪个调用」，适合数据源本身已经确定。
cache()：把结果留给后续订阅
如果意图是「只调用一次，结果共享给所有订阅者」，cache() 是最直接的手段：
AtomicInteger fetches = new AtomicInteger();
Mono<Integer> cached = Mono.fromCallable(() -> {
System.out.println("fetch " + fetches.incrementAndGet());
return 42;
}).cache();
cached.subscribe(v -> System.out.println("A: " + v));
cached.subscribe(v -> System.out.println("B: " + v));
实测输出：
fetch 1
A: 42
B: 42
第一次订阅触发实际调用，第二次直接拿到缓存值。Mono.cache() 缓存的是终止信号，成功和失败都缓存：
AtomicInteger attempts = new AtomicInteger();
Mono<Integer> failing = Mono.<Integer>fromCallable(() -> {
attempts.incrementAndGet();
throw new IllegalStateException("boom");
}).cache();
failing.subscribe(v -> {}, e -> System.out.println("A error: " + e));
failing.subscribe(v -> {}, e -> System.out.println("B error: " + e));
System.out.println("attempts=" + attempts.get());
实测输出：
A error: java.lang.IllegalStateException: boom
B error: java.lang.IllegalStateException: boom
attempts=1
第二次订阅不会重试，只会立刻收到同一个错误。需要失败后重试或按条件失效时，用 Mono.cache(ttl)、Mono.cacheInvalidateIf(...)、Mono.cacheInvalidateWhen(...) 这些重载，而不是 cache()。
Flux.cache 的重载与内存
Flux.cache() 在第一次订阅时把整个序列缓存下来，后订阅者从头回放。cache(int) 只保留最近 n 个元素给后订阅者：
Flux<Integer> full = Flux.range(1, 5).cache();
Flux<Integer> lastTwo = Flux.range(1, 5).cache(2);
两个序列都在第一次订阅时完整执行，第二次订阅的结果是：
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(2) 只是限制给后订阅者的历史长度，不限制正在订阅的消费者。它对无限流尤其重要：
// 危险：无界缓存 + 无限流
Flux.interval(Duration.ofMillis(100))
.cache()
.subscribe();
cache() 会保留所有已发出的元素，长时间运行必然 OOM。cache(int)、cache(Duration)、cache(int, Duration) 可以给出上界；但上界内的元素仍然常驻内存，容量要按业务峰值估算，而不是「应该不会太多」。
另外，cache() 的订阅发生在第一个订阅者出现时，不是装配时。只创建 cached 而不订阅，源不会执行。
Cold vs Hot
| 维度 | Cold | Hot |
|---|---|---|
| 启动时机 | subscribe() 时 | connect() / autoConnect(n) / 第一个订阅者 |
| 订阅独立性 | 每次独立 | 共享同一条源流 |
| 订阅前的数据 | 订阅后完整产生 | 已产生但无人接收的部分丢失 |
| 内存 | 按需产生，无共享缓冲 | 取决于是否 replay，以及慢订阅者的队列 |
Cold 的内存不能用一个「订阅数 × 元素大小」的公式概括：元素是按需生成的，真正的占用来自在途元素、操作符的预取队列和你的缓冲策略。Flux.range(1, 1000) 两次订阅消耗的是两次遍历的临时状态，不会把 2000 个整数同时留在内存里。
何时选择 Cold
适合：
- 每个订阅者都需要完整、独立的结果（按用户 ID 查询、读取文件）；
- 数据源可以低成本重复访问；
- 不需要在订阅者之间共享状态。
不适合：
- 实时行情、传感器、聊天这类「只关心订阅之后」的流——应该用 Hot；
- 每次订阅都触发昂贵远程调用，且多个订阅者需要同一份结果——用
cache()或有界replay。
阅读导航
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。