我很难理解如何使用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来实现这一点(不感兴趣的缓存策略)。 有什么想法吗?