前言
在之前的文章已經(jīng)分析了SharedFlow原理云石，在這個(gè)基礎(chǔ)上再來分析StateFlow就簡單多了。Kotlin SharedFlow 源碼解析
StateFlow使用與特性分析
如下代碼，創(chuàng)建了一個(gè)StateFlow對象，必須有個(gè)初始值，這里設(shè)置為1。發(fā)射了3條數(shù)據(jù)樟遣，但是收集者只收到value為3的這條數(shù)據(jù)。
runBlocking {
val stateFlow = MutableStateFlow(1)
stateFlow.value = 2
stateFlow.value = 3
stateFlow.value = 3
launch {
stateFlow.collect {
delay(3000)
print("collect2 :$it")
}
}
}
StateFlow最新的收集者只能收到最新的1條數(shù)據(jù)身笤；如果數(shù)據(jù)相同豹悬，不會收到重復(fù)的數(shù)據(jù)。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那樣采用緩存數(shù)組屿衅，只是用atomic引用類型的狀態(tài)值，來保存最新的值莹弊。
private class StateFlowImpl<T>(
initialState: Any
) : AbstractSharedFlow<StateFlowSlot>(), MutableStateFlow<T>, CancellableFlow<T>, FusibleFlow<T> {
private val _state = atomic(initialState) // 保存狀態(tài)值
private var sequence = 0 // 狀態(tài)值更新的序列號
@Suppress("UNCHECKED_CAST")
public override var value: T
get() = NULL.unbox(_state.value)
set(value) { updateState(null, value ?: NULL) } //CAS方式更新數(shù)據(jù)
override fun compareAndSet(expect: T, update: T): Boolean =
updateState(expect ?: NULL, update ?: NULL)
}
在StateFlow中一般有3種方式發(fā)射數(shù)據(jù)涤久，emit涡尘、setValue,compareAndSet,但是無論哪種方式最終都是調(diào)用updateState方法，所以我們直接看這個(gè)方法就好响迂。
發(fā)射數(shù)據(jù)
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原理的同學(xué)能很好理解這句話的意思考抄，內(nèi)存值和預(yù)期值不同，直接返回false蔗彤；說明是線程安全的川梅，內(nèi)存值和預(yù)期值相同，才會更新數(shù)據(jù)
if (expectedState != null && oldState != expectedState) return false // CAS support
//如果2次發(fā)射的值相同然遏，則丟棄贫途，也就是做了防抖
if (oldState == newState) return true // Don't do anything if value is not changing, but CAS -> true
//CAS方式更新值
_state.value = newState
curSequence = sequence
//狀態(tài)更新序號變?yōu)榕紨?shù)
if (curSequence and 1 == 0) { // even sequence means quiescent state flow (no ongoing update)
curSequence++ // +1，變成基數(shù)
sequence = curSequence
} else {
// 序號是奇數(shù)待侵，說明該值已經(jīng)更新過丢早，立即返回
sequence = curSequence + 2 // +2，可以保持繼續(xù)奇數(shù)
return true // updated
}
curSlots = slots // read current reference to collectors under lock
}
這里大家暫時(shí)只需要關(guān)心2件事情秧倾，StateFlow通過CAS的方式保存和更新數(shù)據(jù)怨酝；如果前后2次發(fā)射的數(shù)據(jù)相同，會丟棄后一次那先，也就是說相同數(shù)據(jù)的發(fā)射农猬，做了防抖過濾。
收集數(shù)據(jù)
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一樣的方式斤葱，開啟了一個(gè)死循環(huán)獲取數(shù)據(jù)
while (true) {
val newState = _state.value
// check協(xié)程是否取消
collectorJob?.ensureActive()
// 空安全檢查，如果前后2次發(fā)射數(shù)據(jù)不同勋又，就會調(diào)用FlowCollector的emit方法苦掘，
//也就是會執(zhí)行到collect{ }中的代碼塊
if (oldState == null || oldState != newState) {
collector.emit(NULL.unbox(newState))
oldState = newState
}
// 修改狀態(tài)换帜，如果之前不是PENGDING狀態(tài)
if (!slot.takePending()) {
// 則掛起等待新數(shù)據(jù)更新
slot.awaitPending()
}
}
} finally {
// 釋放已分配的StateFlowSlot類型的對象
freeSlot(slot)
}
}
這個(gè)方法楔壤，我們暫時(shí)也不需要關(guān)注太多。只需要知道也是和ShareFlow一樣惯驼，開啟了死循環(huán)來獲取數(shù)據(jù)蹲嚣，數(shù)據(jù)的獲取就是前面通過atomic定義的_state,所以state數(shù)據(jù)的收發(fā)都是通過CAS完成的。獲取到數(shù)據(jù)之后祟牲，會調(diào)用FlowCollector的emit方法隙畜，也就是會執(zhí)行到collect{ }中的代碼塊，這一點(diǎn)在之前文章介紹過说贝。一個(gè)案例讓你秒懂kotlin flow原理
收集者(訂閱者)的管理
同ShareFlow议惰，StateFlow也是用到了AbstractSharedFlow類，因此二者訂閱者管理的邏輯大致相同乡恕。區(qū)別是ShareFlow訂閱者數(shù)組中存儲的對象類為SharedFlowSlot,而StateFlow中存儲的對象為StateFlowSlot言询。
private val NONE = Symbol("NONE")
@SharedImmutable
private val PENDING = Symbol("PENDING")
// StateFlow 的每一個(gè)收集者都會分配一個(gè)卡槽
private class StateFlowSlot : AbstractSharedFlowSlot<StateFlowImpl<*>>() {
//用于存儲StateFlowSlot的狀態(tài)俯萎，一共有4個(gè)這狀態(tài)，初始值為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的值設(shè)置為NONE辆憔，表示已被分配
_state.value = NONE // allocated
return true
}
override fun freeLocked(flow: StateFlowImpl<*>): Array<Continuation<Unit>?> {
//釋放槽位
//將_state的值設(shè)置為null撇眯，表示槽位現(xiàn)在可用。返回一個(gè)空的Continuation數(shù)組虱咧，表示沒有更多的操作需要執(zhí)行
_state.value = null // free now
return EMPTY_RESUMES // nothing more to do
}
StateFlowSlot類型的對象共有四種狀態(tài)：
null：表示已經(jīng)空閑釋放熊榛，可以分配給消費(fèi)者收集器NONE： 表示已經(jīng)分配給消費(fèi)者接收器，但既沒有掛起腕巡，也沒有在處理當(dāng)前的數(shù)據(jù)来候。PENDING：表示表示上游已更新新值，待發(fā)送給收集器逸雹。CancellableContinuationImpl<Unit>：表示收集器已掛起在等待上游數(shù)據(jù)
