RxJava2 & Retrofit：如何获取数据页答案

【问题标题】：RxJava2 & Retrofit: How to get pages of dataRxJava2 & Retrofit：如何获取数据页
【发布时间】：2018-11-05 07:51:21
【问题描述】：

目标：我想重复调用一个返回分页数据的改造服务 (GET)，直到我用完它的页面。从第 0 页转到第 n 页。

首先，我已经查看了 these two 的答案。第一个确实有效，但我不太喜欢递归解决方案，因为它可能导致堆栈溢出。当您尝试使用调度程序时，第二个失败。

这是第二个的示例：

Observable.range(0, 5/*Integer.MAX_VALUE*/) // generates page values
    .subscribeOn(Schedulers.io())           // need this to prevent UI hanging
    // gamesService uses Schedulers.io() by default
    .flatMapSingle { page -> gamesService.getGames(page) }
    .takeWhile { games -> games.isNotEmpty() } // games is a List<Game>
    .subscribe(
        { games -> db.insertAll(games) },
        { Logger.e(TAG, it, "Error getting daily games: ${it.message}") }
    )

我期望这样做是在gamesService.getGames(page) 返回一个空列表的那一刻停止。相反，它会继续命中端点的次数不确定，并增加页面值。我在Single.just(intVal) 的单元测试中进行了一些实验，并确定问题似乎是我的服务在Schedulers.io() 上自动订阅的事实。这就是我定义改造服务的方式：

private inline fun <reified T> createService(okClient: OkHttpClient): T {
    val rxAdapter = RxJava2CallAdapterFactory.createWithScheduler(Schedulers.io())
    val retrofit = Retrofit.Builder()
        .baseUrl(config.apiEndpoint.endpoint())
        .client(okClient)
        .addCallAdapterFactory(rxAdapter)
        .addConverterFactory(moshiConverterFactory())
        .build()

    return retrofit.create(T::class.java)
}

不在这里使用createWithScheduler()真的不是一个选项。

这是我尝试的另一个想法：

val atomic = AtomicInteger(0)
Observable.generate<Int> { it.onNext(atomic.getAndIncrement()) }
    .subscribeOn(Schedulers.io())
    .flatMapSingle { page -> gamesService.getGames(page) }
    .takeWhile { games -> games.isNotEmpty() }
    .subscribe(
        { games -> dailyGamesDao.insertAll(games) },
        { Logger.e(TAG, it, "Error getting daily games: ${it.message}") }
    )

这是另一种情况，直到我介绍了Scheduler，它才按预期工作。当takeWhile 发现一个空列表时，生成器会生成太多方式值。

我也尝试过各种concat（concatWith、concatMap 等）。

在这一点上，我真的只是想找人来帮助我纠正我对 RxJava 运算符的明显（对他们）和完全基本的误解。

【问题讨论】：

标签： kotlin retrofit rx-java

【解决方案1】：

我找到了部分解决方案。（如果我找到我的“最终”解决方案，我可能会在以后编辑这个答案。）

tl;dr 我应该将我的Singles 转换为Observables 并使用带有maxConcurrency 参数的flatMap 重载。例如：

Observable.range(0, SOME_SUFFICIENTLY_LARGE_NUMBER)
    .subscribeOn(Schedulers.io())
    .flatMap({ page -> gamesService.getGames(page).toObservable }, 1 /* maxConcurrency */)
    .takeWhile { games -> games.isNotEmpty() }
    .subscribe(
        { games -> dailyGamesDao.insertAll(games) },
        { Logger.e(TAG, it, "Error getting daily games: ${it.message}") }
    )

基本上就是这样。通过将并发线程的数量限制为 1，我现在有了我所寻求的“一个接一个”的行为。对此我唯一不喜欢的是，我想这是一个小小的抱怨，是我的基础 Observable.range() 仍然可以发出 lot 的值 - 比以往任何时候都更多地被下游Singles/Observables.

PS：我之前找不到这个解决方案的一个原因是我使用的是 RxJava 2.1.9。当我将它推送到 2.1.14 时，我可以访问新的重载。哦，好吧。

【讨论】：