文章详情

短信预约-IT技能 免费直播动态提醒

请输入下面的图形验证码

提交验证

短信预约提醒成功

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

2024-04-02 19:55

关注

一.协程间的通信

当需要进行协程间的通信时,可以调用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)上下游容量都不为默认值,则相加取和

到此这篇关于Kotlin协程开发之Flow的融合与Channel容量及溢出策略介绍的文章就介绍到这了,更多相关Kotlin Flow的融合内容请搜索编程网以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程网!

阅读原文内容投诉

免责声明:

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析(完整版)

    难度     807人已做
    查看
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     351人已做
    查看
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     314人已做
    查看
  • 2024年上半年软考高项第一、二批次真题考点汇总(完整版)

    难度     433人已做
    查看
  • 2024年上半年系统架构设计师考试综合知识真题

    难度     221人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机
位置:首页-资讯-移动开发
咦!没有更多了?去看看其它编程学习网 内容吧
首页课程
资料下载
问答资讯