前言
在之前的文章已经分析了SharedFlow原理，在这个基础上再来分析StateFlow就简单多了。Kotlin SharedFlow 源码解析
StateFlow使用与特性分析
如下代码，创建了一个StateFlow对象，必须有个初始值，这里设置为1。发射了3条数据，但是收集者只收到value为3的这条数据。
runBlocking {
val stateFlow = MutableStateFlow(1)
stateFlow.value = 2
stateFlow.value = 3
stateFlow.value = 3
launch {
stateFlow.collect {
delay(3000)
print("collect2 :$it")
}
}
}
StateFlow最新的收集者只能收到最新的1条数据；如果数据相同，不会收到重复的数据。StateFlow可以看作如下定义的ShareFlow。
MutableSharedFlow<Int>(replay = 1, extraBufferCapacity = 0, onBufferOverflow = BufferOverflow.DROP_OLDEST)
StateFlow源码分析
初始化
public fun <T> MutableStateFlow(value: T): MutableStateFlow<T> = StateFlowImpl(value ?: NULL)
StateFlow并没有像SharedFlow那样采用缓存数组，只是用atomic引用类型的状态值，来保存最新的值。
private class StateFlowImpl<T>(
initialState: Any
) : AbstractSharedFlow<StateFlowSlot>(), MutableStateFlow<T>, CancellableFlow<T>, FusibleFlow<T> {
private val _state = atomic(initialState) // 保存状态值
private var sequence = 0 // 状态值更新的序列号
@Suppress("UNCHECKED_CAST")
public override var value: T
get() = NULL.unbox(_state.value)
set(value) { updateState(null, value ?: NULL) } //CAS方式更新数据
override fun compareAndSet(expect: T, update: T): Boolean =
updateState(expect ?: NULL, update ?: NULL)
}
在StateFlow中一般有3种方式发射数据，emit、setValue,compareAndSet,但是无论哪种方式最终都是调用updateState方法，所以我们直接看这个方法就好。
发射数据
private fun updateState(expectedState: Any?, newState: Any): Boolean {
var curSequence = 0
var curSlots: Array<StateFlowSlot?>? = this.slots // benign race, we will not use it
synchronized(this) {
val oldState = _state.value
//看过CAS原理的同学能很好理解这句话的意思，内存值和预期值不同，直接返回false；说明是线程安全的，内存值和预期值相同，才会更新数据
if (expectedState != null && oldState != expectedState) return false // CAS support
//如果2次发射的值相同，则丢弃，也就是做了防抖
if (oldState == newState) return true // Don't do anything if value is not changing, but CAS -> true
//CAS方式更新值
_state.value = newState
curSequence = sequence
//状态更新序号变为偶数
if (curSequence and 1 == 0) { // even sequence means quiescent state flow (no ongoing update)
curSequence++ // +1，变成基数
sequence = curSequence
} else {
// 序号是奇数，说明该值已经更新过，立即返回
sequence = curSequence + 2 // +2，可以保持继续奇数
return true // updated
}
curSlots = slots // read current reference to collectors under lock
}
这里大家暂时只需要关心2件事情，StateFlow通过CAS的方式保存和更新数据；如果前后2次发射的数据相同，会丢弃后一次，也就是说相同数据的发射，做了防抖过滤。
收集数据
override suspend fun collect(collector: FlowCollector<T>) {
val slot = allocateSlot() //分配槽位，和ShareFlow是一样的
try {
if (collector is SubscribedFlowCollector) collector.onSubscription()
val collectorJob = currentCoroutineContext()[Job]
var oldState: Any? = null
// 也是和ShareFlow一样的方式，开启了一个死循环获取数据
while (true) {
val newState = _state.value
// check协程是否取消
collectorJob?.ensureActive()
// 空安全检查，如果前后2次发射数据不同，就会调用FlowCollector的emit方法，
//也就是会执行到collect{ }中的代码块
if (oldState == null || oldState != newState) {
collector.emit(NULL.unbox(newState))
oldState = newState
}
// 修改状态，如果之前不是PENGDING状态
if (!slot.takePending()) {
// 则挂起等待新数据更新
slot.awaitPending()
}
}
} finally {
// 释放已分配的StateFlowSlot类型的对象
freeSlot(slot)
}
}
这个方法，我们暂时也不需要关注太多。只需要知道也是和ShareFlow一样，开启了死循环来获取数据，数据的获取就是前面通过atomic定义的_state,所以state数据的收发都是通过CAS完成的。获取到数据之后，会调用FlowCollector的emit方法，也就是会执行到collect{ }中的代码块，这一点在之前文章介绍过。一个案例让你秒懂kotlin flow原理
收集者(订阅者)的管理
同ShareFlow，StateFlow也是用到了AbstractSharedFlow类，因此二者订阅者管理的逻辑大致相同。区别是ShareFlow订阅者数组中存储的对象类为SharedFlowSlot,而StateFlow中存储的对象为StateFlowSlot。
private val NONE = Symbol("NONE")
@SharedImmutable
private val PENDING = Symbol("PENDING")
// StateFlow 的每一个收集者都会分配一个卡槽
private class StateFlowSlot : AbstractSharedFlowSlot<StateFlowImpl<*>>() {
//用于存储StateFlowSlot的状态，一共有4个这状态，初始值为null
private val _state = atomic<Any?>(null)
override fun allocateLocked(flow: StateFlowImpl<*>): Boolean {
// 如果_state的值不为null，表示槽位已被占用
if (_state.value != null) return false // not free
//_state的值为null，表示槽位可用，将_state的值设置为NONE，表示已被分配
_state.value = NONE // allocated
return true
}
override fun freeLocked(flow: StateFlowImpl<*>): Array<Continuation<Unit>?> {
//释放槽位
//将_state的值设置为null，表示槽位现在可用。返回一个空的Continuation数组，表示没有更多的操作需要执行
_state.value = null // free now
return EMPTY_RESUMES // nothing more to do
}
StateFlowSlot类型的对象共有四种状态：
null：表示已经空闲释放，可以分配给消费者收集器NONE： 表示已经分配给消费者接收器，但既没有挂起，也没有在处理当前的数据。PENDING：表示表示上游已更新新值，待发送给收集器。CancellableContinuationImpl<Unit>：表示收集器已挂起在等待上游数据
