share() 本身不触发订阅，需确保上游已连接（如通过 autoConnect(n) 或显式 connect()），否则后续订阅将收不到数据；关键在于连接时机与订阅顺序的协同。
`share()` 本身不触发订阅，需确保上游已连接（如通过 `autoconnect(n)` 或显式 `connect()`），否则后续订阅将收不到数据；关键在于连接时机与订阅顺序的协同。
在 Reactor 中，share() 是一个便捷操作符，用于将普通 Flux 转换为热流（hot publisher），其底层等价于 replay(1).refCount(1)。但值得注意的是：share() 仅作用于已“连接”的 ConnectableFlux；若上游未真正启动（即未调用 connect() 或未满足 autoConnect(n) 的订阅阈值），则即使链式调用了 share()，也不会触发数据发射。
? 问题根源分析
原始代码中：
Flux<Long> flux = Flux.interval(Duration.ofSeconds(1)).take(5); flux.subscribe(); // ❌ 该订阅作用于原始 cold flux，且立即完成（因 take(5) 发完即结束） flux = flux.replay().autoConnect(2).share(); // ✅ 创建 hot 流，但 autoConnect(2) 要求 2 个订阅才触发连接 flux.subscribe(...); // 第 1 个订阅 → 未达阈值，不连接，不发射 sleep(2000); flux.subscribe(...); // 第 2 个订阅 → 此时才连接，但此时原始 interval 已因前一个 `flux.subscribe()` 完成而终止！
⚠️ 关键陷阱：
- Flux.interval(...).take(5) 是冷流（cold），每次订阅都从头开始计时并发射 0~4；
- 但第一个 flux.subscribe() 独立触发了一次完整订阅，导致该冷流立即执行、发射完毕并终止；
- 后续 replay().autoConnect(2).share() 所基于的 flux，其源头已被“消耗”——replay() 缓存的是空序列（因上游早已结束），所以无论多少订阅，都无数据可重放。
✅ 正确做法：延迟连接，统一管理生命周期
✅ 方案一：避免提前订阅，让 autoConnect(n) 主导连接时机
Flux<Long> source = Flux.interval(Duration.ofSeconds(1))
.take(5)
.doOnNext(v -> System.out.println("emitting: " + v));
// 构建 connectable hot 流，但不提前订阅
Flux<Long> hotFlux = source
.replay(5) // 缓存最多 5 个元素（覆盖全部）
.autoConnect(2); // 等待 2 个订阅后自动连接
// 此时 hotFlux 还未连接 → 无数据发射
hotFlux.subscribe(v -> System.out.println("first: " + v));
Thread.sleep(1500);
hotFlux.subscribe(v -> System.out.println("second: " + v));
// 输出：first: 0, first: 1, second: 1, first: 2, second: 2, ...? replay(5) 显式指定缓存容量，避免默认 replay()（缓存无限）在长流中引发 OOM；autoConnect(2) 在第 2 个订阅到达时触发连接，所有订阅者共享同一数据流。
✅ 方案二：显式 connect() 控制时机（更清晰）
ConnectableFlux<Long> connectable = Flux.interval(Duration.ofSeconds(1))
.take(5)
.replay(5);
// 订阅者先注册
connectable.subscribe(v -> System.out.println("first: " + v));
connectable.subscribe(v -> System.out.println("second: " + v));
// 手动触发连接 → 数据开始发射
connectable.connect(); // ✅ 关键：此时 interval 才真正启动✅ 方案三：直接使用 share()（推荐简洁场景）
Flux<Long> hotFlux = Flux.interval(Duration.ofSeconds(1))
.take(5)
.share(); // 等价于 replay(1).refCount(1)，首次订阅即连接
hotFlux.subscribe(v -> System.out.println("first: " + v));
Thread.sleep(1500);
hotFlux.subscribe(v -> System.out.println("second: " + v));
// 输出：first: 0, first: 1, second: 1, first: 2, second: 2, ...✅ share() 内部使用 refCount(1)，意味着首个订阅即触发连接，最后一个取消订阅即断开，适合轻量级热共享。
⚠️ 注意事项总结
- replay().autoConnect(n) 的 n 是并发订阅数阈值，不是“最大订阅数”；
- share() 不是魔法开关——它依赖上游是否可连接且已连接；
- 避免对冷流做多次独立订阅（尤其含 take()/timeout() 等终止操作），否则会提前耗尽源头；
- 日志辅助调试：善用 doOnSubscribe, doOnNext, doOnComplete 观察生命周期事件；
- 生产环境建议显式指定 replay(size) 容量，防止内存泄漏。
掌握连接时机（connect timing）与订阅顺序（subscription order），是驾驭 Reactor 热流的核心前提。
