如何从不同的功能发出流量值？ Kotlin 协程

Question

我有一个流程：

val myflow = kotlinx.coroutines.flow.flow<Message>{}

并想用函数发出值：

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

但是编译器不允许我这样做，有什么办法可以解决这个问题吗？

Answer 1

您可以将 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
}

您可以通过运行以下 sn-p 来测试此代码。

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

Answer 2

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/) 的一部分发布。当需要在异步执行上下文中进行状态管理时，可以并且应该使用这些工具。

Answer 3

流是自包含的，一旦流中的块（lambda）被执行，流就结束了，你必须在里面进行操作并从那里发出它们。

这是类似的github issue，说：

Afaik Flow 被设计为自包含、可重放的冷流，因此来自其自身范围之外的发射不会成为合同的一部分。我认为您正在寻找的是一个频道。

恕我直言，您可能正在查看Channels，或者专门针对多个接收器的ConflatedBroadcastChannel。普通频道和广播频道的区别在于，多个接收者可以使用openSubscription 函数收听广播频道，该函数返回与广播频道关联的接收频道。

Answer 4

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

流程的初始化：

val myFlow = MutableSharedFlow<Message>()

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

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