RxJava反应式编程

Reactive Extensions

｜

43

ConnectableObservable 有很多应用场景，例如：不管 Subscriber 何时订阅，都要确保

它们接收到相同的事件序列。如果订阅时会有重要的副作用，即便还没有“真正”的

Subscriber，ConnectableObservable 也能强迫订阅。后续很快就将讨论上述的所有场景。

Subject 是创建 Observable 的必要方式，而 ConnectableObservable 会保护原始的上游

Observable，并确保最多只能有一个 Subscriber 可以接触到它。不管有多少 Subscriber 连

接到 ConnectableObservable，系统只会打开一个 Observable 的订阅，这个订阅是基于该

Observable 创建的。

2.7.1

　使用

publish().refCount()

实现单次订阅

回顾一下：只有一个到底层资源的句柄，如到

Twitter

状态更新流的

HTTP

连接。但是，推

送事件的这个 Observable 将会被多个 Subscriber 共享。前面创建的 Observable 的原生实

现并没有对此进行控制，因此每个 Subscriber 会启动自己的连接。如下所示的程序代码是

非常浪费的。

Observable<Status> observable = Observable.create(subscriber -> {

System.out.println("Establishing ...