RxJava中subscribe的Lambda内无法使用return，如何返回流中的Int值？

我编写了如下Kotlin代码：

private fun getTaskConditionByChapterIdAndTaskId(chapterId: Int, taskId: Int): Int {
    compositeDisposable.add(
        viewModel.getTaskConditionByChapterIdAndTaskId(chapterId, taskId)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe {
                return it
            }
    )
}

但编译器提示错误：'return' is not allowed here，请问如何从该Lambda中返回it值？

我的DAO代码如下：

@Query("SELECT isCompleted FROM Task WHERE chapterId = :chapterId AND taskId = :taskId")
abstract fun getTaskConditionByChapterIdAndTaskId(chapterId: Int, taskId: Int): Flowable<Int>

Repository实现类代码：

override fun getTaskConditionByChapterIdAndTaskId(chapterId: Int, taskId: Int): Flowable<Int> {
    return roomClientPlatform.courseDao().getTaskConditionByChapterIdAndTaskId(
        chapterId, taskId
    ).subscribeOn(Schedulers.io())
}

ViewModel代码：

fun getTaskConditionByChapterIdAndTaskId(chapterId: Int, taskId: Int): Flowable<Int> {
    return repository.getTaskConditionByChapterIdAndTaskId(chapterId, taskId)
}

解决方案 #

首先得明确：你当前的写法逻辑上走不通——RxJava的Flowable是异步流，外层函数是同步执行的，异步操作还没拿到结果，外层函数就已经执行完了，根本没法直接返回值。而且Lambda里的return默认会指向外层函数，这里语法也不允许这么用。

下面给几种可行的处理方式：

方案1：用回调接收结果 #

把外层函数改成带回调参数的形式，拿到结果后通过回调传递：

private fun getTaskConditionByChapterIdAndTaskId(chapterId: Int, taskId: Int, onResult: (Int) -> Unit) {
    compositeDisposable.add(
        viewModel.getTaskConditionByChapterIdAndTaskId(chapterId, taskId)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe { result ->
                onResult(result)
            }
    )
}

调用的时候这样用：

getTaskConditionByChapterIdAndTaskId(chapterId, taskId) { condition ->
    // 在这里处理拿到的任务状态值
}

方案2：改用LiveData（Android场景适配） #

如果是Android项目，推荐把Flowable转成LiveData，在UI层观察结果：

1. 先修改ViewModel代码（需要依赖androidx.lifecycle:lifecycle-reactivestreams）：

fun getTaskConditionByChapterIdAndTaskId(chapterId: Int, taskId: Int): LiveData<Int> {
    return repository.getTaskConditionByChapterIdAndTaskId(chapterId, taskId)
        .observeOn(AndroidSchedulers.mainThread())
        .toLiveData()
}

2. 在Activity/Fragment里观察LiveData：

viewModel.getTaskConditionByChapterIdAndTaskId(chapterId, taskId).observe(this) { condition ->
    // 处理任务状态
}

方案3：用Kotlin协程处理（推荐） #

如果项目已经用了协程，可以把Flowable转成挂起函数，在协程作用域里获取结果：

1. 定义挂起函数：

private suspend fun getTaskConditionByChapterIdAndTaskId(chapterId: Int, taskId: Int): Int {
    return viewModel.getTaskConditionByChapterIdAndTaskId(chapterId, taskId)
        .firstOrError() // 获取第一个结果，无结果时抛出错误
        .await() // 转成挂起函数，等待结果返回
}

2. 在协程作用域中调用（比如Activity里用lifecycleScope）：

lifecycleScope.launch {
    val condition = getTaskConditionByChapterIdAndTaskId(chapterId, taskId)
    // 拿到结果后处理逻辑
}

内容的提问来源于stack exchange，提问作者onesector