Kotlin Coroutines：Hot Flow 掌握线程

Question

我正在尝试使用Flow 模拟BehaviourSubject。

以下示例中的代码无法正常运行，因为collect {} 不允许协程继续：

@FlowPreview
@ExperimentalCoroutinesApi
fun main() = runBlocking {
    val scope = CoroutineScope(Dispatchers.IO)
    val channel = MutableStateFlow(0)
    channel.value++
    val flow = channel.broadcastIn(scope).asFlow()
    channel.value++
    flow.collect { println("First: $it") } // Code will stop executing here ...

    flow.collect { println("Second: $it") } // Will not execute
    channel.value++  // Wil not execute

    return@runBlocking

}

输出：

First: 2
// process not exited

我不认为接受回调也应该持有线程。 问题出在哪里？

这样的问题不会发生在感冒Flow：

fun main() = runBlocking {
    val data = listOf(1, 2, 3, 4, 5).asFlow()
    data.collect { println(it) }
    println("Done")
}

输出：

1
2
3
4
5
Done

我错过了什么吗？

科特林：1.4.10
协程：1.3.9

Answer 1

在这两种情况下Flow 的行为相同，您错过的是Flow.collect 暂停直到流程完成。

在冷流的情况下，它显然在发出 5 个值后结束，因此它完成了执行。同时，在热流程中，流程的完成取决于在collect 之后执行的代码会返回，这是一个典型的死锁。

如果您在launch 块中异步执行collect，它在两种情况下的行为都相同。但由于 StateFlow 无法按设计“结束”，它最终会永远等待不可能发生的终止。