您可以使用***StateFlow***或***SharedFlow***接口来实现此类用例，下面是***StateFlow***的使用示例代码。

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

val chatFlow = MutableStateFlow<String>("")
    
fun main() = runBlocking {

    // Observe values
    val job = launch {
        chatFlow.collect {
            print("$it ")
        }
    }

    // Change values
    arrayOf("Hey", "Hi", "Hello").forEach {
        delay(100)
        sendMessage(it)
    }

    delay(1000)
    
    // Cancel running job
    job.cancel()
    job.join()
}

suspend fun sendMessage(message: String) {
    chatFlow.value = message
}

您可以通过运行下面的代码段来测试此代码。

<iframe src="https://pl.kotl.in/DUBDfUnX3" style="width:600px;"></iframe>

The answer of Animesh Sahu非常正确。您也可以将Channel作为流返回（参见BroadcastChannel上的consumeAsFlow或asFlow）。
但是Kotlin团队目前正在开发一个名为StateFlow的东西，它在某种程度上是为了实现类似的行为，尽管它何时准备就绪还不清楚。
编辑：StateFlow和SharedFlow已经作为稳定API（https://blog.jetbrains.com/kotlin/2020/10/kotlinx-coroutines-1-4-0-introducing-stateflow-and-sharedflow/）的一部分发布。当在异步执行上下文中需要状态管理时，可以并且应该使用这些工具。

流是自包含的，一旦流中的块（lambda）被执行，流就结束了，你必须在里面做操作，然后从那里发出它们。
下面是类似的github issue，说：
Afaik流被设计成一个独立的，可重放的，冷流，所以从它自己的范围之外的排放不会是合同的一部分。
恕我直言，你可能正在看频道，或者特别是多个接收器的ConflatedBroadcastChannel。普通频道和广播频道之间的区别在于，多个接收器可以使用openSubscription函数收听广播频道，该函数返回与BroadcastChannel相关联的ReceiveChannel。

使用SharedStateFlow它有你需要的一切。

初始化您的流程：

val myFlow = MutableSharedFlow<Message>()

现在它应该像你之前尝试的那样工作：

override suspend fun sendMessage(chat: Chat, message: Message) {
    myFlow.emit(message)
}