我很难理解如何使用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
替换为ReplaySubject
。
您可以使用BehaviorSubject
应用缓存,创建一个Observable
,然后将其转换为BehaviorSubject
,然后订阅此Subject
以获取数据。
现在,这个Subject
将作为Observable
进行响应,当订阅它时,它将向您提供最新获取的数据,并在更新时发出新数据。
请在此处查看更多详细信息和实现:https://medium.com/@jaerencoathup/repository-pattern-using-rxjava-and-room-4ce79e4ffc5c
另一种方法:https://medium.com/@elye.project/rxjava-clean-way-of-prefreshing-data-and-use-later-54800f2652d4
通过简单的例子了解更多关于主题的信息:https://blog.mindorks.com/understanding-rxjava-subject-publish-replay-behavior-and-async-subject-224d663d452f