rx-java - 使用 RxAndroid 进行智能轮询

Question

我想要实现的目标:我想通过每 5 分钟轮询一次的方式从网络上轮询一些资源，但前提是有观察者订阅。我使用BehaviorSubject 和interval observable 进行池化。我设法实现了它，但我是 Rx 新手，我认为它可以做得更好。

我就是这样做的:

private BehaviorSubject<String> observable;
private Subscription> subscription;

public Subscription subscribe(final Action1<String> action) {
    if (observable == null) {
        observable = BehaviorSubject.create();
    }
    if (subscription == null) {
        Observable<String> source = Observable.interval(5, TimeUnit.MINUTES).map(new Func1<Long, String>() {
            @Override
            public String call(Long aLong) {
                return getDataFromServer();
            }
        }).observeOn(AndroidSchedulers.mainThread());
        subscription = source.subscribe(new Action1<String>() {

            @Override
            public void call(String s) {
                if (observable.hasObservers()) {
                    observable.onNext(s);
                } else {
                    subscription.unsubscribe();
                    subscription = null;
                }
            }
        });
    }
    return observable.subscribe(action);
}

想法:- 我有一个用于轮询的可观察源和另一个客户端可以订阅的可观察源(使用 BehaviourSubject 实现 - 所以他们总是可以获得最新数据)。当源 observable 发出一些东西时，如果行为主体有订阅者，它就会被传递，否则什么也不会传递，我从源取消订阅，所以它会停止。

Answer 1

关于:

Observable<String> observable = Observable.interval(0, 5, TimeUnit.SECONDS)
    .doOnNext(new LoggingAction1<Long>("doOnNext"))
    .flatMap(new Func1<Long, Observable<String>>() {
        @Override
        public Observable<String> call(Long pulse) {
            return Observable.just(String.format("Request %d", pulse));
        }
    })
    .replay(1)
    .refCount();

我认为它可以满足您希望您的设置执行的所有操作:

• 只要没有订阅者，它就不会执行任何操作。
• 当第一个订阅者订阅时，间隔开始并立即发出一个值，然后每 5 秒发出一个值。
• 新的订阅者将立即获取最后一项，然后获取所有后续项。
• 只会启动一个间隔 - 因此每 5 秒只会执行一个网络请求 - 无论有多少订阅者。
• 当所有订阅者取消订阅时，间隔将停止发出项目。