RxJava反应式编程

Reactive Extensions

｜

39

如果有人进行订阅却只想得到流的一小部分，

Observable 会确保进行资源清理。我们知道

还有第二项实现清理的技术，它不需要等待上游的事件。订阅者取消订阅，样例就立即调

用 shutdown() 以触发清理行为（最后一行），而不必等待下一条推文到达。

twitterStream.addListener(new StatusListener() {

//回调……

});

twitterStream.sample();

subscriber.add(Subscriptions.create(twitterStream::shutdown));

比较有意思的是，这个 Observable 模糊了

hot

流和

cold

流的差异。一方面，它代表了外

部的事件，这些事件的出现并不受控制（

hot

流的行为）。另一方面，实际订阅之前，事

件并不会开始流动（没有底层的

HTTP

连接）。我们还忽略了另外一个不慎出现的副作

用：每个新的 subscribe() 都会开启一个新的后台线程和一条到外部系统的新连接。相

同的 Observable<Status> 实例应该能够跨多个订阅者实现重用。因为 Observable 是延

迟执行的，所以需要在技术上能够在启动的时候就调用 observe()，并将其保留在某个

单例对象中。但是当前的实现只是简单地打开新连接，为每个 Subscriber 通过网络多次

获取相同的数据。当然，我们想要为这个流注册多个 Subscriber，但是没有理由为每个 ...