当前位置：首页 > news >正文

Reactor ConnectableFlux支持多订阅者

news 2025/7/2 18:07:37

在 Reactor 中，ConnectableFlux 是一种用于处理响应式流的机制，它允许你控制何时开始订阅和数据生成。通常情况下，订阅者（subscriber）在订阅时会立即开始接收数据，但有时你可能希望多个订阅者“会面”（rendezvous）之后再触发订阅和数据生成。这就是 ConnectableFlux 的用途。

1. ConnectableFlux 的主要模式

Flux API 提供了两种主要的模式来返回 ConnectableFlux：publish 和 replay。

publish：
publish 会动态地尝试满足各个订阅者的需求（即背压），并通过将这些请求转发到源来实现。如果任何订阅者的挂起需求为 0，publish 会暂停对源的请求。
例如，你可以使用 publish() 方法将一个冷发布者（cold publisher）转换为热发布者（hot publisher），从而允许多个订阅者共享同一个数据源。
replay：
replay 会缓存第一次订阅看到的数据，并在达到可配置的限制（如时间和缓冲区大小）后，将数据重放给后续的订阅者。
例如，你可以使用 replay(2) 来缓存最近的 2 个数据点，并在新订阅者到来时重放这些数据。

2. ConnectableFlux 的管理方法

ConnectableFlux 提供了多种方法来管理订阅和源的连接：

connect()：
你可以手动调用 connect() 方法，当达到足够的订阅数时，触发对上游源的订阅。例如：
```
ConnectableFlux<String> connectableFlux = Flux.just("A", "B", "C").publish();
connectableFlux.connect();
```
在调用 connect() 之前，connectableFlux 不会开始发送数据。
autoConnect(n)：
autoConnect(n) 可以自动执行与 connect() 类似的操作，当有 n 个订阅者订阅时，自动触发对源的订阅。例如：
```
Flux<String> flux = Flux.just("A", "B", "C");
ConnectableFlux<String> autoConnectFlux = flux.publish().autoConnect(2);
```
这意味着当有 2 个或更多订阅者订阅时，autoConnectFlux 会自动开始发送数据。
refCount(n)：
refCount(n) 不仅可以自动跟踪传入的订阅，还可以检测订阅是否被取消。如果订阅者数量不足，refCount 会断开与源的连接，直到有新的订阅者出现。例如：
```
Flux<String> flux = Flux.just("A", "B", "C");
ConnectableFlux<String> refCountFlux = flux.publish().refCount(2);
```
这意味着当有 2 个订阅者订阅时，refCountFlux 会自动开始发送数据；当所有订阅者取消订阅后，refCountFlux 会断开连接。
refCount(int, Duration)：
refCount(int, Duration) 增加了一个“宽限期”（grace period），即在订阅者数量低于阈值时，等待指定的时间后再断开连接。例如：
```
Flux<String> flux = Flux.just("A", "B", "C");
ConnectableFlux<String> refCountWithGrace = flux.publish().refCount(2, Duration.ofSeconds(10));
```
这意味着在订阅者数量低于 2 时，refCountWithGrace 会等待 10 秒，看看是否有新的订阅者出现。

3. 应用场景

ConnectableFlux 适用于需要多个订阅者“会面”后再触发订阅和数据生成的场景。例如：

实时数据推送：在实时数据推送中，你可能希望多个客户端在连接到服务器后才开始接收数据。使用 ConnectableFlux 可以确保所有客户端都准备好后再开始发送数据。
分布式系统：在分布式系统中，你可能希望多个节点在协调一致后再触发数据生成。使用 ConnectableFlux 可以确保所有节点都准备好后再开始处理数据。
IoT 数据可视化：在 IoT 数据可视化中，你可能希望多个设备在连接到服务器后才开始发送数据。使用 ConnectableFlux 可以确保所有设备都准备好后再开始处理数据。

4. 相关案例

展示如何使用 ConnectableFlux 实现流的多订阅和延迟连接

package org.example;import reactor.core.publisher.ConnectableFlux;
import reactor.core.publisher.Flux;/*** Main0011 类演示了如何使用 Reactor 创建和操作 Flux 流* 该类展示了如何使用 ConnectableFlux 实现流的多订阅和延迟连接*/
public class Main0011 {/*** 主函数展示了如何创建一个 Flux 流并使用 ConnectableFlux 进行订阅和连接操作** @param args 命令行参数* @throws InterruptedException 线程睡眠时可能抛出的异常*/public static void main(String[] args) throws InterruptedException {// 创建一个 Flux 源，产生 1 到 3 的整数序列，并在订阅时打印消息Flux<Integer> source = Flux.range(1, 3).doOnSubscribe(s -> System.out.println("subscribed to source"));// 将 Flux 源转换为 ConnectableFlux，以便进行延迟连接和多订阅ConnectableFlux<Integer> co = source.publish();// 订阅 ConnectableFlux，此时不会开始产生数据co.subscribe(System.out::println, e -> {}, () -> {});// 再次订阅，演示多个订阅者co.subscribe(System.out::println, e -> {}, () -> {});// 打印消息，表明订阅已完成，但数据流尚未开始System.out.println("done subscribing");// 线程睡眠，模拟在连接前的准备或其他操作Thread.sleep(500);// 打印消息，表明即将连接数据流System.out.println("will now connect");// 连接数据流，使数据开始流动到所有已订阅的消费者co.connect();}
}

演示Reactor库中Flux的自动连接（autoConnect）功能

package org.example;import reactor.core.publisher.Flux;/*** 该类用于演示Reactor库中Flux的自动连接（autoConnect）功能* 它展示了如何使用autoConnect方法在多个订阅者之间共享一个数据流，* 并在达到指定的订阅者数量后自动开始数据流的发布*/
public class Main0012 {/*** 主函数，用于演示autoConnect的使用** @param args 命令行参数* @throws InterruptedException 当线程因中断策略被中断时抛出此异常*/public static void main(String[] args) throws InterruptedException {// 创建一个Flux数据源，范围从1到3，同时在订阅时打印消息Flux<Integer> source = Flux.range(1, 3).doOnSubscribe(s -> System.out.println("subscribed to source"));// 使用autoConnect方法使数据源在有两个订阅者时自动连接Flux<Integer> autoCo = source.publish().autoConnect(2);// 第一个订阅者订阅数据流，并在接收到数据时打印出来autoCo.subscribe(System.out::println, e -> {}, () -> {});System.out.println("subscribed first");// 暂停500毫秒以模拟时间流逝Thread.sleep(500);System.out.println("subscribing second");// 第二个订阅者订阅数据流，此时达到autoConnect设定的条件，数据流开始发布autoCo.subscribe(System.out::println, e -> {}, () -> {});}
}

5. 总结

ConnectableFlux 是 Reactor 中用于处理响应式流的机制，它允许你控制何时开始订阅和数据生成。通过 publish 和 replay 模式，你可以实现多个订阅者“会面”后再触发订阅和数据生成。通过 connect()、autoConnect(n)、refCount(n) 和 refCount(int, Duration) 方法，你可以灵活地管理订阅和源的连接。

查看全文

http://www.lqws.cn/news/603001.html