由于您的数据层不具备响应式特性,因此它无法正常工作,尽管使用了 Rx。
Realm 本质上是一个具有响应式数据源,其管理的对象也是可变的(由 Realm 原地更新)并且线程限制的(只能在打开 Realm 的同一线程上访问)。
要使您的代码起作用,您需要从 Realm 中复制数据。
@Override
public Single<List<ResponseZone>> entityList() {
return Single.fromCallable(() -> {
try(Realm realm = Realm.getDefaultInstance()) {
return realm.copyFromRealm(realm.where(ResponseZone.class).findAll());
}
});
}
我冒昧地将您的Single
表示为Single
,考虑到它不是Observable,它不会监听变化,只有1个事件,即列表本身。因此,通过ObservableEmitter
发送它没有意义,因为它不会发出事件。
因此,这就是我说的原因:您的数据层不具有响应性。您没有监听变化。您只是直接获取数据,并且永远不会收到任何更改的通知;尽管使用了Rx。
我在画图板中画了一些图片来说明我的观点。(蓝色表示的是副作用)
![clean-architecture non-reactive](https://istack.dev59.com/ntRSz.webp)
在您的情况下,您调用一次性操作从多个数据源(缓存、本地、远程)检索数据。一旦您获取它,您就不会监听变化;如果您在一个地方编辑数据并在另一个地方,更新的唯一方法是“强制缓存手动检索新数据”;对于此,您必须知道您在其他地方修改了数据。为此,您需要一种直接调用回调或发送消息/事件(通知更改)的方法。
因此,在某种程度上,您必须创建一个缓存失效通知事件。如果您监听它,解决方案可能再次具有响应性。除了您手动执行此操作。
----------------------------------------------------------------------
考虑到Realm已经是一种响应式数据源(类似于用于SQLite的SQLBrite),它能够通过提供更改通知来“使您的缓存失效”。
事实上,如果您的本地数据源是唯一的数据源,并且从网络中的任何写入都是您监听的更改,则您的“缓存”可以写为replay(1).publish().refCount()
(重复最新数据以供新订阅者,使用新数据替换数据进行评估),这是RxReplayingShare。
![clean architecture reactive](https://istack.dev59.com/xBmWi.webp)
使用从处理程序线程的循环器创建的调度程序,您可以在后台线程上监听Realm中的更改,创建一种响应式数据源,返回最新的未管理副本,您可以在线程之间传递它们(尽管直接映射到不可变领域模型比copyFromRealm()
更受欢迎,如果您选择这条路线-路线是清晰的架构)。
return io.reactivex.Observable.create(new ObservableOnSubscribe<List<ResponseZone>>() {
@Override
public void subscribe(ObservableEmitter<List<ResponseZone>> emitter)
throws Exception {
final Realm observableRealm = Realm.getDefaultInstance();
final RealmResults<ResponseZone> results = observableRealm.where(ResponseZone.class).findAllAsync();
final RealmChangeListener<RealmResults<ResponseZone>> listener = results -> {
if(!emitter.isDisposed()) {
if(results.isValid() && results.isLoaded()) {
emitter.onNext(observableRealm.copyFromRealm(results));
}
}
};
emitter.setDisposable(Disposables.fromRunnable(() -> {
if(results.isValid()) {
results.removeChangeListener(listener);
}
observableRealm.close();
}));
results.addChangeListener(listener);
}
}).subscribeOn(looperScheduler).unsubscribeOn(looperScheduler);
Looper调度程序的获取方式为
handlerThread = new HandlerThread("LOOPER_SCHEDULER");
handlerThread.start();
synchronized(handlerThread) {
looperScheduler = AndroidSchedulers.from(handlerThread.getLooper());
}
这就是如何使用Realm创建响应式的清洁架构。
补充:
如果您打算在Realm上实际实施Clean Architecture,那么需要使用LooperScheduler。这是因为默认情况下,Realm鼓励您将数据对象用作域模型,并提供惰性加载的线程本地视图,当更新时会原地变异;但是Clean Architecture则建议您使用不可变的域模型(与数据层无关)。因此,如果您想要创建响应式的清洁架构,在任何时候当Realm发生更改时都要从后台线程复制,则需要一个looper scheduler(或在后台线程上观察,但要从刷新的Realm上进行复制Schedulers.io()
)。
通常情况下,您会希望使用RealmObjects作为域模型,并依赖于惰性计算。在这种情况下,您不使用copyFromRealm()
,也不将RealmResults映射到其他内容;但可以将其公开为Flowable
或LiveData
。
您可以在这里阅读相关内容。
entityList()
的线程上进行观察。暂时地,你需要避免这样做或使用copyFromRealm()
将Realm数据复制到Java内存中。 - Christian Melchior