此篇內(nèi)容均是來(lái)自書(shū)籍《RxJava響應(yīng)式編程》李衍順 著

3.9 與Connectable Observable相關(guān)的操作符

我們前面所學(xué)的Observable造垛，他們又一個(gè)共同的特性魔招，就是只有當(dāng)訂閱者來(lái)訂閱時(shí)才會(huì)開(kāi)始發(fā)送數(shù)據(jù)，否則什么也不發(fā)生五辽，這就是懶加載办斑。那什么是Connectable Observable呢， 它是一種特殊的Observable杆逗，并不是在訂閱者訂閱時(shí)才發(fā)送數(shù)據(jù)乡翅，而是只要對(duì)其應(yīng)用connect操作符就開(kāi)始發(fā)送數(shù)據(jù)。

3.9.1 publish 和connect

publish操作符就是用來(lái)將一個(gè)普通的Observable轉(zhuǎn)化為一個(gè)Connectable Observable的罪郊。需要注意的是蠕蚜，如果發(fā)送數(shù)據(jù)已經(jīng)開(kāi)始了再進(jìn)行訂閱的話，就只能接收以后發(fā)送的數(shù)據(jù)悔橄。

connect操作符就是用來(lái)觸發(fā)Connectable Observable發(fā)送數(shù)據(jù)的靶累。應(yīng)用connect操作符后會(huì)返回一個(gè)Subscription對(duì)象， 通過(guò)這個(gè)Subscription對(duì)象癣疟，我們可以調(diào)用其unsubscribe方法來(lái)終止數(shù)據(jù)的發(fā)送挣柬。另外，如果還沒(méi)有訂閱者訂閱就應(yīng)用connect操作符睛挚，也是可以使其開(kāi)始發(fā)送數(shù)據(jù)的邪蛔。

/**
     * publish && connect
     */
    private void publishConnectTest(){
        Observable<Long> obser = Observable.interval(1, TimeUnit.SECONDS, Schedulers.trampoline());
        final ConnectableObservable<Long> observable = obser.publish();
        final Action1 action2 = new Action1() {
            @Override
            public void call(Object o) {
                log("action2: " + o);
            }
        };
        Action1 action1 = new Action1() {
            @Override
            public void call(Object o) {
                log("action1: " + o);
                if((long)o==3)
                    observable.take(6).subscribe(action2);
            }
        };
        observable.take(6).subscribe(action1);
        observable.connect();
    }
結(jié)果：
action1: 0
action1: 1
action1: 2
action1: 3
action1: 4
action2: 4
action1: 5
action2: 5
action2: 6
action2: 7
action2: 8
action2: 9

3.9.2 refCount

refCount 操作符能夠?qū)⒁粋€(gè)Connectable Observable對(duì)象再重新轉(zhuǎn)化為一個(gè)普通的Observable對(duì)象，這時(shí)候如果有訂閱者進(jìn)行訂閱將會(huì)觸發(fā)數(shù)據(jù)的發(fā)送竞川。

/**
     * refCount
     */
    private void refCountTest(){
        Observable<Long> obser = Observable.interval(1, TimeUnit.SECONDS, Schedulers.trampoline());
        ConnectableObservable<Long> observable = obser.publish();
        observable.refCount().take(5).subscribe(new Action1<Long>() {
            @Override
            public void call(Long aLong) {
                log("refCount: "+ aLong);
            }
        });
    }
訂閱后會(huì)讓Observable立刻開(kāi)始生產(chǎn)并發(fā)送數(shù)據(jù)
refCount: 0
refCount: 1
refCount: 2
refCount: 3
refCount: 4

3.9.3 replay

replay操作符返回一個(gè)Connectable Observable對(duì)象并且可以緩存其發(fā)送過(guò)的數(shù)據(jù)店溢，這樣即使有訂閱者在其發(fā)送數(shù)據(jù)之后進(jìn)行訂閱，也能收到其之前發(fā)送過(guò)的數(shù)據(jù)委乌。不過(guò)使用replay操作符最好還是限定緩存大小床牧， 否則如果緩存的數(shù)據(jù)太多的話，可會(huì)占用很多內(nèi)存遭贸。對(duì)緩存的控制可以從空間和時(shí)間兩個(gè)維度來(lái)實(shí)現(xiàn)戈咳。

/**
     * replay
     */
    private ConnectableObservable<Long> replayCountObserver(){
        Observable<Long> obser = Observable.interval(1, TimeUnit.SECONDS, Schedulers.trampoline());
        return obser.replay(2);
    }
    private ConnectableObservable<Long> replayTimeObserver(){
        Observable<Long> obser = Observable.interval(1, TimeUnit.SECONDS, Schedulers.trampoline());
        return obser.replay(3, TimeUnit.SECONDS);
    }
    private void replayTest(){
        final ConnectableObservable<Long> observer = replayCountObserver();
        final Action1 action2 = new Action1() {
            @Override
            public void call(Object o) {
                log("action2: " + o);
            }
        };
        Action1 action1 = new Action1() {
            @Override
            public void call(Object o) {
                log("action1: " + o);
                if((long)o==3)
                    observer.take(6).subscribe(action2);
            }
        };
        observer.take(10).subscribe(action1);
        log("relaycount");
        observer.connect();
    }

這時(shí)將會(huì)得到如下的結(jié)果。Action1在接收到3后吧Action2也訂閱上了，由于緩存的空間是2著蛙，所以Action2可以接收到之前的兩個(gè)數(shù)據(jù)2和3删铃，之后Action1和Action2會(huì)共同接收后面的數(shù)據(jù)

relaycount
action1: 0
action1: 1
action1: 2
action1: 3
action2: 2
action2: 3
action1: 4
action2: 4
action1: 5
action2: 5
action1: 6
action2: 6
action1: 7
action2: 7
action1: 8
action1: 9

下面我們使用時(shí)間緩存的Observable來(lái)訂閱，使用connect操作符后我們得到如下結(jié)果踏堡。Action1在接收到數(shù)據(jù)3之后把Action2也訂閱上了猎唁，Action2收到了之前3秒緩存的所有數(shù)據(jù)。之后共同接收后面的數(shù)據(jù)顷蟆。

timecount
action1: 0
action1: 1
action1: 2
action1: 3
action2: 0
action2: 1
action2: 2
action2: 3
action1: 4
action2: 4
action1: 5
action2: 5
action1: 6
action1: 7
action1: 8
action1: 9