Share() 用法详解：将 ConnectableFlux 转换为热流

本文深入剖析了 Reactor 中 `share()` 操作符的本质与使用陷阱：它并非简单的“变热”开关，而是将已连接的 `ConnectableFlux` 转换为热流的便捷封装（等价于 `replay(1).refCount(1)`），其成败关键在于上游是否真正完成连接——若在调用 `share()` 前冷流已被提前订阅并终止（如误用 `take(5)` 后独立触发一次订阅），后续所有订阅都将一无所获；文章通过典型反例揭示根源，系统对比 `autoConnect(n)`、显式 `connect()` 和 `share()` 三种方案的适用场景与生命周期控制逻辑，并强调缓存容量设定、订阅时序协同及日志调试等实战要点，助你真正掌握热流构建的核心原则——连接时机决定一切。

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 热流的核心前提。

以上就是《Share() 用法详解：将 ConnectableFlux 转换为热流》的详细内容，更多关于的资料请关注golang学习网公众号！