Kotlin协程开发之Flow的融合与Channel容量及溢出策略介绍

// 继承SendChannel接口和ReceiveChannel接口 public interface Channel : SendChannel, ReceiveChannel { // 枚举常量 public companion object Factory { // Channel的容量为无限 public const val UNLIMITED: Int = Int.MAX_VALUE // Channel的容量为0，没有缓存 public const val RENDEZVOUS: Int = 0 // Channel的容量为1，溢出策略为DROP_OLDEST， // 后一个的数据会覆盖前一个数据 public const val CONFLATED: Int = -1 // Channel的容量为默认值CHANNEL_DEFAULT_CAPACITY， // 默认溢出策略为SUSPEND，send方法会发生挂起 // 当容量策略为BUFFERED，而溢出策略不为SUSPEND时，Channel的容量为1 public const val BUFFERED: Int = -2 // 协程内部使用的一个默认枚举值，不对外暴露 internal const val OPTIONAL_CHANNEL = -3 // 用于手动配置容量策略为BUFFERED时的默认值 public const val DEFAULT_BUFFER_PROPERTY_NAME: String = "kotlinx.coroutines.channels.defaultBuffer" // 容量策略为BUFFERED时的默认值 // 默认64，最小1，最大为Int.MAX_VALUE-1 internal val CHANNEL_DEFAULT_CAPACITY = systemProp(DEFAULT_BUFFER_PROPERTY_NAME, 64, 1, UNLIMITED - 1 ) } } 

public enum class BufferOverflow { // 当容量已满时，挂起调用send方法的协程 SUSPEND, // 当容量已满时，删除旧数据，将新的数据添加进去，不挂起调用send方法的协程 DROP_OLDEST, // 当容量已满时，忽略当前要添加的数据，不挂起调用send方法的协程 DROP_LATEST } 

@InternalCoroutinesApi public interface FusibleFlow : Flow { // 用于流的融合 public fun fuse( context: CoroutineContext = EmptyCoroutineContext, capacity: Int = Channel.OPTIONAL_CHANNEL, onBufferOverflow: BufferOverflow = BufferOverflow.SUSPEND ): Flow } 

@InternalCoroutinesApi public abstract class ChannelFlow( // 上游流的上下文 @JvmField public val context: CoroutineContext, // 上下游之间流的缓存容量 @JvmField public val capacity: Int, // 溢出策略 @JvmField public val onBufferOverflow: BufferOverflow ) : FusibleFlow { ... public override fun fuse(context: CoroutineContext, capacity: Int, onBufferOverflow: BufferOverflow): Flow { // CONFLATED是一个复合的类型，需要拆解成capacity = 0, onBufferOverflow = DROP_OLDEST assert { capacity != Channel.CONFLATED } // 计算融合后流的上下文 val newContext = context + this.context // 用于保存融合后流的容量 val newCapacity: Int // 用于保存融合后流的溢出策略 val newOverflow: BufferOverflow // SUSPEND为默认溢出策略，如果溢出策略不是默认的策略 if (onBufferOverflow != BufferOverflow.SUSPEND) { // 直接保存 newCapacity = capacity newOverflow = onBufferOverflow } else { // 如果是默认策略 // 计算并保存新的容量 newCapacity = when { // 如果之前的容量为默认枚举值，则使用新的 this.capacity == Channel.OPTIONAL_CHANNEL -> capacity // 如果新的容量为默认枚举值，则使用原来的 capacity == Channel.OPTIONAL_CHANNEL -> this.capacity // 如果原来的容量为默认值CHANNEL_DEFAULT_CAPACITY，则使用新的 this.capacity == Channel.BUFFERED -> capacity // 如果新的容量为默认值CHANNEL_DEFAULT_CAPACITY，则使用原来的 capacity == Channel.BUFFERED -> this.capacity // 如果不为默认值或默认枚举值 else -> { // 检查容量都是大于等于0的 assert { this.capacity >= 0 } assert { capacity >= 0 } // 将原来的容量和新的容量进行相加 val sum = this.capacity + capacity // 如果相加后大与等于0，则容量为相加后的结果，否则为无限 if (sum >= 0) sum else Channel.UNLIMITED } } // 保存溢出策略 newOverflow = this.onBufferOverflow } // 如果融合的两个流的上下文相同，容量相同，溢出策略也相同 if (newContext == this.context && newCapacity == this.capacity && newOverflow == this.onBufferOverflow) // 则直接返回 return this // 有变化则根据新计算出得参数，创建融合后的流 return create(newContext, newCapacity, newOverflow) } // 由子类进行重写 protected abstract fun create(context: CoroutineContext, capacity: Int, onBufferOverflow: BufferOverflow): ChannelFlow ... }