一、描述

让一个可连接的Observable行为像普通的Observable。

可连接的Observable (connectable Observable)与普通的Observable差不多，不过它并不会在被订阅时开始发射数据，而是直到使用了Connect操作符时才会开始。用这种方法，你可以在任何时候让一个Observable开始发射数据。

RefCount操作符把从一个可连接的Observable连接和断开的过程自动化了。它操作一个可连接的Observable，返回一个普通的Observable。当第一个订阅者订阅这个Observable时，RefCount连接到下层的可连接Observable。RefCount跟踪有多少个观察者订阅它，直到最后一个观察者完成才断开与下层可连接Observable的连接。

二、示意图

refCount

三、示例代码

System.out.println("<========before refCount()=======>");

ConnectableObservable connectableObservable = Observable.range(1, 50000).sample(10, TimeUnit.MILLISECONDS).publish();

connectableObservable.subscribe(new Subscriber<Integer>() {
    @Override
    public void onCompleted() {
        System.out.println("onCompleted1.");
    }

    @Override
    public void onError(Throwable e) {
        System.out.println("onError1: " + e.getMessage());
    }

    @Override
    public void onNext(Integer integer) {
        System.out.println("onNext1: " + integer);
    }
});

connectableObservable.subscribe(new Subscriber<Integer>() {
    @Override
    public void onCompleted() {
        System.out.println("onCompleted2.");
    }

    @Override
    public void onError(Throwable e) {
        System.out.println("onError2: " + e.getMessage());
    }

    @Override
    public void onNext(Integer integer) {
        System.out.println("onNext2: " + integer);
    }
});

connectableObservable.connect();

System.out.println("<========after refCount()=======>");

Observable observable = connectableObservable.refCount();

observable.subscribe(new Subscriber<Integer>() {
    @Override
    public void onCompleted() {
        System.out.println("onCompleted3.");
    }

    @Override
    public void onError(Throwable e) {
        System.out.println("onError3: " + e.getMessage());
    }

    @Override
    public void onNext(Integer integer) {
        System.out.println("onNext3: " + integer);
    }
});

observable.subscribe(new Subscriber<Integer>() {
    @Override
    public void onCompleted() {
        System.out.println("onCompleted4.");
    }

    @Override
    public void onError(Throwable e) {
        System.out.println("onError4: " + e.getMessage());
    }

    @Override
    public void onNext(Integer integer) {
        System.out.println("onNext4: " + integer);
    }
});

四、运行结果

<========before refCount()=======>
onNext1: 11653
onNext2: 11653
onNext1: 21317
onNext2: 21317
onNext1: 32731
onNext2: 32731
onNext1: 40808
onNext2: 40808
onCompleted1.
onCompleted2.
<========after refCount()=======>
onNext3: 15517
onNext3: 30592
onNext3: 45208
onCompleted3.
onNext4: 16052
onNext4: 16090
onNext4: 20936
onNext4: 31274
onNext4: 47841
onCompleted4.