kotlin - 如何将挂起功能转换为 RX Single(或 Completable)？

Question

我们正在将我们的项目从 RX 重构为 Kotlin 协程，但不是一次性的，因此我们需要在一段时间内让我们的项目同时使用这两者。

现在我们有很多像这样使用 RX single 作为返回类型的方法，因为它们是繁重、长时间运行的操作，例如 API 调用。

fun foo(): Single<String> // Heavy, long running opertation

我们希望它是这样的:

suspend fun foo(): String // The same heavy, long running opertation

在我们使用这种方法的地方，我们仍然想使用 RX。

我们一直在这样做:

foo()
    .subscribeOn(Schedulers.io())
    .map { ... }
    .subscribe { ... }

现在我应该如何将我的暂停乐趣转换为一个我可以使用的单曲？

这是个好主意吗？

Single.fromCallable {
    runBlocking {
        foo() // This is now a suspend fun
    }
}
    .subscribeOn(Schedulers.io())
    .map { ... }
    .subscribe { ... }

Answer 1

我没有给它太多时间来推理它，但我认为你创建它的方式可能会在挂起函数的持续时间内使用一个额外的线程。

有一个官方的 kotlinx-coroutines-rx3 库可以进行转换。 rxSingle功能在这里是相关的。在后台，它将以挂起的方式调用挂起函数，而不会阻塞额外的线程。

我自己没有使用过这个，所以你可能想自己测试一下，但我认为你不需要在此之后立即使用 subscribeOn，因为实际工作已经完成在协程中。您的 Foo 挂起函数，因为它是一个挂起函数，如果它执行阻塞 IO，应该已经在内部使用 Dispatchers.IO。

rxSingle { foo() }
    .map { ... }
    .subscribe { ... }