0 前言

本片文章，从以下几个方向分析下 StateFlow 和 SharedFlow 原理：

• 冷流和热流的区别；
• StateFlow 和 SharedFlow 的区别；
• StateFlow 和 SharedFlow 的源码分析；

1 冷流和热流

下面是一个表格，通过对比能够看出冷流和热流的差异：

特性维度	冷流 (Cold Flow)	热流 (Hot Flow)
数据生产	只有存在 Collector 的时候才会产生数据	主动生产，不依赖于 Collector
数据共享	每个收集者获得独立数据流，比例 1: 1	所有收集者共享相同数据，比例 1 : N
重放机制	无，每次收集都重新生产	SharedFlow 可配置重放缓存（粘性数据）
内存占用	低，无缓存	较高，有缓冲区 / 重放缓存
生命周期	跟随收集者生命周期	独立生命周期–需手动管理
取消影响	停止数据生产	不影响数据生产
背压处理	支持多种策略（buffer、conflate 等）	通过缓冲区配置处理
多订阅者	每个订阅者独立执行	数据共享
初始值要求	不需要	StateFlow 需要，SharedFlow 不需要
数据完整性	保证完整数据流	可能丢失早期数据

2 SharedFlow

SharedFlow 是一个有缓存区 buffer 的热流，发射数据不依赖收集者，数据可被多个收集者共享。

看了简单的例子：

fun main() = runBlocking {
    //【1】创建一个 sharedFlow
    val sharedFlow = MutableSharedFlow<Int>(replay = 1)
 
    //【2】模拟发射协程
    launch {
        for (i in 1..10) {
            sharedFlow.emit(i)
            delay(500)
        }
    }
 
    //【3】模拟多个订阅协程
    launch {
        delay(800) // 延时后开始订阅，能收到之前 replay 的值
        sharedFlow.collect { value ->
            println("Subscriber 1 received: $value")
        }
    }
 
    launch {
        delay(1500)
        sharedFlow.collect { value ->
            println("Subscriber 2 received: $value")
        }
    }
}

接口描述

SharedFlow

基础接口，提供了 replayCache 和 collect 接口：

public interface SharedFlow<out T> : Flow<T> {
  
    public val replayCache: List<T>

    override suspend fun collect(collector: FlowCollector<T>): Nothing
}

MutableSharedFlow

MutableSharedFlow 则是对 SharedFlow 的扩展，同时实现了 FlowCollector，所以它既是一个 Flow 又是一个 FlowCollector：

1. 作为消费者：实现了 FlowCollector，可以接收上游的数据，再发射给下游。
2. 作为生产者：实现了 Flow，可以产生数据，通过收集器发送。

/**
 * MutableSharedFlow 是一种可变的热数据流，它可以同时作为数据源和数据接收器。
 * 与冷流（cold flow）不同，它的生命周期不依赖于收集者，多个收集者可以共享同一数据流。
 * 
 * 核心特性：
 * 1. 线程安全 - 所有方法都支持并发调用
 * 2. 背压处理 - 通过缓冲区策略控制数据流速，防止生产者过快导致消费者无法处理
 * 3. 数据重播 - 新订阅者可以获取之前发送的数据（取决于配置）
 * 4. 多订阅者支持 - 允许多个收集者同时监听数据流
 */
public interface MutableSharedFlow<T> : SharedFlow<T>, FlowCollector<T> {
    
    override suspend fun emit(value: T)
    public fun tryEmit(value: T): Boolean

    public val subscriptionCount: StateFlow<Int>
      
    @ExperimentalCoroutinesApi
    public fun resetReplayCache()
}

MutableSharedFlow 构造函数

构造一个 SharedFlow 实例：

@Suppress("FunctionName", "UNCHECKED_CAST")
public fun <T> MutableSharedFlow(
    replay: Int = 0, // 粘性数据的长度
    extraBufferCapacity: Int = 0, // 扩展容量
    onBufferOverflow: BufferOverflow = BufferOverflow.SUSPEND // 缓冲区溢出后的策略，默认是 suspend；
): MutableSharedFlow<T> {
    require(replay >= 0) { "replay cannot be negative, but was $replay" }
    require(extraBufferCapacity >= 0) { "extraBufferCapacity cannot be negative, but was $extraBufferCapacity" }
    require(replay > 0 || extraBufferCapacity > 0 || onBufferOverflow == BufferOverflow.SUSPEND) {
        "replay or extraBufferCapacity must be positive with non-default onBufferOverflow strategy $onBufferOverflow"
    }
  
    //【1】对容量做最终的调整;
    val bufferCapacity0 = replay + extraBufferCapacity
    val bufferCapacity = if (bufferCapacity0 < 0) Int.MAX_VALUE else bufferCapacity0 // coerce to MAX_VALUE on overflow
      
    //【2】创建 SharedFlowImpl 对象;
    return SharedFlowImpl(replay, bufferCapacity, onBufferOverflow)
}

创建了一个 SharedFlowImpl 对象，最终交给它来处理。

SharedFlowImpl

SharedFlowImpl 实现了 FusibleFlow 接口，所以也可以压缩混合路径，提升性能。

SynchronizedObject 是内部的一个同步对象。

//【1】SharedFlowImpl
internal open class SharedFlowImpl<T>(
    private val replay: Int,
    private val bufferCapacity: Int, // 通过参数来设置缓存区容量
    private val onBufferOverflow: BufferOverflow
) : AbstractSharedFlow<SharedFlowSlot>(), MutableSharedFlow<T>, CancellableFlow<T>, FusibleFlow<T> { 
	... ... ...
}  
  
//【2】AbstractSharedFlow
internal abstract class AbstractSharedFlow<S : AbstractSharedFlowSlot<*>> : SynchronizedObject() {
  protected var slots: Array<S?>? = null // 槽位数组，按需分配，每个元素是 AbstractSharedFlowSlot 的子类实例;
      private set // 仅读

  protected var nCollectors = 0 // 已分配但还没有释放的槽位数
      private set // 仅读

  private var nextIndex = 0 // 下一个空闲槽位的起始索引
  private var _subscriptionCount: SubscriptionCountStateFlow? = null // 记录订阅数量的 StateFlow（懒加载）
  ... ... ...
}   

这里能看出 SharedFlowImpl 的实现。

AbstractSharedFlowSlot

AbstractSharedFlowSlot 用于表示一个订阅者，内部会有关联的订阅者的协程体列表：

//【3】AbstractSharedFlowSlot    
internal abstract class AbstractSharedFlowSlot<F> {
    abstract fun allocateLocked(flow: F): Boolean
    abstract fun freeLocked(flow: F): Array<Continuation<Unit>?> 
    ... ... ...  
}

当某个订阅者的取消订阅后，那么 freeLocked 会被执行，然后返回内部的依赖的协程体，统一的 resume 他们。

参数 flow: F 表示被订阅，被观察的协程 Flow 对象对象。

SharedFlowSlot – 收集器状态管理

SharedFlowSlot 是 AbstractSharedFlowSlot 的具体实现，用于管理 SharedFlow 收集器（collector）状态：

internal class SharedFlowSlot : AbstractSharedFlowSlot<SharedFlowImpl<*>>() {
    @JvmField
    var index = -1L // 该收集器当前 "待发射" 的值在缓冲区中的索引;

    @JvmField
    var cont: Continuation<Unit>? = null // 收集器关联的被挂起的 continuation 对象;

    override fun allocateLocked(flow: SharedFlowImpl<*>): Boolean {
        if (index >= 0) return false
        //【1】从 SharedFlow 获取要消费的数据的索引 index;
        index = flow.updateNewCollectorIndexLocked()
        return true
    }

    override fun freeLocked(flow: SharedFlowImpl<*>): Array<Continuation<Unit>?> {
        assert { index >= 0 }
        val oldIndex = index
        index = -1L
        cont = null
        //【2】通知 SharedFlow 收集器已移除，并返回需要 resume 的 continuation 数组
        return flow.updateCollectorIndexLocked(oldIndex)
    }
}

属性分析：

• index ：”待发射” 的值在缓冲区中的索引，用于对数据进行跟踪；

  • -1：插槽空闲，没有收集器使用；

  • >= 0：有效索引，指向缓冲区中下一个要发送给收集器的值位置；

• cont ：挂起的 continuation 对象。

  • 收集器会关联一个协程体，当收集器等待上游的数据时，关联的协程会挂起，continuation 对象被存储在这里；
  • 等到上游有新值后，continuation 会被唤醒；

方法分析：

• allocateLocked：检查插槽是否空闲，如果空闲，从 SharedFlow 获取要消费的数据的索引 index；不空闲，就返回 false；
• freeLocked：通知 SharedFlow 该收集器已移除，并返回可能需要恢复的 continuation 对象数组；

AbstractSharedFlow – 数据管理核心

再来看看 AbstractSharedFlow，他是 SharedFlow 数据结构管理的抽象，基于 Slot 去管理内部的订阅者。

internal abstract class AbstractSharedFlow<S : AbstractSharedFlowSlot<*>> : SynchronizedObject() {
  // 槽位数组，按需分配，每个元素是 AbstractSharedFlowSlot 的子类实例，对应一个订阅者;
  protected var slots: Array<S?>? = null
      private set

  protected var nCollectors = 0 // 已分配但还没有释放的槽位数（订阅者）
      private set

  private var nextIndex = 0 // 下一个空闲槽位的起始索引
    
  // 记录订阅者数量的 StateFlow（懒加载）  
  private var _subscriptionCount: SubscriptionCountStateFlow? = null
  val subscriptionCount: StateFlow<Int> // 获取订阅数量
      get() = synchronized(this) {
          _subscriptionCount ?: SubscriptionCountStateFlow(nCollectors).also {
              _subscriptionCount = it
          }
      }

  protected abstract fun createSlot(): S
  protected abstract fun createSlotArray(size: Int): Array<S?>
    
  ... ... ...
}  

基类暴露了一些 create 方法，用于创建 AbstractSharedFlowSlot，表示一个订阅者：也就是数据流的观察者。

同时，他的订阅者数量 subscriptionCount 是 StateFlow 类型，这样设计的目的是：当订阅者数量变化时能自动通知观察者，Flow 天然支持这种机制，同时也方便和协程结合。

sharedFlow.subscriptionCount.collect { count ->
    println("订阅者数量: $count")
}

subscriptionCount 的初始化使用了懒加载。

allocateSlot – Slot 创建

当有新的 collector 来订阅流的时候，那么就会创建一个 Slot，保存订阅者：

protected fun allocateSlot(): S {
    // 局部变量，用于在锁外更新订阅计数 StateFlow
    val subscriptionCount: SubscriptionCountStateFlow?
    
    //【1】创建 slot，synchronized 同步机制。
    val slot = synchronized(this) {
        //【1.1】判断 slot 数组是否初始化、是否需要扩容：
        val slots = when (val curSlots = slots) {
            // 1：数组尚未创建，创建初始容量为 2 的数组
            null -> createSlotArray(2).also { slots = it }
            // 2：数组已存在但容量不足，2 倍扩容
            else -> if (nCollectors >= curSlots.size) {
                curSlots.copyOf(2 * curSlots.size).also { slots = it }
            } else {
                // 3：容量足够，使用现有数组
                curSlots
            }
        }
        
        //【1.2】查找并分配一个空闲插槽，nextIndex 是下一个查询的起始位置。
        var index = nextIndex
        var slot: S
        while (true) {
            // 获取当前位置的槽位，如果不存在则创建新 slot
            slot = slots[index] ?: createSlot().also { slots[index] = it }
            
            index++
            if (index >= slots.size) index = 0 // 循环查找
            
            // 检查 slot 是否可以使用，如果成功，跳出循环。
            // 注意：slot 存在复用情况，比如之前分配后又释放了，allocateLocked 函数负责检查是否可用
            if ((slot as AbstractSharedFlowSlot<Any>).allocateLocked(this)) break
            // 如果插槽已被占用，继续循环查找
        }
        
        //【1.3】记录下一个搜索起始位置 nextIndex，优化查询效率，增加活跃收集器计数
        nextIndex = index
        nCollectors++ 
        
        subscriptionCount = _subscriptionCount 
        
        // 返回分配的插槽 slot
        slot
    }
    
    //【2】增加订阅者计数对象 StateFlow 内部的值。
    subscriptionCount?.increment(1)
    
    return slot
}

每当下游有一个 collecotor 的时候，都会通过 createSlot() 创建一个插槽 Slot ，起始就是一个管理对象，负责管理 Flow 和订阅者的关系。

freeSlot – Slot 释放

释放一个 Slot，当订阅者取消订阅或完成收集时调用此方法：

protected fun freeSlot(slot: S) {
    val subscriptionCount: SubscriptionCountStateFlow?
    
    //【1】释放订阅者，然后返回需要回复的协程体的列表：
    val resumes = synchronized(this) {
        //【1.1】减少活跃收集器计数
        nCollectors--
        
        // 如果没有任何活跃的，重置查找起点
        if (nCollectors == 0) nextIndex = 0
        
        subscriptionCount = _subscriptionCount 
        
        //【1.2】释放槽位并获取需要恢复的协程列表
        // freeLocked 会标记 slot 为空闲状态，然后返回因为缓冲区满而被挂起的协程列表（如果有的话）
        (slot as AbstractSharedFlowSlot<Any>).freeLocked(this)
    }
    
    /*
     *【2】恢复被挂起的协程。
     * 这种情况发生在：
     * 1. SharedFlow 有缓冲区限制
     * 2. 慢速订阅者的槽位已满，导致发射者被挂起
     * 3. 当这个慢速订阅者被释放时，之前排队等待的数据可以继续发送
     * 4. 相关的发射者协程需要被恢复
     */
    for (cont in resumes) cont?.resume(Unit)
    
    //【3】减少订阅者计数对象 StateFlow 内部的值。
    subscriptionCount?.increment(-1)
}

核心逻辑

collect – 收集器

核心的逻辑在 collect 方法里面；

collect 是挂起函数，所以有一个隐性参数：Continuation 对象，这个 Continuation 是下游的协程体实例。

//【*】下游调用 collect 传入 FlowCollector 对象;
// collect 是挂起函数，所以有一个隐性参数：Continuation 对象。
override suspend fun collect(collector: FlowCollector<T>): Nothing {
    //【1】为当前收集器分配一个 SharedFlowSlot 对象，对数据和状态做管理；
    val slot = allocateSlot()
    try {
        //【2】如果收集器实现了 SubscribedFlowCollector，调用 onSubscription 通知已经订阅；
        // 比如：onStart 操作符能够创建这种类型的收集器，实现通知；
        if (collector is SubscribedFlowCollector) collector.onSubscription()
          
        //【3】获取 collect 调用时所处的协程的 job；，来自下游的协程;
        val collectorJob = currentCoroutineContext()[Job]
        while (true) {
            var newValue: Any?
            while (true) {
                //【4】自旋，获取下一个要发送的值;
                // 如果能获取到 break，执行 collector.emit 发送;
                newValue = tryTakeValue(slot) // attempt no-suspend fast path first
                if (newValue !== NO_VALUE) break
                
                //【5】获取不到，那么收集者会进入挂起状态;
                awaitValue(slot) // await signal that the new value is available
            }
          
            // 校验当前的协程是否活跃;
            collectorJob?.ensureActive()

            //【6】发送数据给 FlowCollector，然后继续自旋读取下一个值。
            collector.emit(newValue as T)
        }
    } finally {
        freeSlot(slot) // 释放 slot
    }
}

核心点：

• collect 是挂起函数，所以有一个隐性参数：Continuation 对象，也就是所在的外部协程的协程体；

• allocateSlot 为当前的 FlowCollector 申请了一 slot，管理流发送过程中的 FlowCollector 的数据、关联协程体；

• 双重循环结构 tryTakeValue 持续收集直到协程被取消，如果有值就 break 出去执行发送，如果没有值 awaitValue 会挂起收集者所在的协程；

这里可以看到，职责分离的设计思想：

• FlowCollector 设计用于值的处理和转换、发送；
• SharedFlowSlot 设计用于管理状态，包括数据索引、关联协程；

这种分离的设计思想提升了性能：每个 SharedFlowSlot 只修改和自己相关的数据，互不干扰；

其次，FlowCollector 也可以在不同的 SharedFlow 之间复用；

其中，核心的方法：tryTakeValue 和 awaitValue。

tryTakeValue

从缓存区中获取 value，参数 slot 内记录了当前协程需要获取数据的 index：

// returns NO_VALUE if cannot take value without suspension
private fun tryTakeValue(slot: SharedFlowSlot): Any? {
    var resumes: Array<Continuation<Unit>?> = EMPTY_RESUMES
    val value = synchronized(this) {
        //【1】通过 SharedFlowSlot 返回合适的 index
        val index = tryPeekLocked(slot)
        if (index < 0) { // -1 表示要挂起;
            NO_VALUE
        } else {
            //【2】计算下一个 index，并更新 slot；
            // 获取当前要访问的值：newValue;
            val oldIndex = slot.index
            val newValue = getPeekedValueLockedAt(index)
            slot.index = index + 1 // points to the next index after peeked one
              
            //【3】更新收集器 index，并尝试恢复等待的发射器协程，目的：提升并发性能。
            resumes = updateCollectorIndexLocked(oldIndex)
            newValue
        }
    }
    //【4】我的理解是尝试唤醒其他的收集器协程，提升读取效率！
    for (resume in resumes) resume?.resume(Unit)
    return value
}

tryPeekLocked

通过 SharedFlowSlot 返回合适的 index：

• 参数：slot - 当前订阅者的插槽
• 返回值：Long - 要读取的缓冲区索引（>=0）或 -1（表示需要挂起）

// returns -1 if cannot peek value without suspension
private fun tryPeekLocked(slot: SharedFlowSlot): Long {
    // return buffered value if possible
    val index = slot.index
    //【1】当前索引小于缓冲区数据的末尾索引，说明缓冲区有数据可读，那就直接返回；
    // 否则，说明此时缓存区里面没有可读数据，或者没有缓冲区。   
    if (index < bufferEndIndex) return index
      
    //【2】检查是否有缓冲区，bufferCapacity 通过参数指定。
    // 如果有缓存区但是没缓存数据，那么返回 -1，表示需要挂起，收集器永远不会尝试与发射者进行会合，也就是直接传递数据。  
    if (bufferCapacity > 0) return -1L // if there's a buffer, never try to rendezvous with emitters
      

    //【3】这里是说明没有配置缓存区，属于同步共享流。
    // head 是缓存数据的起始下标，index 大于 head 说明按照顺序，这个收集器靠后，不能立刻获取到数据。
    // 返回 -1 表示挂起，确保永远第一个收集器接收发射者的数据。
    // Synchronous shared flow (bufferCapacity == 0) tries to rendezvous
    if (index > head) return -1L // ... but only with the first emitter (never look forward)
      
    //【4】如果没有等待的发射者（发射队列为空），无法进行会合，那就返回 -1，表示挂起。
    if (queueSize == 0) return -1L // nothing there to rendezvous with
      
    //【5】满足所有条件，与第一个等待的发射者直接会合。
    return index // rendezvous with the first emitter
}

这里的 rendezvous 直译过来是会合，简单理解就是：

不需要经过缓存区，发射器直接传递数据给收集器，这种同步共享流。

在同步共享流（无缓冲区）中：

• 发射者发送值时，如果没有订阅者准备好接收，发射者会挂起
• 订阅者尝试读取时，如果没有值可用，但有等待的发射者，它们可以直接 “会合”
• 这种会合允许值直接从发射者传递到订阅者，不经过缓冲区。

getPeekedValueLockedAt

private fun getPeekedValueLockedAt(index: Long): Any? =
    //【1】从环形缓冲区中获取指定索引处的元素
    when (val item = buffer!!.getBufferAt(index)) {
        //【2】如果是 Emitter 对象，从中获取值。 
        is Emitter -> item.value
        //【3】其他的情况，直接获取到缓存值。  
        else -> item
    }

// Emitter
private class Emitter(
    @JvmField val flow: SharedFlowImpl<*>,
    @JvmField var index: Long,
    @JvmField val value: Any?,
    @JvmField val cont: Continuation<Unit> // 发射器所在的协程体。
) : DisposableHandle {
    override fun dispose() = flow.cancelEmitter(this)
}

这里分析下 Emitter：

• Emitter ：在同步共享流（bufferCapacity = 0）中，当发射者发送值但无订阅者时，会创建一个 Emitter 对象放入队列
• Emitter.value：存储了发射者要发送的实际值

目的：为了存储发射者的协程体（continuation），以便与订阅者会合时恢复发射者。

updateCollectorIndexLocked

这个方法的作用是：更新收集器的数据索引，并且尝试唤醒发射器和收集器。

该方法逻辑很长，我们重点关注核心的逻辑：

internal fun updateCollectorIndexLocked(oldIndex: Long): Array<Continuation<Unit>?> {
    ... ... ...
    
    //【1】更新 bufferEndIndex 缓存区边界索引。
    // 计算如果我们丢弃不再需要的项且没有发射者被恢复时的新缓冲区大小：
    // 我们必须保留从 newMinIndex 到缓冲区末尾的所有项（这部分是要被发送的缓存数据）
    var newBufferEndIndex = bufferEndIndex // 可变变量，当等待者被恢复时会增长
    
    //【2】计算最大的发射器恢复数量（最多可以恢复多少个等待的发射者）
    val maxResumeCount = if (nCollectors > 0) {
        //【2.1】如果我们有 nCollectors 个收集器，我们可以恢复最多 maxResumeCount 个等待的发射者
        // a) queueSize -> 我们有多少等待的发射者
        // b) bufferCapacity - newBufferSize0 -> 在不超出缓冲区容量的情况下可以恢复的数量：
        // 																			 也就是已经读取的数据个数；
        val newBufferSize0 = (newBufferEndIndex - newMinCollectorIndex).toInt() // 缓冲区内未读取数据的大小（基于新最小索引）
        // 在发射器个数和能再存入的数据个数之间取较小值；  
        minOf(queueSize, bufferCapacity - newBufferSize0) 
    } else {
        //【2.2】如果没有收集器了，我们必须恢复所有等待的发射者（避免它们永远挂起）
        queueSize // 等待发射者的数量（最多）
    }
    
    var resumes: Array<Continuation<Unit>?> = EMPTY_RESUMES
    
    val newQueueEndIndex = newBufferEndIndex + queueSize
    
    //【3】如果需要恢复发射者（maxResumeCount > 0），那么就把 emitter 的值取出来，放到 buffer 区内。
    // 然后恢复这些 emitter。 
    if (maxResumeCount > 0) {
        // 创建数组来存储要恢复的 continuation
        resumes = arrayOfNulls(maxResumeCount)
        var resumeCount = 0
        val buffer = buffer!! // 获取缓冲区引用（非空断言）
        
        //【3.1】遍历等待的发射者队列区间
        for (curEmitterIndex in newBufferEndIndex until newQueueEndIndex) {
            // 获取当前位置的发射者
            val emitter = buffer.getBufferAt(curEmitterIndex) 
            if (emitter !== NO_VALUE) {
                emitter as Emitter // 类型断言：必须是 Emitter 类
                
                // 1. 保存发射者的 continuation 以便后续恢复
                resumes[resumeCount++] = emitter.cont
                
                // 2. 将原位置标记为 NO_VALUE（表示已取消或已处理）
                buffer.setBufferAt(curEmitterIndex, NO_VALUE)
                
                // 3. 将发射者的值移到缓冲区的前部，就是替换 newBufferEndIndex 位置的值为数据；
                buffer.setBufferAt(newBufferEndIndex, emitter.value)
                newBufferEndIndex++ // 缓冲区结束索引向后移动
                
                // 4. 如果已经恢复了足够数量的发射者，停止恢复；
                if (resumeCount >= maxResumeCount) break
            }
        }
    }
    
    ... ... ...
    
    //【4】更新缓冲区状态: replayIndex、minCollectorIndex、bufferSize、queueSize、隐式更新 head
    // 计算 newHead，并清除掉 head 和 newHead 之前的废弃数据。
    updateBufferLocked(newReplayIndex, newMinCollectorIndex, newBufferEndIndex, newQueueEndIndex)
    
    //【5】清理缓冲区尾部连续的无效条目，同步共享流不清理，中间的 NO_VALUE 会在 emit 的时候覆盖。
    // 1、当发射者被取消时，其对应的 Emitter 对象会被标记为 NO_VALUE；
    // 2、在 updateCollectorIndexLocked 中恢复发射者后，原位置会被设为 NO_VALUE；
    cleanupTailLocked()
    
    //【6】恢复发射者协程，我们需要唤醒挂起的收集器
    if (resumes.isNotEmpty()) resumes = findSlotsToResumeLocked(resumes)
    
    return resumes
}

里面的细节就不详细分析了。

awaitValue

当无法获取值后，需要挂起 FlowCollector 所在的协程：

awaitValue 是挂起函数，所以有一个隐性参数：Continuation 对象，这个 Continuation 来自 collect 方法。

private suspend fun awaitValue(slot: SharedFlowSlot): Unit = suspendCancellableCoroutine { cont ->
    synchronized(this) lock@{
        //【1】这里是二次检测，防止此时有缓存依然挂起的情况；
        val index = tryPeekLocked(slot) // recheck under this lock
        if (index < 0) {
            //【2】挂起，并且将 Continuation 记录到 slot 内部;
            slot.cont = cont // Ok -- suspending
        } else {
            cont.resume(Unit) // has value, no need to suspend
            return@lock
        }
        slot.cont = cont // suspend, waiting
    }
}

emit – 发射器

emit 用于在上游协程触发数据的发送：

override suspend fun emit(value: T) {
    //【1】快速路径：尝试直接发送
    if (tryEmit(value)) return // fast-path
      
    //【2】慢速路径：无法直接发送时挂起
    emitSuspend(value)
}

tryEmit

尝试直接发送：

override fun tryEmit(value: T): Boolean {
    var resumes: Array<Continuation<Unit>?> = EMPTY_RESUMES  // 需要恢复的收集器列表
    
    val emitted = synchronized(this) { // 确保核心数据再锁内执行。
        // 尝试发送数据；
        if (tryEmitLocked(value)) {
            //【1】如果发射成功，找出所有可以立即读取该数据的被挂起的收集器
            resumes = findSlotsToResumeLocked(resumes)
            true // 发射成功
        } else {
            false // 发射失败（缓冲区满且无法立即传送）
        }
    }
    
    //【2】恢复收集器的协程（避免在锁内执行用户代码）
    for (cont in resumes) cont?.resume(Unit)
    // 返回是否发射成功
    return emitted 
}

emitSuspend

无法直接发送时挂起：

private suspend fun emitSuspend(value: T) = suspendCancellableCoroutine<Unit> sc@{ cont ->
    var resumes: Array<Continuation<Unit>?> = EMPTY_RESUMES
    val emitter = synchronized(this) lock@{
        //【1】检查缓冲区（确保真的满了）
        if (tryEmitLocked(value)) {
            cont.resume(Unit)  // 直接恢复发射者协程（成功了）
            resumes = findSlotsToResumeLocked(resumes)  // 尝试唤醒收集器协程
            return@lock null   // 返回 null 表示不需要创建 Emitter
        }
        
        //【2】缓冲区确实满了，创建挂起的发射者对象
        Emitter(this, head + totalSize, value, cont).also {
            enqueueLocked(it)      // 将发射者加入等待队列
            queueSize++            // 等待队列大小增加
            
            //【2.1】同步共享流的特殊处理：可能立即与等待收集器会合
            if (bufferCapacity == 0) resumes = findSlotsToResumeLocked(resumes)
        }
    }
    
    //【3】注册取消回调（如果创建了 Emitter）
    emitter?.let { cont.disposeOnCancellation(it) }
    
    //【4】恢复可以继续的收集器协程
    for (r in resumes) r?.resume(Unit)
}

tryEmitLocked

尝试发送数据：

private fun tryEmitLocked(value: T): Boolean {
    // ---------- 快速通道：处理没有收集器的情况 ----------
    //【1】如果没有收集器，不需要考虑背压、不需要协调慢的收集器，只需要管好重播缓存就行。
    if (nCollectors == 0) return tryEmitNoCollectorsLocked(value)

    // ---------- 慢速流程：有收集器 ----------
    //【2】先判断缓存区是否满了，需要满足下面 2 个条件，才按背压策略处理：
    // 1. 缓存本身已经满了 (bufferSize >= bufferCapacity)
    // 2. 最慢的收集器的 index 超过了重播范围 (minCollectorIndex <= replayIndex)
    // 那它肯定不需要队列最前面的老数据了，那些数据其实可以丢掉腾地方。
    if (bufferSize >= bufferCapacity && minCollectorIndex <= replayIndex) {
        // 根据背压策略处理：
        when (onBufferOverflow) {
            // 策略是“挂起”：发射器挂起
            BufferOverflow.SUSPEND -> return false
            // 策略是“丢新的”：丢弃新值，继续执行下次发射
            BufferOverflow.DROP_LATEST -> return true
            // 策略是“丢旧的”：丢弃旧数据，继续向下执行。
            BufferOverflow.DROP_OLDEST -> { /* 不返回，继续往下走硬塞 */ }
        }
    }

    //【3】新值进缓冲区，尾部插入。
    enqueueLocked(value)
    bufferSize++ 

    //【4】如果超过了容量，把队头最老的丢掉
    // 在 DROP_OLDEST 策略下会发生，也可能发生在虽然物理满了但能回收空间的情况下。
    if (bufferSize > bufferCapacity) dropOldestLocked()

    //【5】清理重播缓存：确保只保留最新的重播数据。
    // 核心就是更新 replayIndex，确保只有 replay 个重播数据；
    val currentReplaySize = (bufferEndIndex - replayIndex).toInt()
    if (currentReplaySize > replay) {
        updateBufferLocked(replayIndex + 1, minCollectorIndex, bufferEndIndex, queueEndIndex)
    }

    //【6】插入成功。
    return true
}

/**
 * 当没有任何收集器时，此时快速发送，此时没有背压问题，只需要处理 replay 数据。
 */
private fun tryEmitNoCollectorsLocked(value: T): Boolean {
    // 没收集器才能走这个函数
    assert { nCollectors == 0 }

    //【1】不支持重播缓存，那么这个值丢弃掉，假装发送成功。
    if (replay == 0) return true

    //【2】数据进缓存队列。
    enqueueLocked(value)
    bufferSize++

    //【3】对重播数据做调整，根据背压机制。
    if (bufferSize > replay) dropOldestLocked()

    //【4】没有收集器，所以当前所有的数据都不需要被处理，那么 minCollectorIndex 直接设置到 buffer 末尾。  
    minCollectorIndex = head + bufferSize

    return true 
}

上面是插入的逻辑分析：

三个背压策略的行为对比

策略	当缓冲区满且被阻塞时的行为	返回值	发射者状态
SUSPEND	发射者需要挂起等待，不发射	false	挂起，直到有空间
DROP_LATEST	丢弃新发射的值	true	继续逻辑，新值会丢失
DROP_OLDEST	接受新值，丢弃缓冲区中最旧的值	true	继续逻辑，旧值会丢失

缓冲区结构分析

如下是 SharedFlow 内部的核心数据结构：

// Stored state
private var buffer: Array<Any?>? = null // allocated when needed, allocated size always power of two
private var replayIndex = 0L // minimal index from which new collector gets values
private var minCollectorIndex = 0L // minimal index of active collectors, equal to replayIndex if there are none
private var bufferSize = 0 // number of buffered values
private var queueSize = 0 // number of queued emitters

// Computed state
private val head: Long get() = minOf(minCollectorIndex, replayIndex)
private val replaySize: Int get() = (head + bufferSize - replayIndex).toInt()
private val totalSize: Int get() = bufferSize + queueSize
private val bufferEndIndex: Long get() = head + bufferSize
private val queueEndIndex: Long get() = head + bufferSize + queueSize

下面简单的梳理下注释里面讲的 SharedFlow 缓存区的几个知识点。

我们重点看这个图示：

/*
							buffered values
         /-----------------------\
                      replayCache      queued emitters
                      /----------\/----------------------\
     +---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+
     |   | 1 | 2 | 3 | 4 | 5 | 6 | E | E | E | E | E | E |   |   |   |
     +---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+
           <sup>           </sup>           <sup>                      </sup>
           |           |           |                      |
        1.head         |      head + bufferSize     head + totalSize 
                																		  4.queueEndIndex
           |           |           |
 index of the slowest  |    index of the fastest
  possible collector   |     possible collector
                							3.bufferEndIndex
           |           |
           |     2.replayIndex == new collector's index
           \---------------------- /
      range of possible 5.minCollectorIndex
      
      head == minOf(minCollectorIndex, replayIndex) // by definition
      totalSize == bufferSize + queueSize // by definition

   INVARIANTS:
      minCollectorIndex = activeSlots.minOf { it.index } ?: (head + bufferSize)
      replayIndex <= head + bufferSize
*/      

根据图示我们做一些简单的梳理：

缓存区的核心存储区

private var buffer: Array<Any?>? = null // allocated when needed, allocated size always power of two

buffer 是一个数组，内部存储 2 种类型的数据：

• 真实数据：这个是指发射器和收集器要处理的数据；
• emiter：发射器，用于记录发送协程体的信息；

[value 1 ] [value 2] [value 3] [value 4] … …[value N] [emiter 1][emiter 2][emiter 3]… …[emiter N]

是一个环形数组：

// 使用按位与 (&) 代替取模 (%)，。
private fun Array<Any?>.getBufferAt(index: Long) = get(index.toInt() and (size - 1))
private fun Array<Any?>.setBufferAt(index: Long, item: Any?) = set(index.toInt() and (size - 1), item) 

缓冲区大小必须是 2 的幂

类似于 hashmap 和 hashmap 根据 hash 计算 index 的策略，buffer 的长度必须是 2 的幂。

缓存区的索引

head- 缓冲数据起始索引

缓存数据的起始下标。

private val head: Long get() = minOf(minCollectorIndex, replayIndex)
// head == minOf(minCollectorIndex, replayIndex) // by definition

值取 minCollectorIndex 和 replayIndex 的较小值，目的是确保不会丢失数据。

head 代表着缓冲区中实际存储的最老的数据的索引。

replayIndex- 重播起始索引

private var replayIndex = 0L // minimal index from which new collector gets values
// replayIndex <= head + bufferSize

重放缓存（replayCache）的起始索引，保证新订阅者至少看到之前订阅者看到的数据量

• 重放范围：replayIndex 到 bufferEndIndex - 1
• 重放大小： bufferEndIndex - replayIndex ≤ replay 参数
• 数据解释：哪些数据需要作为粘性数据为新收集器保留。
• 只能增长（单调递增），公式：max(当前 replayIndex, bufferEndIndex - min(replay, bufferSize))

bufferEndIndex- 缓冲数据结束索引

private val bufferEndIndex: Long get() = head + bufferSize

缓冲数据的 buffered values 的结束边界位置，[head, bufferEndIndex) 为所有的缓存数据。

minCollectorIndex - 最小的收集器访问数据索引

private var minCollectorIndex = 0L // minimal index of active collectors, equal to replayIndex if there are none
/*  
   INVARIANTS:
      minCollectorIndex = activeSlots.minOf { it.index } ?: (head + bufferSize)
 */     

表示收集器最先收集的数据索引，上面可以看到 minCollectorIndex 的计算方式 minOf，范围同样是：[head, bufferEndIndex)

索引 < minCollectorIndex 的数据会被丢弃，也就是说：[minCollectorIndex, bufferEndIndex) 区间的数据才是收集器的目标数据。

// 示例：
收集器A: index = 3 (需要数据 3,4,5...)
收集器B: index = 5 (需要数据 5,6,7...)
minCollectorIndex = 3
// - 索引 0,1,2 的数据可以丢弃（没有收集器需要它们）
// - 索引 3,4 的数据必须保留（至少收集器A需要）

• 和其他索引的关系：

head ≤ replayIndex ≤ minCollectorIndex ≤ bufferEndIndex + queueSize

queueEndIndex - 结束索引

private val queueEndIndex: Long get() = head + bufferSize + queueSize

发射器（queued emitters）的结束位置，[bufferEndIndex, queueEndIndex) 为发射器的存储区间。

缓存的清理

可以被安全清理的数据：

1. index < head 的数据（废弃数据已经清理）
2. index < minCollectorIndex 且 索引 < replayIndex 的数据（不再被访问且不再被重播）

必须保留的数据：

1. index ≥ minCollectorIndex 的数据（收集器需要）
2. index ≥ replayIndex 的数据（重放缓存）

缓冲区的区间

基于前面的 index，我们继续梳理如下的区间概念：

replay cache - 重播缓存

private val replaySize: Int get() = (head + bufferSize - replayIndex).toInt()

靠近缓冲区的尾部，为新订阅者提供历史粘性数据，范围由 replay 参数配置。

override val replayCache: List<T>
    get() = synchronized(this) {
        val replaySize = this.replaySize
        if (replaySize == 0) return emptyList()
        val result = ArrayList<T>(replaySize)
        val buffer = buffer!! // must be allocated, because replaySize > 0
        @Suppress("UNCHECKED_CAST")
        // 这里能看到范围；  
        for (i in 0 until replaySize) result += buffer.getBufferAt(replayIndex + i) as T
        result
    }

buffered values - 缓冲值

private var bufferSize = 0 // number of buffered values

在 replayCache 的后面，表示：已发射但尚未被所有收集器消费的值。

这里的发射表示上游的 Flow 通过 emit 执行数据的提交，数据会基于 replayIndex 的位置添加到缓存区。

queued emitters（发射器队列）

private var queueSize = 0 // number of queued emitters

在缓冲区尾部这里，在 bufferEndIndex 之后。

用于存储挂起的发射器 emitters（当缓冲区满时 / 同步流情况下无收集器的情况），注意：存储的是发射器而不是数据。

3 StateFlow

StateFlow 用于在协程体系下对标 LiveData，他能提供 LiveData 近乎一样的效果：

StateFlow 可以被看做一个简化版本的 SharedFlow：只能存一个值的 SharedFlow。但是其内部的实现不并不是基于 SharedFlow，由于其内部只有一个值，所以实现相对来说更加简单。

@Suppress("FunctionName")
public fun <T> MutableStateFlow(value: T): MutableStateFlow<T> = StateFlowImpl(value ?: NULL)

StateFlowImpl

构造器：

private class StateFlowImpl<T>(
    initialState: Any // T | NULL
) : AbstractSharedFlow<StateFlowSlot>(), MutableStateFlow<T>, CancellableFlow<T>, FusibleFlow<T> {
    @Suppress("UNCHECKED_CAST")
    public override var value: T
        get() = NULL.unbox(_state.value)
        set(value) { updateState(null, value ?: NULL) }
    ... ... ...

    override val replayCache: List<T>
        get() = listOf(value)

可以看到，和 SharedFlowImpl 是继承实现关系一样，关键在于内部的数据结构不同：

数据结构

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) }

• 私有变量

  • _state：状态变量，存储数据，原子操作确保线程安全，而且需要有初始值。

  • sequence：奇数表示有更新在进行，偶数表示空闲状态。

• 公有变量

  • value：对外提供一个统一的入口，实现了状态变量的更新和读取。

so，回顾下：StateFlow 本质上是重播数据 replay=1 的特殊 SharedFlow，他们在设计上有很多相似之处。

StateFlowSlot – 插槽

和 SharedFlow 一样，需要对收集器进行 slot 保存：

private val NONE = Symbol("NONE")

private val PENDING = Symbol("PENDING")

// StateFlowSlot
private class StateFlowSlot : AbstractSharedFlowSlot<StateFlowImpl<*>>() {
    /**
     * Each slot can have one of the following states:
     *
     * * `null` -- it is not used right now. Can [allocateLocked] to new collector.
     * * `NONE` -- used by a collector, but neither suspended nor has pending value.
     * * `PENDING` -- pending to process new value.
     * * `CancellableContinuationImpl<Unit>` -- suspended waiting for new value.
     *
     * It is important that default `null` value is used, because there can be a race between allocation
     * of a new slot and trying to do [makePending] on this slot.
     */
    private val _state = atomic<Any?>(null)

可以看到，StateFlowSlot 内部是一个状态机：

private val _state = atomic<Any?>(null)

通知翻译注释，我们能知道有如下的状态：

// null    -> 槽位空闲，未分配给收集器
// NONE    -> 已分配，收集器正在运行，无待处理值
// PENDING -> 有待处理的新值
// CancellableContinuationImpl<Unit> -> 收集器挂起等待新值

状态转换

经过分析源码，得知他的状态切换的核心流程：

// collect 注册收集器的时候：
// 1、已分配还未分配值 NONE  -------> awaitPending 挂起状态 CancellableContinuation -------> makePending 恢复去处理新值 NONE

// 特定条件触发：收集器正在运行（没有挂起）时来了新值
// 2、已分配还未分配值 NONE -------> 有待处理的值 PENDING  -------> 值被消费后进入再分配状态 NONE

allocateLocked 和 freeLocked – 创建和释放

代码很简单，不多解释：

override fun allocateLocked(flow: StateFlowImpl<*>): Boolean {
    // No need for atomic check & update here, since allocated happens under StateFlow lock
    if (_state.value != null) return false // not free
    _state.value = NONE // allocated
    return true
}

override fun freeLocked(flow: StateFlowImpl<*>): Array<Continuation<Unit>?> {
    _state.value = null // free now
    return EMPTY_RESUMES // nothing more to do
}

StateFlowSlot 相比 SharedFlowSlot 不一样的地方在于，如何被唤醒和处理数据。

takePending 和 awaitPending[suspend] – 检测并进入等待状态

takePending 判断是否进入处理值的状态。

awaitPending 则是 cas 切换到处理值的状态。

fun takePending(): Boolean = _state.getAndSet(NONE)!!.let { state ->
    assert { state !is CancellableContinuationImpl<*> }  // can be NONE or PENDING
    return state === PENDING //【1】state 为 PENDING 表示准备处理值。
}

@Suppress("UNCHECKED_CAST")
suspend fun awaitPending(): Unit = suspendCancellableCoroutine sc@ { cont ->
    assert { _state.value !is CancellableContinuationImpl<*> } // can be NONE or PENDING
    //【1】尝试将状态从 NONE 改为挂起                                                              
    if (_state.compareAndSet(NONE, cont)) return@sc // 挂起成功，等待被 resume，suspend 返回。

    //【2】cas 失败，说明已经是 pending 状态了。  
    assert { _state.value === PENDING }
                                                   
    //【3】PENDING 状态 直接处理新的数据，恢复协程。                                                                
    cont.resume(Unit)
}

可以看到：awaitPending 是进入挂起状态的。

只有在 PENDING 下在恢复协程。

makePending – 通知操作

将 slot 更改为 PENDING 状态，准备处理数据：

fun makePending() {
    _state.loop { state ->
        when {
            state == null -> return // this slot is free - skip it
            state === PENDING -> return // already pending, nothing to do
            state === NONE -> { 
                //【1】将 slot 更改为 PENDING 状态，这表明收集器协程序在处理新值的时候，又来了新值。
                if (_state.compareAndSet(state, PENDING)) return
            }
            else -> { // must be a suspend continuation state
                // we must still use CAS here since continuation may get cancelled and free the slot at any time
                //【2】恢复协程，并修改为 NONE 状态。
                if (_state.compareAndSet(state, NONE)) {
                    (state as CancellableContinuationImpl<Unit>).resume(Unit)
                    return
                }
            }
        }
    }
}

核心逻辑

collect – 收集器

这是收集的过程：

// StateFlowImpl
override suspend fun collect(collector: FlowCollector<T>): Nothing {
    //【1】和 SharedFlow 一样的操作，但是创建的是 StateFlowSlot 对象。
    val slot = allocateSlot()
    try {
        if (collector is SubscribedFlowCollector) collector.onSubscription()
        val collectorJob = currentCoroutineContext()[Job]
          
        //【2】旧状态数据。   
        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
        //【3】自旋，不断的读取数据。  
        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()

            //【3】比较数据的区别，只有第一次发送或者数据不同，会立刻发送值；  
            if (oldState == null || oldState != newState) {
                collector.emit(NULL.unbox(newState))
                oldState = newState
            }
          
            //【3】判断是否进入了 pending 状态，没有的话执行 awaitPending 等待 pending
            // awaitPending 可能会挂起协程！！
            // 如果被 cancelled 的话，那么会跑出 canceledException 异常，然后退出 loop，执行 freeSlot。
            if (!slot.takePending()) { // try fast-path without suspending first
                slot.awaitPending() // only suspend for new values when needed
            }
        }
    } finally {
        freeSlot(slot)
    }
}

可以看到：

• 如果有新值的话， 会先发送，再立刻尝试挂起协程。
• 如果已经是 pending 状态，会直接返回。

emit – 发射器

很简单，核心是对状态变量 value 的 set：

override suspend fun emit(value: T) {
    this.value = value
}

updateState

更新状态变量：

• expectedState: CAS 的期望状态
• newState: 新状态

private fun updateState(expectedState: Any?, newState: Any): Boolean {
    var curSequence: Int 
    var curSlots: Array<StateFlowSlot?>?
    
    //【1】初步更新（加锁），可以看到锁力度很小，只有在处理 flow 内部自身数据和状态时才会加锁
    // 目的是避免死锁，有些情况比如 Dispatchers.Unconfined 下可能多个协作会切换到同一个线程中执行。
    synchronized(this) {
        val oldState = _state.value
        
        // CAS 操作验证，旧值不匹配，返回 false
        if (expectedState != null && oldState != expectedState) return false 
        //【1.1】CAS 操作验证，旧值和新值一样，返回 true，不通知收集器。
        // 这是和 livedata 不一样的地方。  
        if (oldState == newState) return true 
        
        //【1.2】更新新值（atomic）。
        _state.value = newState
        
        //【1.3】序列号校验和状态判断。
        curSequence = sequence
        // 根据序列号的奇偶性来判断 StateFlow 当前状态
        if (curSequence and 1 == 0) { 
            // 偶数表示 StateFlow 处于空闲状态（没有正在进行的更新）
            // 将序列号加 1 使其变为奇数，表示"有更新正在进行"
            curSequence++ 
            sequence = curSequence
        } else {
            // 奇数，表示：StateFlow 正在进行更新数据中
            // 将序列号加 2（保持奇数），通知有新的状态变更，直接返回，由正在进行中的更新操作更新。
            sequence = curSequence + 2 
            return true
        }
        
        // 获取 Flow 内部的 slots 缓存
       curSlots = slots 
    }
    
    //【2】自旋状态，通知所有的收集器，这里涉及到收集器自身的逻辑。
    while (true) {
        //【2.1】通知所有收集器有状态更新
        // 遍历 slot，将其标记为 pending 状态，唤醒等待的收集器协程
        curSlots?.forEach { slot ->
            slot?.makePending() //【2.1】唤醒收集器：--> NONE
        }
        
        //【2.2】在触发完更新通知后 (此时收集器协程已经在处理数据了 NONE)，接着：
        // 检查更新通知期间是否有新值到达，自旋状态下 curSequence 始终是本次的序列号缓存。
        // 但是 updateState 可能会被其他更新操作调用，从而更新 Flow 本身的 sequence。
        synchronized(this) {
            if (sequence == curSequence) { 
                // 序列号没有变化，表示在我们通知期间没有新的更新到达
                // 将序列号加 1 使其恢复为偶数，表示 StateFlow 回到空闲状态，然后返回。
                sequence = curSequence + 1 
                return true
            }
            
            // 序列号已变化，表示在我们通知期间有新的更新到达
            // 重新读取所有信息，在下个自旋中处理这些新更新
            curSequence = sequence 
            curSlots = slots // 重新读取 slots 数组（可能有收集器加入或被移除）
        }
        
    }
}

整个流程很清晰，不多说。

根据前面的状态分析：

• 如果收集器注册的时候没有更新，那么 StateFlowSlot 的状态就修改为：NONE –> 挂起。等待后续 updateState 触发更新后，在 makePending 里恢复协程，挂起 —> NONE。
• 协程会从 awaitPending 的位置恢复，然后再次进入自旋，发送数据，再次 NONE –> 挂起。

另外：

• 当收集器协程在处理数据的时候为 NONE，如果此时很短时间内又来了 1 个或多个新值 或者协程处理很慢导致新值来了还没处理完，那么就会 NONE –> PENDING。
• 当 takePending 再次判断的时候，就不会进入 awaitPending 进行挂起，而是自旋继续发射值。

到这里 2 个热流的实现就已经分析完了，大家有兴趣可以对着我的博客跟跟代码。