Kotlin Coroutines - 返回 Flow 的暂停函数永远运行答案

【问题标题】：Kotlin Coroutines - Suspend function returning a Flow runs foreverKotlin Coroutines - 返回 Flow 的暂停函数永远运行
【发布时间】：2023-04-04 13:49:01
【问题描述】：

我正在制作一个支持多个数据检索配置的网络存储库，因此我想将这些配置的逻辑分离为函数。

但是，我有一个配置，它以指定的时间间隔连续获取数据。当我将这些值发送到原始流时，一切都很好。但是当我将逻辑带入另一个函数并通过它返回另一个 Flow 时，它不再关心它的协程范围。即使在范围取消后，它仍会继续获取数据。

TLDR：当 currentCoroutineContext 用于控制其循环的终止时，返回流的暂停函数将永远运行。

我在这里做错了什么？这是我的代码的简化版本：

调用 viewmodels 函数的片段基本上调用 getData()

 lifecycleScope.launch {
            viewModel.getLatestDataList()
        }

存储库

suspend fun getData(config: MyConfig): Flow<List<Data>>
{
    return flow {

        when (config)
        {
            CONTINUOUS ->
            {
                //It worked fine when fetchContinuously was ingrained to here and emitted directly to the current flow
                //And now it keeps on running eternally
                fetchContinuously().collect { updatedList ->
                    emit(updatedList)
                }
            }
        }
    }
}


//Note logic of this function is greatly reduced to keep the focus on the problem
private suspend fun fetchContinuously(): Flow<List<Data>>
{
    return flow {
        while (currentCoroutineContext().isActive)
        {

            val updatedList = fetchDataListOverNetwork().await()

            if (updatedList != null)
            {
                emit(updatedList)
            }

            delay(refreshIntervalInMs)
        }

        Timber.i("Context is no longer active - terminating the continuous-fetch coroutine")
    }
}


private suspend fun fetchDataListOverNetwork(): Deferred<List<Data>?> =

    withContext(Dispatchers.IO) {

        return@withContext async {

            var list: List<Data>? = null

            try
            {
                val response = apiService.getDataList().execute()

                if (response.isSuccessful && response.body() != null)
                {
                    list = response.body()!!.list
                }
                else
                {
                    Timber.w("Failed to fetch data from the network database. Error body: ${response.errorBody()}, Response body: ${response.body()}")
                }
            }
            catch (e: Exception)
            {
                Timber.w("Exception while trying to fetch data from the network database. Stacktrace: ${e.printStackTrace()}")
            }
            finally
            {
                return@async list
            }
            list //IDE is not smart enough to realize we are already returning no matter what inside of the finally block; therefore, this needs to stay here
        }

    }

【问题讨论】：

标签： android multithreading kotlin kotlin-coroutines

【解决方案1】：

我不确定这是否可以解决您的问题，但您不需要有一个返回 Flow 的挂起函数。您传递的 lambda 本身就是一个挂起函数：

fun <T> flow(block: suspend FlowCollector<T>.() -> Unit): Flow<T> (source)

这是一个重复我正在使用的 (GraphQl) 查询（简化 - 没有类型参数）的流程示例：

    override fun query(query: Query,
                       updateIntervalMillis: Long): Flow<Result<T>> {
    return flow {
        // this ensures at least one query
        val result: Result<T> = execute(query)
        emit(result)

        while (coroutineContext[Job]?.isActive == true && updateIntervalMillis > 0) {
            delay(updateIntervalMillis)

            val otherResult: Result<T> = execute(query)
            emit(otherResult)
        }
    }
}

【讨论】：

【解决方案2】：

我不太擅长 Flow，但我认为问题在于您只延迟了 getData() 流，而不是同时延迟了它们。尝试添加：

suspend fun getData(config: MyConfig): Flow<List<Data>>
{
    return flow {

        when (config)
        {
            CONTINUOUS ->
            {
                fetchContinuously().collect { updatedList ->
                    emit(updatedList)
                    delay(refreshIntervalInMs)
                }
            }
        }
    }
}

记下delay(refreshIntervalInMs)。

【讨论】：