Kotlin StateFlow 源码解析

前言

在之前的文章已经分析了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>：表示收集器已挂起在等待上游数据