我们通常这样定义一个stateflow

val weatherState: StateFlow<Result<WeatherData>> = _weatherState

我们跟踪一下其源码

public interface StateFlow<out T> : SharedFlow<T> {
    /**
     * The current value of this state flow.
     */
    public val value: T
}

哎呀，只是一个接口，我们右键 go to implementations

private class StateFlowImpl<T>(
    initialState: Any // T | NULL
) : AbstractSharedFlow<StateFlowSlot>(), MutableStateFlow<T>, CancellableFlow<T>, FusibleFlow<T> {
    private val _state = atomic(initialState)

这是一个很申请的定义哦，ai问出来的结果是

是 Kotlin 中使用 KotlinX Atomic（来自 kotlinx.atomicfu 库）或类似原子操作库（如 kotlinx.coroutines 中的 MutableStateFlow 配合原子性思维，但更可能指 atomicfu）来创建一个线程安全的可变状态变量的典型写法。

我们不必细究，总之就是保证线程安全。

我们通常抓取流的做法是

viewModel.weatherState.collect { result ->
    when (result) {
        is Result.Loading -> {
            // 显示加载动画
            //progressBar.visibility = View.VISIBLE
        }
        is Result.Success<WeatherData> -> {
            // 隐藏加载，展示数据
            //progressBar.visibility = View.GONE
            val weatherData = result.data
            //updateUi(weatherData)
        }
        //统一处理错误
        is Result.Failure -> {
            //Toast.makeText(this@MainActivity,result.getErrorMsgOrNull(), Toast.LENGTH_SHORT).show()

            result.handleFailure(this@MainActivity);
        }
    }
}

我们看看collect的做了什么

override suspend fun collect(collector: FlowCollector<T>): Nothing {
    val slot = allocateSlot()
    try {
        if (collector is SubscribedFlowCollector) collector.onSubscription()
        val collectorJob = currentCoroutineContext()[Job]
        var oldState: Any? = null // previously emitted T!! | NULL (null -- nothing emitted yet)
        // The loop is arranged so that it starts delivering current value without waiting first
        while (true) {
            // Here the coroutine could have waited for a while to be dispatched,
            // so we use the most recent state here to ensure the best possible conflation of stale values
            val newState = _state.value
            // always check for cancellation
            collectorJob?.ensureActive()
            // Conflate value emissions using equality
            if (oldState == null || oldState != newState) {
                collector.emit(NULL.unbox(newState))
                oldState = newState
            }
            // Note: if awaitPending is cancelled, then it bails out of this loop and calls freeSlot
            if (!slot.takePending()) { // try fast-path without suspending first
                slot.awaitPending() // only suspend for new values when needed
            }
        }
    } finally {
        freeSlot(slot)
    }
}

参数是一个函数式接口，用于调用我们定义的表达式，suspend的修饰符说明这个函数必须在协程里面执行。while循环中还顺便校验了值是否真的有变化通过emit调用我们定义的表达式，emit是一个挂起函数。

最后一段是启动一个挂起函数，挂起当前的while循环，如果有新数据过来，通知挂起函数完成可以继续执行循环获取数据。

我们来看看

public fun <T> MutableStateFlow(value: T): MutableStateFlow<T> = StateFlowImpl(value ?: NULL)

构造函数的参数不能为空，所以必须有初始值

val slot = allocateSlot()

是要为每一个观察者创建一个插槽，用于挂起和恢复