java - RXJava 2 调用两次

Question

我仍在尝试使用 RXJava 2 来实现多个订阅者使用的轮询服务。它运行良好，只是它总是调用两次。

我尝试使用publish(1)、take(1)、share()、refCount()等，但结果总是相同的。调用两次或多次。

我还是不明白为什么它被调用了两次。新订阅者应该收到最新发出的值，并且仅在更改后的值(如果 HashMap 更改)作为修改后的列表返回。

我的轮询可观察(单例)

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;
    }
}

最后我的 Disposable 订阅了 Observable。

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");
    }
}

重播分享来自 Jack Wharton https://github.com/JakeWharton/RxReplayingShare

Answer 1

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

如果您只想订阅一次并且永不取消订阅，请使用 autoConnect() 而不是 refCount()。例如:

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