这一节解释ConnectableObservable 和它的子类以及它们的操作符：

一个可连接的Observable与普通的Observable差不多，除了这一点：可连接的Observable在被订阅时并不开始发射数据，只有在它的connect()被调用时才开始。用这种方法，你可以等所有的潜在订阅者都订阅了这个Observable之后才开始发射数据。

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:
下面的示例代码展示了两个订阅者订阅同一个Observable的情况。第一种情形，它们订阅一个普通的Observable；第二种情形，它们订阅一个可连接的Observable，并且在两个都订阅后再连接。注意输出的不同：

示例 #1:

  1. def firstMillion  = Observable.range( 1, 1000000 ).sample(7, java.util.concurrent.TimeUnit.MILLISECONDS);
  3. firstMillion.subscribe(
  4.    { println("Subscriber #1:" + it); },       // onNext
  5.    { println("Error: " + it.getMessage()); }, // onError
  6.    { println("Sequence #1 complete"); }       // onCompleted
  7. );
  9. firstMillion.subscribe(
  10.     { println("Subscriber #2:" + it); },       // onNext
  11.     { println("Error: " + it.getMessage()); }, // onError
  12.     { println("Sequence #2 complete"); }       // onCompleted
  13. );
  1. Subscriber #1:211128
  2. Subscriber #1:411633
  3. Subscriber #1:629605
  4. Subscriber #1:841903
  5. Sequence #1 complete
  6. Subscriber #2:244776
  7. Subscriber #2:431416
  8. Subscriber #2:621647
  9. Subscriber #2:826996
  10. Sequence #2 complete

示例 #2:

  1. def firstMillion  = Observable.range( 1, 1000000 ).sample(7, java.util.concurrent.TimeUnit.MILLISECONDS).publish();
  3. firstMillion.subscribe(
  4.    { println("Subscriber #1:" + it); },       // onNext
  5.    { println("Error: " + it.getMessage()); }, // onError
  6.    { println("Sequence #1 complete"); }       // onCompleted
  7. );
  9. firstMillion.subscribe(
  10.    { println("Subscriber #2:" + it); },       // onNext
  11.    { println("Error: " + it.getMessage()); }, // onError
  12.    { println("Sequence #2 complete"); }       // onCompleted
  13. );
  15. firstMillion.connect();
  1. Subscriber #2:208683
  2. Subscriber #1:208683
  3. Subscriber #2:432509
  4. Subscriber #1:432509
  5. Subscriber #2:644270
  6. Subscriber #1:644270
  7. Subscriber #2:887885
  8. Subscriber #1:887885
  9. Sequence #2 complete
  10. Sequence #1 complete

参见：