用 ConnectableFlux 向多个订阅者广播

Doubao 中英对照 Broadcasting to Multiple Subscribers with ConnectableFlux Broadcasting to Multiple Subscribers with ConnectableFlux

有时，您可能不只是想把某些处理延迟到某一个订阅者的订阅时刻，而是希望让几个订阅者先集合到一起，然后再触发对数据源的订阅和数据生成。

这正是 ConnectableFlux 的用途。Flux API 中有两种主要模式会返回 ConnectableFlux：publish 和 replay。

• publish 会在背压方面动态地尽量兼顾各个订阅者的需求，做法是把这些请求转发给数据源。最值得注意的是：只要有任何一个订阅者的待处理需求为 0，publish 就会暂停向数据源请求数据。

• replay 会缓冲通过第一次订阅所见到的数据，上限可配置（在时间和缓冲大小两方面）。它会把这些数据重放给后续的订阅者。

ConnectableFlux 提供了一些额外方法，用来管理“下游的订阅”与“对原始数据源的订阅”之间的关系。这些额外方法包括：

• connect()：当您对该 Flux 的订阅达到足够数量后，可以手动调用它。这会触发对上游数据源的订阅。

• autoConnect(n)：在已有 n 次订阅后自动完成同样的工作。

• refCount(n)：不仅自动跟踪陆续到来的订阅，还能检测这些订阅在何时被取消。当跟踪到的订阅者不足时，数据源会被“断开连接”；此后若再有新的订阅者出现，就会重新对数据源发起一次订阅。

• refCount(int, Duration)：增加了一个“宽限期”。当被跟踪的订阅者数量过少时，它会先等待这个 Duration 再断开数据源连接，从而有可能让足够多的新订阅者赶在断开前进来、再次越过连接阈值。

请看下面这个示例：

Flux<Integer> source = Flux.range(1, 3)
                           .doOnSubscribe(s -> System.out.println("subscribed to source"));

ConnectableFlux<Integer> co = source.publish();

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();

上例代码产生如下输出：

done subscribing
will now connect
subscribed to source
1
1
2
2
3
3

下面这段代码使用了 autoConnect：

Flux<Integer> source = Flux.range(1, 3)
                           .doOnSubscribe(s -> System.out.println("subscribed to source"));

Flux<Integer> autoCo = source.publish().autoConnect(2);

autoCo.subscribe(System.out::println, e -> {}, () -> {});
System.out.println("subscribed first");
Thread.sleep(500);
System.out.println("subscribing second");
autoCo.subscribe(System.out::println, e -> {}, () -> {});

上例代码产生如下输出：

subscribed first
subscribing second
subscribed to source
1
1
2
2
3
3