java - RxJava 异步缓存 : proper way to dispose replay(). autoConnect() Observable

Question

我必须为可观察结果提供一个短期缓存。

查看选项，我看到以下内容:

1. 缓存 replay(1).refCount()当数据准备好时，缓存实际值。缓存检索将检查实际数据并执行 Observable.just或返回挂起的 Observable 或发起新的请求。

2. 缓存 replay(1).autoConnect(1)并始终返回那个

后者看起来更直接，但它有一个警告，即当缓存必须失效时如何正确处理 observable。

有一个签名:

public Observable<T> autoConnect(int numberOfSubscribers, Consumer<? super Disposable> connection)

但很难说我如何跟踪未完成的订阅，以及处理是否妥当。

前者将负责资源释放，但您必须产生更复杂的逻辑。

Answer 1

为什么不是 .cache()？

public class CachedObservable<K,V> {
  private Function<K, Observable<V>> actual;
  private CachedObservable(Function<K, Observable<V>> actual){this.actual=actual;}
  private final Map<K, Observable<V>> cacheMap = new ConcurrentHashMap<>();

  public Observable<V> get(K key) {
    return cacheMap.computeIfAbsent(key, k -> this.actual.call(k).cache());
  }
  public void invalidate(K key){cacheMap.remove(key);}
}