在 Rxjava 中与 concat 运算符集成缓存时出现问题答案

【问题标题】：Problem when integrate caching with concat operator in Rx java在 Rxjava 中与 concat 运算符集成缓存时出现问题
【发布时间】：2020-08-08 02:39:48
【问题描述】：

通过以下链接实现缓存： https://blog.mindorks.com/implement-caching-in-android-using-rxjava-operators

fun getSavedAddressList(): Maybe<List<SavedAddress>?>? {
    return Observable.concat(
            getAddressListMemory(),
            getAddressListDatabase(),
            getAddressListNetwork()).firstElement()
}

   fun getAddressListDatabase(): Observable<List<SavedAddress>?> {
        return Observable.create(ObservableOnSubscribe { emitter: ObservableEmitter<List<SavedAddress>?> ->
            val list: ArrayList<SavedAddress> = addressDao.getAddressList() as ArrayList<SavedAddress>
            if (list.isNotEmpty()) {
                emitter.onNext(list)
            }
            emitter.onComplete()
            if (list.isNotEmpty())


    getAddressListNetwork().subscribeOn(schedulerProvider.io())?.observeOn(schedulerProvider.io())?.subscribe()
        })
    }

数据库中的项目在存储到数据库后可以完美检索

问题是从数据库中获取列表后没有发生网络调用

我想依次获取三个数据源，并将最新的数据存储到数据库中

【问题讨论】：

确保getAddressListDatabase()调用onComplete()，否则concat不会切换到下一个Observable
@borichellow 更新了我的代码
最后一行getAddressListNetwork().subscribeOn()...的目的是什么？为什么需要它？
@borichellow 在成功响应后从数据库调用 getAddressListNetwork 获取数据后将新数据存储到数据库

标签： kotlin retrofit rx-java rx-java2

【解决方案1】：

首先，您在其中泄漏了getAddressListNetwork Disposable，因为您试图在getAddressListDatabase 内部做太多事情。

我想你想要的是这样的：

fun getSavedAddressList(): Observable<List<SavedAddress>> {
    return Observable.concat(
            getAddressListMemory(),
            getAddressListDatabase(),
            getAddressListNetwork()).distinctUntilChanged())
}

这将始终尝试从 3 个来源获取地址，并且仅在数据与之前的发射不同时发射，这意味着数据“更新鲜”。

老实说，我认为你需要看看“陈旧数据”和“缓存失效”的概念。

【讨论】：