Kotlin Flow的融合
LeeDuo. 人气:0一.协程间的通信
当需要进行协程间的通信时，可以调用Channel方法，创建一个Channel接口指向的对象，通过调用该对象的send方法和receive方法实现消息的发送与接收。协程对Channel接口的实现，本质上与阻塞队列类似，这里不再赘述。
1.通道容量
事实上，send方法与receive方法并没有定义在Channel接口中，而是分别定义在SendChannel接口和ReceiveChannel接口中。Channel接口中只是定义了一些与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
)
}
}
从上面的代码可以看出Channel接口继承自SendChannel接口和ReceiveChannel接口。因此，一个Channel接口指向的对象，既可以用于发送消息，也可以用于接收消息。
2.溢出策略
Channel除了容量策略外，还有溢出策略，用于决定当Channel的容量已满时，而下一个消息到来时的行为。溢出策略定义在枚举类BufferOverflow中，代码如下：
public enum class BufferOverflow {
// 当容量已满时，挂起调用send方法的协程
SUSPEND,
// 当容量已满时，删除旧数据，将新的数据添加进去，不挂起调用send方法的协程
DROP_OLDEST,
// 当容量已满时，忽略当前要添加的数据，不挂起调用send方法的协程
DROP_LATEST
}
二.FusibleFlow接口
FusibleFlow接口继承自Flow接口。一个类实现了该接口，表示该类创建的流可以与其上游或下游相邻的流进行融合，当流发生融合时，就会调用接口中定义的fuse方法，代码如下：
@InternalCoroutinesApi
public interface FusibleFlow<T> : Flow<T> {
// 用于流的融合
public fun fuse(
context: CoroutineContext = EmptyCoroutineContext,
capacity: Int = Channel.OPTIONAL_CHANNEL,
onBufferOverflow: BufferOverflow = BufferOverflow.SUSPEND
): Flow<T>
}
FusibleFlow接口的fuse方法，默认容量为OPTIONAL_CHANNEL，默认溢出策略为SUSPEND。
流的融合
在Flow中，当channelFlow方法、flowOn方法、buffer方法、produceIn方法、broadcastIn方法相邻调用时，就会触发流的融合。
具体融合的过程，其实是将下游流的容量、溢出策略、上下文传递给上游的流处理，上游的流根据自身的容量、溢出策略、上下文以及下游的流的容量、溢出策略、上下文重新计算，得到新的容量、溢出策略、上下文，并返回一个融合后的流。
三.ChannelFlow类
ChannelFlow类是一个抽象类，实现了FusibleFlow接口。下面分析一下fuse方法对于上下游流融合的策略，代码如下：
@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>
...
}流融合的原则
根据上面对fuse方法的分析，可以总结出fuse方法在计算容量和溢出策略时的四个原则：
1）下游优先于上游
2）溢出策略优先于容量
3）非默认值优先于默认值
4）上下游容量都不为默认值，则相加取和
加载全部内容