我们在 Android 应用程序中使用多种服务。这些服务提供无限的数据 Observables , 通常通过组合 Observables 来构造其他服务。 这些的 build Observables可能会很昂贵。此外，这些服务通常在多个地方使用，因此它们的 Observable应该在订阅者之间共享。

示例:

• LocationService , 提供无限 Observable<Location> , 发出当前位置
• ReminderService , 提供无限 Observable<List<Reminder>> , 在数据集中每次更改后发出所有存储提醒的列表
• LocationAwareReminderService , 提供无限 Observable<List<Reminders>> Observable.combineLatest 附近的提醒Observables前两项服务的

第一种方法:内部 BehaviorSubjects 作为缓存

每个服务结合消耗的 Observables并订阅它的内部BehaviorSubject到生成的提要。然后消费者可以订阅这个 BehaviorSubject . LocationAwareReminderService例如:

public class LocationAwareReminderService {

    Observable<List<Reminder>> feed;

    public LocationAwareReminderService(ReminderService reminderService, LocationService locationService) {
        BehaviorSubject<List<Reminder>> cache = BehaviorSubject.create();
        Observable.combineLatest(reminderService.getFeed(), locationService.getFeed(), new Func2<List<Reminder>, Location, List<Reminder>>() {
            @Override
            public List<Reminder> call(List<Reminder> reminders, Location location) {
                return calculateNearbyReminders(reminders, location);
            }
        }).subscribe(cache);

        feed = cache.asObservable();
    }

    public Observable<List<Reminder>> getFeed() {
        return feed;
    }
}

缺点:

• 由于 Behavior 主题，reminderService 和 locatoinService 的提要永远不会被拆除。即使没有消费者
• 如果它们依赖于不断发布新项目的服务(例如 LocationService)，这尤其成问题
• 由于构造函数中的订阅(缓存)，即使没有订阅者，服务也会开始计算附近的提醒

优点:

• 生成的提要由所有订阅者共享
• 因为 feed 永远不会被拆除，所以没有订阅者的短时间不会崩溃整个管道

第二种方法:replay(1).refCount().

public class LocationAwareReminderService {

    Observable<List<Reminder>> feed;

    public LocationAwareReminderService(ReminderService reminderService, LocationService locationService) {
        feed = Observable.combineLatest(reminderService.getFeed(), locationService.getFeed(), new Func2<List<Reminder>, Location, List<Reminder>>() {
            @Override
            public List<Reminder> call(List<Reminder> reminders, Location location) {
                return calculateNearbyReminders(reminders, location);
            }
        }).replay(1).refCount();
    }

    public Observable<List<Reminder>> getFeed() {
        return feed;
    }
}

缺点:

• 短时间没有 Subscriber折叠整个管道。在下一次订阅期间，需要重建整个管道。
• 从 Activity 的转换A到Activity B，都订阅了LocationAwareReminderService.getFeed() , 导致管道的完全去除和重建

优点:

• 在最后Subscriber之后取消订阅，LocationAwareReminderService也将取消订阅 LocationService.getFeed()和 reminderService.getFeed() Observables .
• LocationAwareReminderService仅在第一个 Subscriber 之后才开始提供 nearbyReminders已订阅
• 生成的提要由所有 Subscriber 共享小号

第三种方法:使用超时取消订阅 refCount

因此我构建了一个 Transformer在最后一个 Subscriber 之后的定义时间段内保持订阅有效退订

public class RxPublishTimeoutCache<T> implements Observable.Transformer<T, T> {

    private long keepAlive;
    private TimeUnit timeUnit;

    public RxPublishTimeoutCache(long keepAlive, TimeUnit timeUnit) {
        this.keepAlive = keepAlive;
        this.timeUnit = timeUnit;
    }

    @Override
    public Observable<T> call(Observable<T> upstream) {

        final Observable<T> sharedUpstream = upstream.replay(1).refCount();

        return Observable.create(new Observable.OnSubscribe<T>() {
            @Override
            public void call(Subscriber<? super T> subscriber) {
                if (subscriber.isUnsubscribed())
                    return;
                // subscribe an empty Subscriber that keeps the subsription of refCount() alive
                final Subscription keepAliveSubscription = sharedUpstream.subscribe(new NopSubscriber<T>());
                // listen to unsubscribe from the subscriber
                subscriber.add(Subscriptions.create(new Action0() {
                    @Override
                    public void call() {
                        // the subscriber unsubscribed
                        Observable.timer(keepAlive, timeUnit).subscribe(new Action1<Long>() {
                            @Override
                            public void call(Long _) {
                                // unsubscribe the keep alive subscription
                                keepAliveSubscription.unsubscribe();
                            }
                        });
                    }
                }));
                sharedUpstream.subscribe(subscriber);
            }
        });
    }

    public class NopSubscriber<T> extends Subscriber<T> {
        @Override
        public void onCompleted() {}
        @Override
        public void onError(Throwable e) {}
        @Override
        public void onNext(T o) {}
    }
}

LocationAwareReminderService利用RxPublishTimeoutCache

public class LocationAwareReminderService {

    Observable<List<Reminder>> feed;

    public LocationAwareReminderService(ReminderService reminderService, LocationService locationService) {
        feed = Observable.combineLatest(reminderService.getFeed(), locationService.getFeed(), new Func2<List<Reminder>, Location, List<Reminder>>() {
            @Override
            public List<Reminder> call(List<Reminder> reminders, Location location) {
                return calculateNearbyReminders(reminders, location);
            }
        }).compose(new RxPublishTimeoutCache<List<Reminder>>(10, TimeUnit.SECONDS));
    }

    public Observable<List<Reminder>> getFeed() {
        return feed;
    }
}

优点:

• LocationAwareReminderService仅在第一个 Subscriber 之后才开始提供 nearbyReminders已订阅
• 生成的提要由所有订阅者共享
• 短时间没有订阅者不会让整个管道崩溃
• 在定义的时间段内没有订阅后，整个管道将被拆除

缺点:

• 也许是一些普遍的缺陷？

问题:

• 在 RxJava 中是否已经有其他方法可以实现这一点？
• RxPublishTimeoutCache 中是否存在一些一般设计缺陷？ ？
• 使用 RxJava 组合此类服务的总体策略是否存在缺陷？

最佳答案

我认为这是一个有趣的问题并且似乎是一个有用的运算符所以我在 rxjava-extras 中创建了 Transformers.delayFinalUnsubscribe :

observable
  .publish()
  .refCount()
  .compose(Transformers
      .delayFinalUnsubscribe(1, TimeUnit.MINUTES));

Maven Central 从 0.7.9.1 开始在 rxjava-extras 中可用。如果愿意，请试一试，看看是否有任何问题。

关于android - 如何在最后一个订阅者退订后延迟拆除共享的、无限的 Observables，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/36597039/