RxJava可见。缓存失效

Question

我正在尝试在Android环境中学习rxjava。比方说，我有一个可观察的对象，它发出网络调用的结果。如果我理解正确的话，处理配置更改的一种常见方法是：

>

将可观察对象存储在保留的片段/单例/应用程序对象中

将缓存操作符应用于可观察对象

在正确的生命周期处理程序中订阅/取消订阅

这样做，我们就不会失去可观察到的结果，一旦新配置发生，将重新观察到该结果。

现在，我的问题是：

有没有办法强制可观察对象发出新值（并使缓存的值无效）？每当我想从网络上获取新数据时，我是否需要创建一个新的可观察对象（这在android世界听起来并不是一个糟糕的做法，因为这会让gc做额外的工作）？

谢谢，

费德里科

Answer 1

制作一个自定义的OnSubscribe实现，它可以做你想要的：

public static class OnSubscribeRefreshingCache<T> implements OnSubscribe<T> {

    private final AtomicBoolean refresh = new AtomicBoolean(true);
    private final Observable<T> source;
    private volatile Observable<T> current;

    public OnSubscribeRefreshingCache(Observable<T> source) {
        this.source = source;
        this.current = source;
    }

    public void reset() {
        refresh.set(true);
    }

    @Override
    public void call(Subscriber<? super T> subscriber) {
        if (refresh.compareAndSet(true, false)) {
            current = source.cache();
        }
        current.unsafeSubscribe(subscriber);
    }

}

这段代码演示了使用情况，并显示缓存实际上正在重置：

Observable<Integer> o = Observable.just(1)
        .doOnCompleted(() -> System.out.println("completed"));
OnSubscribeRefreshingCache<Integer> cacher = 
    new OnSubscribeRefreshingCache<Integer>(o);
Observable<Integer> o2 = Observable.create(cacher);
o2.subscribe(System.out::println);
o2.subscribe(System.out::println);
cacher.reset();
o2.subscribe(System.out::println);

输出：

completed
1
1
completed
1

顺便说一下，您可能会注意到。缓存在完成之前不会发出。这是一个应该由rxjava 1.0.14修复的bug。

就GC压力问题而言，每个操作员在应用于可观察对象时，通常通过提升或创建来创建新的可观察对象。与创建新可观察对象相关的基本成员国是对Subscribe函数的引用<代码>缓存与大多数缓存的不同之处在于，它跨订阅保存状态，如果它保存大量状态并经常被丢弃，则可能会产生GC压力。即使您使用相同的可变数据结构在重置过程中保持状态，GC在清除时仍然必须处理数据结构的内容，因此您可能不会获得太多收益。

RxJava缓存操作符是为多个并发订阅而构建的。您可能会想象，重置功能的实现可能会出现问题。如果您想进一步探索，请务必在RxJava github上提出一个问题。