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

// 继承SendChannel接口和ReceiveChannel接口
public interface Channel<E> : SendChannel<E>, ReceiveChannel<E> {
    // 枚举常量
    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<T> : Flow<T> {
    // 用于流的融合
    public fun fuse(
        context: CoroutineContext = EmptyCoroutineContext,
        capacity: Int = Channel.OPTIONAL_CHANNEL,
        onBufferOverflow: BufferOverflow = BufferOverflow.SUSPEND
    ): Flow<T>
}

@InternalCoroutinesApi
public abstract class ChannelFlow<T>(
    // 上游流的上下文
    @JvmField public val context: CoroutineContext,
    // 上下游之间流的缓存容量
    @JvmField public val capacity: Int,
    // 溢出策略
    @JvmField public val onBufferOverflow: BufferOverflow
) : FusibleFlow<T> {
    ...
    public override fun fuse(context: CoroutineContext, capacity: Int, onBufferOverflow: BufferOverflow): Flow<T> {
        // 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<T>
    ...
}