如何正确使用 share() 将 ConnectableFlux 转换为热流-java教程-PHP中文网

如何正确使用 share() 将 ConnectableFlux 转换为热流

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()` 完成而终止！</long>

⚠️ 关键陷阱：

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, ...</long></long>

? 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 才真正启动</long>

✅ 方案三：直接使用 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, ...</long>