一.协程间的通信
????当需要进行协程间的通信时,可以调用Channel方法,创建一个Channel接口指向的对象,通过调用该对象的send方法和receive方法实现消息的发送与接收。协程对Channel接口的实现,本质上与阻塞队列类似,这里不再赘述。
1.通道容量
????事实上,send方法与receive方法并没有定义在Channel接口中,而是分别定义在SendChannel接口和ReceiveChannel接口中。Channel接口中只是定义了一些与Channel容量策略相关的枚举常量,代码如下:
public interface Channel<E> : SendChannel<E>, ReceiveChannel<E> {
public companion object Factory {
public const val UNLIMITED: Int = Int.MAX_VALUE
public const val RENDEZVOUS: Int = 0
public const val CONFLATED: Int = -1
public const val BUFFERED: Int = -2
internal const val OPTIONAL_CHANNEL = -3
public const val DEFAULT_BUFFER_PROPERTY_NAME: String = "kotlinx.coroutines.channels.defaultBuffer"
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 {
SUSPEND,
DROP_OLDEST,
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。
1.流的融合
????在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> {
assert { capacity != Channel.CONFLATED }
val newContext = context + this.context
val newCapacity: Int
val newOverflow: BufferOverflow
if (onBufferOverflow != BufferOverflow.SUSPEND) {
newCapacity = capacity
newOverflow = onBufferOverflow
} else {
newCapacity = when {
this.capacity == Channel.OPTIONAL_CHANNEL -> capacity
capacity == Channel.OPTIONAL_CHANNEL -> this.capacity
this.capacity == Channel.BUFFERED -> capacity
capacity == Channel.BUFFERED -> this.capacity
else -> {
assert { this.capacity >= 0 }
assert { capacity >= 0 }
val sum = this.capacity + capacity
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>
...
}
1.流融合的原则
????根据上面对fuse方法的分析,可以总结出fuse方法在计算容量和溢出策略时的四个原则:
1)下游优先于上游
2)溢出策略优先于容量
3)非默认值优先于默认值
4)上下游容量都不为默认值,则相加取和
|