RXJava 2调用两次

Question

我仍在尝试将RXJava2用于多个订阅者使用的轮询服务。它工作得很好，除了它总是调用两次。

我尝试使用publish(1)、take(1)、share()、refCount()等，但结果总是一样。叫两次或多次。

我还是不明白为什么它叫了两次。新订阅服务器应该接收最新发出的值，并且只在更改后的值（如果hashmap更改了）返回为modified List。

我的民意调查可观察到（单例）

public Observable<List<Light>> lightPolling = CallFactory
            .with(context)
            .getLights(MainApplication.lastaccespoint)
            .repeatWhen(o -> o.concatMap((Function<Object, ObservableSource<?>>) v -> Observable.timer(1000, TimeUnit.MILLISECONDS)))
            .filter(new MapPredicate<>())
            .distinctUntilChanged()

            .map(l -> {
                List<Light> lights = new ArrayList<>();
                for (Map.Entry<String, Light> entry : l.entrySet()) {
                    Light light = entry.getValue();
                    light.setId(entry.getKey());
                    lights.add(light);
                }
                return lights;
            })
            // .replay(1)
            //    .distinct()
            //.publish()
            //.share()
            .compose(ReplayingShare.instance()); // its like .share(), just including the last result etc.

我的谓词用于筛选hashmap是否真的发生了更改。Stringutil.EqualMap只是一个比较两个HashMap的简单方法，效果很好。

 class MapPredicate<T> implements Predicate<Map<String, T>> {

    private Map<String, T> lastMap;

    @Override
    public boolean test(Map<String, T> map) throws Exception {
        if (!StringHelper.equalMap(map, lastMap)) {
            Timber.d("Result in MapPredicate doesnt equals last result");
            lastMap = map;
            return false;
        }
        return true;
    }
}

最后，我的一次性订阅上了可观察的。

lightsDisposable = LightManager
            .getInstance(context)
            .lightPolling
            .subscribeWith(new BulbsObserver());

还有一些测试用的

 LightManager.getInstance(context).lightPolling.debounce(10, TimeUnit.SECONDS).subscribe(lights -> Timber.d("Received light second Subscriber "+ lights.size()));
 LightManager.getInstance(context).lightPolling.debounce(15, TimeUnit.SECONDS).subscribe(lights -> Timber.d("Received light Third Subscriber "+ lights.size()));

private class BulbsObserver extends DisposableObserver<List<Light>> {

    @Override
    public void onNext(List<Light> newLights) {
        Timber.d("Received lights=" + newLights.size());
        lights.clear();
        lights.addAll(newLights);
        adapter.notifyDataSetChanged();
    }

    @Override
    public void onError(Throwable throwable) {
        Timber.e(throwable);
    }

    @Override
    public void onComplete() {
        Timber.d("onComplete called");
    }
}

重播共享来自杰克·沃顿https://github.com/jakewharton/rxreplayingshare

Answer 1

refcount()在第一个订阅者到来时订阅（调用订阅函数），并在最后一个订阅者离开后取消订阅。如果您的订阅者按顺序来（即它们之间的引用计数下降到零），您将经历对原始源的多次订阅/取消订阅。

如果希望只订阅一次并且从不取消订阅，请使用autoconnect()而不是refcount()。例如：

hotObservable = coldObservable.replay(1).autoConnect();