使用concat（）和first（）实现RxJava并发调用的缓存

Question

我正在使用RxJava和concat（）和first（）操作符：

public Observable<List<Entity>> getEntities() {
    invalidateCacheIfNeeded();
    return Observable
            .concat(cachedEntities(), networkEntities())
            .first();
}

cachedEntities返回从缓存列表构建的可观察，而networkEntities方法使用Retrofit获取实体。

除非两个用户快速订阅getEntities（）返回的可观察对象，否则这非常有效。我猜在进行第二次订阅时，第一次订阅的网络请求没有完成。在这种情况下，执行了两个网络请求。我想避免。

我尝试创建一个单线程调度程序，以便仅在第一次调用结束时执行第二次调用，但没有运气：

 mSingleThreadScheduler = Schedulers.from(Executors.newSingleThreadExecutor());

以及：

public Observable<List<Entity>> getEntities() {
    invalidateCacheIfNeeded();
    return Observable
            .concat(cachedEntities(), networkEntities())
            .subscribeOn(mSingleThreadScheduler)
            .first();
}

我曾尝试将subscribeOn调用放在可观察链的较低位置，但我得到了相同的结果。

有什么提示吗？

Answer 1

这似乎是一个可观察到的具有多个订阅者的相对常见的用例。您需要类似的东西

public Observable<List<Entity>> getEntities() {
    invalidateCacheIfNeeded();
    return Observable
            .concat(cachedEntities(), networkEntities())
            .first()
            .replay(1)            
}

有关更深入的解释，请参阅此问题的答案。

Answer 2

给予

public Observable<List<Entity>> getEntities() {
    invalidateCacheIfNeeded();
    return Observable
            .concat(cachedEntities(), networkEntities())
            .first();
}

您应该创建一个AsyncSubject

private Observable<List<Entity>> networkEntities() {
    return mSubject
            .map(Data::getEntities);
}

你的网络通话应该是这样的

public Observable<Data> getDataFromNetwork() {
    return networkOperation()
            .subscribeOn(mSingleThreadScheduler)
            .subscribe(mSubject);
}

Answer 3

我认为使方法线程安全不是一个好主意。因为它会阻塞整个方法从而降低性能。因此建议将数据结构设为线程安全的。在您的情况下，您正在方法中使用List

public Observable<List<Entity>> getEntities() {

}

使用CopyOnWriteArrayList而不是List。它是线程安全的。

public Observable<CopyOnWriteArrayList<Entity>> getEntities() {

}

希望它会起作用。