這一節(jié)解釋ConnectableObservable 和它的子類以及它們的操作符：

• ConnectableObservable.connect(?) — 指示一個(gè)可連接的Observable開始發(fā)射數(shù)據(jù)
• Observable.publish(?) — 將一個(gè)Observable轉(zhuǎn)換為一個(gè)可連接的Observable
• Observable.replay(?) — 確保所有的訂閱者看到相同的數(shù)據(jù)序列，即使它們?cè)贠bservable開始發(fā)射數(shù)據(jù)之后才訂閱
• ConnectableObservable.refCount(?) — 讓一個(gè)可連接的Observable表現(xiàn)得像一個(gè)普通的Observable

一個(gè)可連接的Observable與普通的Observable差不多，除了這一點(diǎn)：可連接的Observable在被訂閱時(shí)并不開始發(fā)射數(shù)據(jù)，只有在它的connect()被調(diào)用時(shí)才開始。用這種方法，你可以等所有的潛在訂閱者都訂閱了這個(gè)Observable之后才開始發(fā)射數(shù)據(jù)。

http://wiki.jikexueyuan.com/project/rx-docs/images/operators/publishConnect.png" alt="publishConnect" />

The following example code shows two Subscribers subscribing to the same Observable. In the first case, they subscribe to an ordinary Observable; in the second case, they subscribe to a Connectable Observable that only connects after both Subscribers subscribe. Note the difference in the output: 下面的示例代碼展示了兩個(gè)訂閱者訂閱同一個(gè)Observable的情況。第一種情形，它們訂閱一個(gè)普通的Observable；第二種情形，它們訂閱一個(gè)可連接的Observable，并且在兩個(gè)都訂閱后再連接。注意輸出的不同：

示例 #1:

def firstMillion  = Observable.range( 1, 1000000 ).sample(7, java.util.concurrent.TimeUnit.MILLISECONDS);

firstMillion.subscribe(
   { println("Subscriber #1:" + it); },       // onNext
   { println("Error: " + it.getMessage()); }, // onError
   { println("Sequence #1 complete"); }       // onCompleted
);

firstMillion.subscribe(
    { println("Subscriber #2:" + it); },       // onNext
    { println("Error: " + it.getMessage()); }, // onError
    { println("Sequence #2 complete"); }       // onCompleted
);

Subscriber #1:211128
Subscriber #1:411633
Subscriber #1:629605
Subscriber #1:841903
Sequence #1 complete
Subscriber #2:244776
Subscriber #2:431416
Subscriber #2:621647
Subscriber #2:826996
Sequence #2 complete

示例 #2:

def firstMillion  = Observable.range( 1, 1000000 ).sample(7, java.util.concurrent.TimeUnit.MILLISECONDS).publish();

firstMillion.subscribe(
   { println("Subscriber #1:" + it); },       // onNext
   { println("Error: " + it.getMessage()); }, // onError
   { println("Sequence #1 complete"); }       // onCompleted
);

firstMillion.subscribe(
   { println("Subscriber #2:" + it); },       // onNext
   { println("Error: " + it.getMessage()); }, // onError
   { println("Sequence #2 complete"); }       // onCompleted
);

firstMillion.connect();

Subscriber #2:208683
Subscriber #1:208683
Subscriber #2:432509
Subscriber #1:432509
Subscriber #2:644270
Subscriber #1:644270
Subscriber #2:887885
Subscriber #1:887885
Sequence #2 complete
Sequence #1 complete

參見：

• ConnectableObservable
• Introduction to Rx: Publish and Connect