我很难理解如何使用 RxJava 构建缓存。我的想法是,我需要从内存缓存获取数据或从数据库(dynamoDb)加载数据。但是,该缓存应该在片段和/或线程之间共享。所以我需要返回当前正在运行且未完成的现有可观察对象。这允许线程赶上而不做不必要的工作。我是 RxJava 的新手,所以这就是我的草图(为了简洁起见,遗漏了一些代码):
public class DBCache<K, T> {
private final ConcurrentHashMap<K, Set<T>> resultCache = new ConcurrentHashMap<>;
private final ConcurrentHashMap<K, Observable<Set<T>>> observableCache = new ConcurrentHashMap<>;
private Observable<Set<T>> getFromCache(final DynamoDbCacheKey<K, T> query) {
return Observable.create(new Observable.OnSubscribe<Set<T>>() {
@Override
public void call(Subscriber<? super Set<T>> subscriber) {
Set<T> results = resultCache.get(query.getKey());
if (results != null && results.size() > 0) {
subscriber.onNext(results);
}
subscriber.onCompleted();
}
});
}
public Observable<Set<T>> get(final QueryCacheKey<K, T> query){
Observable<Set<T>> cachedObservable = observableCache.get(query.getKey());
if (cachedObservable != null) {
return cachedObservable;
}
Observable<Set<T>> observable = Observable
.concat(getFromCache(query), getFromNetwork(query))
.first()
.subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
.cache();
observableCache.putIfAbsent(query.getKey(), observable);
return observable;
}
private Observable<Set<T>> getFromNetwork(final QueryCacheKey<K, T> query) {
return Observable.create(new Observable.OnSubscribe<Set<T>>() {
@Override
public void call(Subscriber<? super Set<T>> subscriber) {
try {
Set<T> results = loadFromDb(query); //omitted
resultCache.putIfAbsent(query.getKey(), results);
subscriber.onNext(results);
subscriber.onCompleted();
observableCache.remove(query.getKey());
} catch (Exception exception) {
subscriber.onError(exception);
}
}
});
}
}
有没有更好的方法通过 RxJava 来实现这一点(对缓存策略不感兴趣)。有什么想法吗?
这是一个进行一次缓存和检索值的简单示例:
public class RxCache<K, V> {
final ConcurrentHashMap<K, AsyncSubject<V>> cache;
final Func1<K, Observable<V>> valueGenerator;
public RxCache(Func1<K, Observable<V>> valueGenerator) {
this.valueGenerator = valueGenerator;
this.cache = new ConcurrentHashMap<>();
}
public Observable<V> get(K key) {
AsyncSubject<V> o = cache.get(key);
if (o != null) {
return o;
}
o = AsyncSubject.create();
AsyncSubject<V> p = cache.putIfAbsent(key, o);
if (p != null) {
return p;
}
valueGenerator.call(key).subscribe(o);
return o;
}
public void remove(K key) {
cache.remove(key);
}
}
如果有多个值,请替换AsyncSubject
with ReplaySubject
.
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)