java

10.2 ConnectableObservable 与 connect

对比 RxJava 的 ConnectableObservable 与 Reactor 的 ConnectableFlux：publish/connect 的延迟连接、refCount 在订阅归零时断开并在下次订阅时重新订阅源、autoConnect(n) 只在凑够 n 个订阅者时连接一次。实测 RxJava refCount 重连后 starts=2、interval 从 0 重新计时，autoConnect(2) 在只有一个订阅者时完全不启动；并纠正「Reactor 的 connect() 没有返回值」的错误说法。示例基于 RxJava 3.1.12 与 reactor-core 3.8.8。

Leeting Yan 4 分钟阅读 1728 字

同一个概念，两套 API

ConnectableObservable 和 ConnectableFlux 都是「可手动启动的 Hot Publisher」：

  • 订阅者可以先挂上去，但源不会因此启动；
  • 调用 connect() 才真正订阅上游；
  • 多个下游共享同一条上游连接。

差异主要在方法签名和几个边界行为，语义本身几乎一一对应。

publish() + connect()

RxJava：

import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.observables.ConnectableObservable;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

AtomicInteger ticks = new AtomicInteger();

ConnectableObservable<Long> hot = Observable.interval(50, TimeUnit.MILLISECONDS)
    .doOnNext(v -> ticks.incrementAndGet())
    .publish();

hot.subscribe(v -> System.out.println("A: " + v));

Disposable connection = hot.connect();   // 返回 Disposable
Thread.sleep(180);
int before = ticks.get();
connection.dispose();                    // 断开上游
Thread.sleep(200);
System.out.println("before dispose=" + before + " after dispose=" + ticks.get());

实测：

before dispose=3 after dispose=3

connect() 返回的 Disposable 控制的是上游连接，不是某个订阅者。调用 dispose() 后，interval 停止，订阅者仍在但收不到新元素。

Reactor 的写法相同，返回值也一样：

ConnectableFlux<Long> hot = Flux.interval(Duration.ofMillis(50)).publish();

hot.subscribe(v -> System.out.println("A: " + v));
Disposable connection = hot.connect();
connection.dispose();

这里要纠正一个常见错误说法：Reactor 的 connect() 是有返回值的，签名是 Disposable connect()；另有一个 connect(Consumer<? super Disposable>) 重载，把断开句柄交给回调。两者都存在，不是「只能传 consumer」。

refCount()：归零断开，再订阅重连

refCount() 把「连接」和「订阅者数量」绑定：第一个订阅者到达时连接，最后一个订阅者取消时断开。

AtomicInteger starts = new AtomicInteger();

Observable<Long> hot = Observable.interval(100, TimeUnit.MILLISECONDS)
    .doOnSubscribe(d -> {
        starts.incrementAndGet();
        System.out.println("source subscribed");
    })
    .doOnDispose(() -> System.out.println("source disposed"))
    .publish()
    .refCount();

List<Long> a = new ArrayList<>();
Disposable d1 = hot.subscribe(v -> { a.add(v); System.out.println("A: " + v); });
Thread.sleep(250);
d1.dispose();

Thread.sleep(150);

List<Long> b = new ArrayList<>();
Disposable d2 = hot.subscribe(v -> { b.add(v); System.out.println("B: " + v); });
Thread.sleep(250);
d2.dispose();

System.out.println("starts=" + starts.get() + " A=" + a + " B=" + b);

某次实测：

source subscribed
A: 0
A: 1
source disposed
source subscribed
B: 0
B: 1
source disposed
starts=2 A=[0, 1] B=[0, 1]

关键点：第二次订阅不是「接着上次的位置」，而是重新订阅源，interval 从 0 重新计时。refCount 不缓存数据，它只管理连接生命周期。

这与 Reactor 的行为完全一致。Reactor 的同一段逻辑：

Flux<Long> hot = Flux.interval(Duration.ofMillis(100))
    .doOnSubscribe(s -> starts.incrementAndGet())
    .doOnCancel(() -> System.out.println("source cancelled"))
    .publish()
    .refCount();

源完成后再次订阅

对有限的同步源，同样会重新订阅：

AtomicInteger starts2 = new AtomicInteger();

Observable<Integer> hot = Observable.range(1, 3)
    .doOnSubscribe(d -> starts2.incrementAndGet())
    .publish()
    .refCount();

List<Integer> x = new ArrayList<>();
List<Integer> y = new ArrayList<>();
hot.subscribe(x::add);
hot.subscribe(y::add);

System.out.println("X=" + x + " Y=" + y + " starts=" + starts2.get());

实测：

X=[1, 2, 3] Y=[1, 2, 3] starts=2

所以那个流传的说法——「RxJava 的 refCount 重新订阅时不会复用旧数据，而 Reactor 会」——是错的。两个库的 refCount 都是「重新订阅源」，都不复用旧数据。需要复用结果时用 replay() 或 cache()，与 refCount 无关。

autoConnect(n)：只连接，不跟踪订阅者

autoConnect(n) 在累计到 n 个订阅者时连接一次，之后不再关心订阅者数量：

AtomicInteger starts3 = new AtomicInteger();

Observable<Long> hot = Observable.interval(100, TimeUnit.MILLISECONDS)
    .doOnSubscribe(d -> starts3.incrementAndGet())
    .publish()
    .autoConnect(2);

List<Long> f = new ArrayList<>();
List<Long> s = new ArrayList<>();

hot.subscribe(f::add);
Thread.sleep(250);
System.out.println("after first: F=" + f + " starts=" + starts3.get());

hot.subscribe(s::add);
Thread.sleep(250);
System.out.println("after second: F=" + f + " S=" + s + " starts=" + starts3.get());

实测：

after first: F=[] starts=0
after second: F=[0, 1] S=[0, 1] starts=1

只有一个订阅者时源根本没有启动；第二个订阅者到达后连接一次，两个订阅者都从 0 开始收。之后再有人订阅或取消，源都不会重新连接或断开。

autoConnect() 无参版本等价于 autoConnect(1)。

延迟断开的 refCount

refCount() 在订阅者归零时立即断开。如果希望「短暂空窗也保持连接」，两个库都提供了带超时的重载：

// RxJava：最后一个订阅者取消后，等待 100ms 再断开
Observable<Long> hot = source.publish().refCount(1, 100, TimeUnit.MILLISECONDS);

// Reactor：同样等待 100ms
Flux<Long> hot = source.publish().refCount(1, Duration.ofMillis(100));

这个重载适合「订阅者会短暂重连」的场景，例如页面切换、请求重试；代价是空窗期内源仍在运行。

API 对照

功能RxJavaReactor
转成可连接流publish()publish()
手动连接connect() 返回 Disposableconnect() 返回 Disposable
连接回调connect(Consumer<Disposable>)connect(Consumer<? super Disposable>)
自动连接autoConnect(n)autoConnect(n)
引用计数refCount() / refCount(n)refCount() / refCount(n)
延迟断开refCount(n, timeout, unit)refCount(n, Duration)
重放replay()replay()
简写share()share()
断开副作用doOnDisposedoOnCancel
重置连接reset()无

两个 share() 的语义都是 publish().refCount(1)，同样会在订阅归零时断开。

两处真正的差异

1. reset() 只有 RxJava 有。 ConnectableObservable.reset() 会清空当前的连接状态，使下一次 connect() 重新订阅源。Reactor 的 ConnectableFlux 没有对应方法：一旦上游终止，只能通过 refCount 的下一次订阅重建连接，或者重新组装整条链。

2. 断开句柄的类型不同。 RxJava 用 io.reactivex.rxjava3.disposables.Disposable，Reactor 用 reactor.core.Disposable。它们的方法名都是 dispose()，但类型不兼容，跨库抽象时要各自包一层。

除此之外，「先订阅、后连接、多个下游共享一条上游」的模型在两边一致。迁移时不需要重写连接逻辑，只需要替换类型和少数方法名。

小结

模式何时连接何时断开重订阅时
publish() + connect()显式调用Disposable.dispose()需重新 connect()
refCount()第一个订阅者订阅者归零重新订阅源
refCount(n, timeout)第 n 个订阅者归零后再等 timeout重新订阅源
autoConnect(n)第 n 个订阅者不因归零断开不会重新连接

refCount 不缓存数据，connect() 在两边都返回 Disposable，reset() 是 RxJava 独有的。记住这三点，迁移时就不会被旧文章里的错误说法带偏。

阅读导航

上一篇：10.1 RxJava Subject：与 Sinks 的对照

下一篇：10.3 RxJava 3 与 Reactor 的迁移对照

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「java」更多文章

  1. 15.3 全书总结与学习路径回顾
  2. 15.2 响应式流处理的演进趋势
  3. 15.1 引擎选型指南