如何在等待MutableStateFlow新值时挂起Kotlin协程？

问题背景 #

现有代码使用MutableStateFlow维护状态，其中一个协程每秒更新状态值，另一个协程通过轮询方式读取状态：

val state = MutableStateFlow<Int>(1)

launch(Dispatchers.Default) {
    var counter = 1
    while (true) {
        state.emit(counter++)
        delay(1000)
    }
}

launch(Dispatchers.Default) {
    while (state.value < 1000) {
        print(state.value)
        delay(250)
    }
}

当前输出为111122223333444...，期望输出为1234...。

实际业务场景中，需要在发送数据前等待连接状态变为IDLE，目前采用的轮询方式性能较差：

while (state.value != ConnectionState.IDLE) { delay(100) }

期望实现逻辑：

• 先检查当前状态值，若为目标值则直接继续执行
• 若不是目标值，挂起协程直到状态变化，再次检查，不符合则继续挂起等待

虽然可以通过Mutex实现类似逻辑，但希望使用官方/标准方案。

官方标准解决方案 #

利用StateFlow的collect方法（Flow核心特性），可以实现高效的状态监听，避免轮询的资源浪费。

解决基础示例问题 #

将轮询的协程改为监听StateFlow数据流，每次状态更新时执行逻辑：

val state = MutableStateFlow<Int>(1)

launch(Dispatchers.Default) {
    var counter = 1
    while (true) {
        state.emit(counter++)
        delay(1000)
    }
}

launch(Dispatchers.Default) {
    state.collect { value ->
        if (value >= 1000) return@collect
        print(value)
    }
}

此时输出为1234...，完全符合期望。

通用等待目标状态的扩展函数 #

针对业务场景中“等待状态满足指定条件”的需求，可封装扩展函数复用逻辑：

import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.collect

suspend fun <T> StateFlow<T>.await(predicate: (T) -> Boolean): T {
    // 先检查当前值，满足条件直接返回，避免不必要挂起
    val current = value
    if (predicate(current)) {
        return current
    }
    // 挂起监听流，直到值满足条件
    return collect { value ->
        if (predicate(value)) {
            return@collect value
        }
    }
}

业务场景中的使用示例 #

// 假设ConnectionState是自定义状态枚举
enum class ConnectionState { IDLE, CONNECTED, DISCONNECTED }

val connectionState = MutableStateFlow(ConnectionState.CONNECTED)

// 发送数据的协程逻辑
launch(Dispatchers.IO) {
    // 等待状态变为IDLE
    connectionState.await { it == ConnectionState.IDLE }
    // 状态满足后执行写入数据操作
    writeDataToSocket()
}

方案优势 #

• 完全基于Kotlin官方Flow API，符合协程设计规范
• 无不必要的轮询和延迟，仅在状态变化时唤醒协程，性能更优
• 代码简洁可复用，适配各种状态监听场景

内容的提问来源于stack exchange，提问作者Jan Vladimir Mostert