rx-java - 是否存在所有观察者都取消订阅 Observable 的事件？

Question

我有一个 Observable，我希望它是冷的，也就是说，它应该只在第一个观察者订阅它时才开始发出项目。

然后，我想确保当所有观察者取消订阅同一个可观察对象时从源中释放所有资源。这可能吗？

Answer 1

您可以使用 ConnectedObservable 的功能来为您处理此问题:

//Replace Observable.range(1,1000) with your Observable implementation
Observable.range(1, 1000).doOnUnsubscribe(() -> freeResources()).share();

share 方法调用方法 publish 和 refCount 。

publish 将您的“普通”Observable 转换为 ConnectedObservable，它将在您调用 connect 时开始发出项目。因此，从技术上讲，您可以根据需要订阅任意数量的观察者，然后调用 connect 开始同时为所有观察者发出项目。

refCount 将您的 ConnectedObservable 再次转换回传统的，但具有新的特性!额外的好处是:这个 observable 现在是冷的(只有当订阅者订阅时才开始发出，它在内部调用使用 publish 创建的原始 ConnectedObservable 的 connect 方法)，并保持跟踪有多少订阅者连接到原始 ConnectedObservable。一旦所有订阅者都取消订阅，它将从源 ConnectedObservable 取消订阅，因此逻辑变得更加简单，因为您只需要处理一个订阅。

这里有一个很好的共享操作图:http://reactivex.io/RxJava/javadoc/rx/Observable.html#share()

或者，如果这不够灵活，我认为您应该能够通过使用 defer 轻松实现此行为，以创建冷可观察对象，以及 doOnSubscribe 和 doOnUnsubscribe 方法。

示例:

    Observable.defer(() -> {
        final AtomicInteger counter = new AtomicInteger();
        return Observable.range(1, 1000)
                .doOnSubscribe(() -> counter.incrementAndGet())
                .doOnUnsubscribe(() -> {
                    if (counter.decrementAndGet() == 0) {
                        freeResources();
                    }
                });
    });

一旦第一个订阅者订阅，此可观察对象将开始发出一系列数字(将其替换为您的可观察实现)，每次订阅都会增加一个计数器，并在所有订阅者取消订阅后释放已使用的资源(替换免费资源满足您的任何需求)。