Kotlin Flow未获取到更新值问题排查求助

我实现了一个简易的ResettableFlow用来重置任意Flow的值，但和其他操作符配合使用时出现异常。调用reset()后，合并后的流始终获取不到重置值，"found!"一直没打印出来。在ViewModel中使用viewModelScope时也有类似问题，偶尔能看到值，疑似竞态条件导致。

以下是代码：

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

class ResettableFlow<T>(source: Flow<T>, private val defaultValue: T): Flow<T> {
    private val resetTrigger = MutableSharedFlow<T>()

    private val mergedFlow = merge(source, resetTrigger).onEach { println("ResettableFlow: onEach $it") }

    override suspend fun collect(collector: FlowCollector<T>) {
        mergedFlow.collect(collector)
    }

    fun reset() {
        resetTrigger.tryEmit(defaultValue)
    }
}

fun <T> Flow<T>.toResettable(defaultValue: T) = ResettableFlow(this, defaultValue)


fun main() {
    runBlocking {
        val state1 = MutableStateFlow(1)
        val resettable1 = state1.toResettable(0)

        val state2 = MutableStateFlow("a")
        val resettable2 = state2.toResettable("...")

        val combined = resettable1.combine(resettable2) { i, s -> "$i $s" }
            .onEach { println(it) }
            .stateIn(this, SharingStarted.WhileSubscribed(5000), "init")

        val job = combined.launchIn(this)

        state1.value = 10
        state2.value = "d"

        delay(1000)

        resettable1.reset()
        resettable2.reset()

        delay(1000)

        // 永远不会退出
        combined.first { it == "0 ..."}

        // 同样永远不会结束
        //while (combined.value != "0 ...") {
        //    delay(1000)
        //}

        println("found!")

        delay(1000)
        job.cancel()

        this.cancel()
    }   
}

问题根源 #

1. MutableSharedFlow默认行为导致事件丢失：你使用的MutableSharedFlow默认参数为replay = 0、extraBufferCapacity = 0，既不会缓存事件，也没有额外缓冲区。当订阅者无法立即接收时，tryEmit()会直接失败，导致重置值无法被发射出去。
2. 与stateIn共享流的订阅特性冲突：stateIn用WhileSubscribed启动策略时，新增订阅（比如调用first{}）会触发上游流重新发射当前值，但resetTrigger的一次性事件早已丢失，无法被新订阅捕获。
3. combine操作符的特性限制：combine需要两个流都发射新值才会生成组合结果，若两个重置事件的发射时机存在偏差，可能只触发一次组合（比如仅生成"0 d"或"10 ..."），而非预期的"0 ..."。

修复方案 #

方案1：优化ResettableFlow内部实现（推荐） #

将ResettableFlow改为基于StateFlow实现，确保始终持有当前最新值，从根源避免事件丢失：

class ResettableFlow<T>(source: Flow<T>, private val defaultValue: T) : Flow<T> {
    private val currentState = MutableStateFlow(defaultValue)

    init {
        // 订阅源流，自动更新当前状态
        CoroutineScope(Dispatchers.Default).launch {
            source.collect { currentState.value = it }
        }
    }

    override suspend fun collect(collector: FlowCollector<T>) {
        currentState.collect(collector)
    }

    fun reset() {
        currentState.value = defaultValue
    }
}

这个实现让ResettableFlow本质上是一个StateFlow，源流的更新会覆盖当前值，reset()直接将状态设回默认值，完全适配combine、stateIn等操作符的特性，不会出现事件丢失或竞态问题。

方案2：修改resetTrigger的参数 #

如果坚持使用merge的方式，调整MutableSharedFlow的参数确保事件能被正确发射：

private val resetTrigger = MutableSharedFlow<T>(replay = 1, extraBufferCapacity = 1)

replay = 1确保新订阅者能获取到最新的重置事件，extraBufferCapacity = 1避免tryEmit()因缓冲区不足失败。

内容的提问来源于stack exchange，提问作者frangulyan