使用stateIn将SharedFlow转为StateFlow后无法接收新值问题排查

问题重现 #

你尝试通过stateIn将MutableSharedFlow转换为StateFlow，但向源流发送新值后，StateFlow的value并未更新，仍然输出初始值。简化代码如下：

import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.GlobalScope
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.SharingStarted
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.stateIn
import kotlinx.coroutines.launch

fun main() = runBlocking {
    val sourceFlow = MutableSharedFlow<Int>()
    val stateFlow = sourceFlow.stateIn(GlobalScope, SharingStarted.Lazily, 0)

    val job = launch { stateFlow.collect() }

    sourceFlow.emit(99)
    println(stateFlow.value) // 输出0，而非预期的99

    job.cancel()
}

问题原因 #

1. SharingStarted.Lazily的启动时机：该策略会仅当第一个订阅者出现时才启动源流的收集。但代码中launch { stateFlow.collect() }是异步启动的，在你调用sourceFlow.emit(99)时，stateFlow可能还未完成对sourceFlow的订阅，导致这次emit的事件被直接错过。

2. GlobalScope的使用问题：GlobalScope的生命周期独立于runBlocking作用域，会导致stateFlow的收集协程启动延迟，无法及时响应源流的事件。

3. 流的异步特性：即使订阅启动，emit是挂起函数，但StateFlow的value更新需要完成事件收集后才会生效，若在emit后立即执行println，可能此时更新还未完成。

解决方案 #

方案1：改用SharingStarted.Eagerly启动策略 #

该策略会立即启动源流的收集，无需等待订阅者，确保源流的事件不会被错过。同时替换GlobalScope为当前runBlocking的作用域，保证协程调度的及时性：

import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.SharingStarted
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.stateIn
import kotlinx.coroutines.launch

fun main() = runBlocking {
    val sourceFlow = MutableSharedFlow<Int>()
    // 使用当前作用域 + Eagerly启动策略
    val stateFlow = sourceFlow.stateIn(this, SharingStarted.Eagerly, 0)

    val job = launch { stateFlow.collect() }

    sourceFlow.emit(99)
    println(stateFlow.value) // 输出99

    job.cancel()
}

方案2：确保订阅启动后再发送事件 #

若要保留SharingStarted.Lazily，可以在启动订阅后，通过yield()让协程调度器完成订阅初始化，确保stateFlow已经开始收集源流事件：

import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.SharingStarted
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.stateIn
import kotlinx.coroutines.launch
import kotlinx.coroutines.yield

fun main() = runBlocking {
    val sourceFlow = MutableSharedFlow<Int>()
    val stateFlow = sourceFlow.stateIn(this, SharingStarted.Lazily, 0)

    val job = launch { stateFlow.collect() }
    yield() // 让协程调度，确保stateFlow完成对sourceFlow的订阅

    sourceFlow.emit(99)
    println(stateFlow.value) // 输出99

    job.cancel()
}

补充说明 #

如果需要保留历史事件，可以在创建MutableSharedFlow时指定replay参数（比如MutableSharedFlow<Int>(replay = 1)），这样即使订阅启动在emit之后，也能收到最近的一次事件。

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