RxJava2:对成功发送到REST api的数据执行操作

Question

我有这个特定的问题，我找不到一个干净的解决方案。

Observable.interval(1, TimeUnit.SECONDS)
.flatMap { constructData() }
.subscribe { data ->
api.syncData(data)
.repeat()
.subscribe { response ->
deleteSyncedData(data)
}
}

因此，正如您从代码中看到的那样，我需要构造一些数据包以将其发送到后端，并且在正确发送后，我将从本地存储系统中删除它。现在它正在工作，但看起来有点像回调地狱 - 如果我想将响应与数据相结合然后发出请求怎么办？

有没有人看到这种操作的更好解决方案？

此外，我想知道如何在这些操作之间的调度程序之间切换？我想在Schedulers.computation()上执行constructData()和deleteSyncedData()，但在Schedulers.io()上执行api.syncData(data)。

Answer 1

来自 SubscribeOn 的 Rx 文档：

您不能在chain中多次使用subscribeOn()。从技术上讲，您可以这样做，但这不会产生任何额外的影响。在代码中，如果使用subscribeOn()链接两个不同的Schedulers，则只有对源observable关闭的那个将生效，而没有其他效果。

但是您可以使用observeOn().您可以使用observeOn()在多个Schedulers之间切换，然后最终observe结果MainThread

例：

Observable.interval(1, TimeUnit.SECONDS)
.observeOn(Schedulers.computation())
.flatMap { constructData() }
.observeOn(Schedulers.io())
.flatMap { data -> api.syncData(data).map { data } }
.observeOn(Schedulers.computation())
.flatMap { data -> deleteSyncedData(data) }
.observeOn(AndroidSchedulers.mainThread())
.subscribe { data ->
}

但在我看来，这也很好。

Observable.interval(1, TimeUnit.SECONDS)
.flatMap { constructData() }
.subscribe { data ->
api.syncData(data)
.repeat()
.subscribe { response ->
deleteSyncedData(data)
}
}