使用 countDownLatch.await() 确保结果已交付答案

【问题标题】：using countDownLatch.await() to make sure result is delivered使用 countDownLatch.await() 确保结果已交付
【发布时间】：2020-11-28 20:11:42
【问题描述】：

完整的源代码可以在这里找到：https://github.com/alirezaeiii/SavingGoals-Cache

这是 LocalDataSource 类：

@Singleton
class QapitalLocalDataSource @Inject constructor(
    private val goalsDao: GoalsDao
) : LocalDataSource {

    override fun getSavingsGoals(): Single<List<SavingsGoal>> =
        Single.create { singleSubscriber ->
            goalsDao.getGoals()
                .subscribe {
                    if (it.isEmpty()) {
                        singleSubscriber.onError(NoDataException())
                    } else {
                        singleSubscriber.onSuccess(it)
                    }
                }
        }
}

Repository 类中使用了上述方法：

@Singleton
class GoalsRepository @Inject constructor(
    private val remoteDataSource: QapitalService,
    private val localDataSource: LocalDataSource,
    private val schedulerProvider: BaseSchedulerProvider
) {

    private var cacheIsDirty = false

    fun getSavingsGoals(): Observable<List<SavingsGoal>> {
        lateinit var goals: Observable<List<SavingsGoal>>
        if (cacheIsDirty) {
            goals = getGoalsFromRemoteDataSource()
        } else {
            val latch = CountDownLatch(1)
            var disposable: Disposable? = null
            disposable = localDataSource.getSavingsGoals()
                .observeOn(schedulerProvider.io())
                .doFinally {
                    latch.countDown()
                    disposable?.dispose()
                }.subscribe({
                    goals = Observable.create { emitter -> emitter.onNext(it) }
                }, { goals = getGoalsFromRemoteDataSource() })
            latch.await()
        }
        return goals
    }
}

如您所见，我正在使用 countDownLatch.await() 来确保将结果发送到订阅或错误块中。有没有比在使用 RxJava 时使用CountDownLatch 更好的解决方案？

【问题讨论】：

我不确定您想要实现什么。您想等到至少一个项目（无论是本地还是远程）发出？
是的，当结果发出时，在 finally 块中，我 countDown()

标签： android rx-java2 countdownlatch

【解决方案1】：

latch.await() 阻塞了线程，这有点违背了使用 RxJava 等异步 API 的全部意义。

RxJava 有像 onErrorResumeNext 这样的 API 来处理异常和 toObservable 来将 Single 结果转换为 Observable 结果。

此外，像这样的 RxJava 类型通常是冷的（在您订阅之前它们不会运行或计算出任何东西），所以我建议在订阅发生之前不要检查 cacheIsDirty。

我会选择类似的东西：

    fun getSavingsGoals(): Observable<List<SavingsGoal>> {
        return Observable
            .fromCallable { cacheIsDirty }
            .flatMap {
                if (it) {
                    getGoalsFromRemoteDataSource()
                } else {
                    localDataSource.getSavingsGoals()
                        .toObservable()
                        .onErrorResumeNext(getGoalsFromRemoteDataSource())
                }
            }
    }

顺便说一句，如果您已经在使用 Kotlin，我强烈推荐协程。然后你的异步代码最终会像常规顺序代码一样读取。

【讨论】：