Hot Publisher 的定义
Hot Publisher：数据流独立于订阅存在。元素一旦产生，立即广播给当前活跃的订阅者；后订阅的（late subscriber）只能收到订阅之后的元素。
判断 Cold 和 Hot，不看「是不是同一个对象」，而看数据是在订阅时生成，还是在订阅之外生成。Flux.range 每次订阅重新计数，是 Cold；publish() 连上源之后，源持续发射，新订阅者只能搭上后半程，是 Hot。
publish() + connect()：先启动，后订阅
publish() 把 Cold 源转换成 ConnectableFlux。它不会在 subscribe() 时自动连接，需要显式 connect()：
import reactor.core.publisher.ConnectableFlux;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicReference;
public class PublishDemo {
public static void main(String[] args) {
AtomicReference<FluxSink<Integer>> ref = new AtomicReference<>();
ConnectableFlux<Integer> hot = Flux.<Integer>create(ref::set).publish();
hot.connect(); // 连接源，但还没有订阅者
List<Integer> a = new ArrayList<>();
List<Integer> b = new ArrayList<>();
ref.get().next(1); // 此时没有订阅者，1 被丢弃
hot.subscribe(a::add);
ref.get().next(2);
hot.subscribe(b::add);
ref.get().next(3);
ref.get().complete();
System.out.println("A=" + a + " B=" + b);
}
}
实测输出：
A=[2, 3] B=[3]
A 在 1 之后订阅，所以错过 1，拿到 2、3；B 在 2 之后订阅，只拿到 3。connect() 之前或两次订阅之间发出的元素不会为后来者保留——这就是 Hot 的「丢失历史」。
connect() 返回 Disposable，调用 dispose() 会断开与源的连接。重复 connect() 不会重复订阅源；ConnectableFlux 只维护一条上游连接。
autoConnect(n)：凑够 n 个订阅者再连接
autoConnect(n) 省掉手动 connect()，但连接的触发条件是「累计 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);
System.out.println("first: A=" + a + " starts=" + starts.get());
hot.subscribe(b::add);
System.out.println("second: A=" + a + " B=" + b + " starts=" + starts.get());
实测输出：
first: A=[] starts=0
second: A=[1, 2, 3] B=[1, 2, 3] starts=1
第一个订阅者只是排队等待；第二个到达时触发连接，range 同步发射完 1、2、3，两个订阅者都收到完整序列，源只被订阅了一次。
autoConnect(n) 的 n 只决定何时连接，不决定何时断开。连接建立后，即使订阅者全部取消，源也会继续运行——对 interval 这类无限源就是持续占用调度器资源。需要「没有订阅者就停」时用 refCount。
refCount(n)：按订阅数自动连接与断开
refCount(n) 在累计 n 个订阅者时连接，在订阅者数量归零时断开。这里的两个数字容易混淆：n 是连接阈值，断开阈值永远是 0。从 n 降到 n-1 不会断开。
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); // 连接，range 发射完并完成
hot.subscribe(b::add); // 源已经终止，refCount 重新连接
System.out.println("A=" + a + " B=" + b + " starts=" + starts.get());
实测输出：
A=[1, 2, 3] B=[1, 2, 3] starts=2
A 结束后连接已经终止；B 订阅时 refCount 重新订阅源，所以 B 又能拿到完整的 1、2、3，starts 变成 2。
对无限源，可以观察到「归零断开」：
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
B 重新从 0 开始，starts=2。interval 是 Cold 源，每次连接都重新计时；如果源本身是 Hot（例如消息队列的消费者），重新连接意味着重新注册消费者，是否能补上断开期间的数据取决于源，而不是 refCount。
share()：refCount(1) 的简写
Flux<Long> hot = Flux.interval(Duration.ofMillis(100)).share();
share() 在 3.8.8 里的实现就是 new FluxRefCount<>(new FluxPublish<>(source, 256, ...), 1)，等价于 publish().refCount(1)。需要自定义预取量时用 publish(n).refCount(1)，share() 不接受参数。
背压：publish 会拖住源，而不是无限缓冲
一个常见的说法是「Hot Publisher 没有背压，慢消费者会导致内存无限增长」。在 Reactor 里这个说法不准确。
publish() 的 drain 循环在源线程上同步调用每个订阅者的 onNext。只要有一个订阅者在 onNext 里阻塞，整条广播就被拖住，源也无法继续发射：
Flux<Long> hot = Flux.interval(Duration.ofMillis(1))
.publish()
.autoConnect(1);
hot.subscribe(v -> {
received.incrementAndGet();
Thread.sleep(100); // 慢消费者
});
实测 1.5 秒内 received=14 errors=0。没有队列暴涨，也没有 OOM，代价是所有订阅者一起被最慢的那个拖慢（head-of-line blocking）。元素在 publish 的预取队列里最多堆积 prefetch 个（默认 256），源随后就被反压住。
如果订阅者完全不请求，情况更直接：publish 只向上游请求预取量，drain 没有需求就不补请求，时间源自己会先失败。实测：
error: reactor.core.Exceptions$OverflowException: Could not emit tick 256 due to lack of requests
(interval doesn't support small downstream requests that replenish slower than the ticks)
要真正解耦慢消费者，必须让消费不再阻塞广播线程，并且接受「数据会被缓冲或丢弃」。publishOn 只能把阻塞点移到另一个线程：如果 slowConsumer 在 onNext 里继续阻塞，publishOn 的工作线程也会被占住，它自己的队列填满后仍然会反压回 publish。把 onBackpressureDrop 直接放在 publish() 之前也不能解决问题——publish 会向上游请求默认 256 个，Drop 看到的一直是有需求的，实测丢弃计数为 0。
真正有损的广播需要生产者主动丢弃，而不是让消费者反压。Sinks.many().multicast().directBestEffort() 会在订阅者跟不上时直接丢元素；Flux.create(sink, FluxSink.OverflowStrategy.DROP) 也能表达同样的策略。这些属于第 9 章的内容。这里先记住结论：publish() 的行为是「拖慢源」，不是「无限缓冲」；想要「不拖慢源、允许丢」，要换的是源与缓冲模型，而不是在链中间随手加一个 onBackpressureDrop。
Hot Publisher 与消息队列
| 特性 | Hot Publisher | 消息队列 |
|---|---|---|
| 持久化 | 无 | 有 |
| 消费者组 | 广播给所有订阅者 | 可分组、可竞争消费 |
| 重放 | replay(n) 在内存中有限重放 | 可配置保留与重放 |
| 顺序 | 单条连接内有序 | 分区内有序 |
| 适用 | 实时状态、进程内广播 | 可靠传输、跨进程解耦 |
Hot Publisher 不解决「消息不能丢」的问题。需要可靠投递时，消息队列或持久化日志仍然是必要的；replay 只是进程内的短期缓冲，进程退出即消失。
小结
| 特性 | Cold | Hot |
|---|---|---|
| 订阅前数据 | 订阅后完整产生 | 已产生的部分丢失 |
| 启动时机 | subscribe | connect / autoConnect(n) / refCount(n) |
| 数据独立性 | 每次订阅独立 | 共享同一条源流 |
| 停止条件 | 订阅结束或取消 | refCount 归零、显式 dispose 或源终止 |
| 内存 | 按需产生 | 取决于 replay 与慢订阅者队列 |
阅读导航
下一篇：8.3 refCount、autoConnect 与 cache
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。