news 2026/6/10 16:34:08

kotlin协程-热数据通道Channel

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
kotlin协程-热数据通道Channel

种一颗树的最好时机是十年前,其次是现在。
学习也一样。
跟着霍老师的《深入理解 Kotlin 携程》学习一下协程。

直奔主题,认识 Channel

Channel 实际上就是一个并发安全的队列,它可以用来连接协程,实现不同协程的通信

suspendfunmain(){valchannel=Channel<Int>()valproducer=GlobalScope.launch{vari=0while(true){delay(1000)channel.send(i++)}}valconsumer=GlobalScope.launch{while(true){valelement=channel.receive()println(element)}}producer.join()consumer.join()}

上述的代码中构造了两个协程 producer 和 consumer,我么呢没有为它们明确制定调度器,所以他们的调度器都是默认的。其中 producer 中每隔 1 秒向 Channel 发送一个整数,而 consumer 中一致在读取 channel 来获取数据并打印,显然发送端比接收端更慢,在没有可以读取的值时,receive 是挂起的,直到有新元素到达,这么看来 receive 一定是一个挂起函数,那么 send 呢?

Channel 的容量

我们查看 send 方法的声明,发现它也是挂起函数。那么发送端为什么要挂起?上面也提到,Channel 实际上就是一个队列,队列中一定存在缓冲区,一旦这个缓冲区满了,一直没有人调用 receive 并取走元素,send 就要挂起,等待接收者取走元素后再写入 Channel。

publicfun<E>Channel(capacity:Int=RENDEZVOUS):Channel<E>=when(capacity){RENDEZVOUS->RendezvousChannel()UNLIMITED->LinkedListChannel()CONFLATED->ConflatedChannel()BUFFERED->ArrayChannel(CHANNEL_DEFAULT_CAPACITY)else->ArrayChannel(capacity)}

我们构造 Cahnnel 的时候调用了一个名为 Channle 的函数,但它不是 Channel 的构造函数。在 Kotlin 中,经常定义一个顶级函数来伪装成同名类型的构造器,这本质上是工厂函数。这里有一个 Int 类型的 capacity 参数,默认值为 RENDEZVOUS。
这时候如果不调用 receive,send 就会一直挂起等待。

UNLIMITED比较好理解,没有限制,来者不拒。
CONFLATED这个名字可能有迷惑性,字面意思是合并,但实际上这个函数的效果是只保留最后一个元素,也就是说缓冲区只有一个元素大小,每次有新元素到来,都会覆盖掉旧元素。
BUFFERED效果类似于 ArrayBlockingQueue,接收一个值作为缓冲区容量大小。

迭代 Channel

我们在发送和读取的时候写了一个while(true)的死循环,因为需要不断地进行读写操作。这里我们可以直接获取一个 Channel 的 Iterator

valconsumer=GlobalScope.launch{valiter=channel.iterator()while(iter.hasNext()){valelement=iter.next()println(element)delay(1000)}}

其中 iter.hasNext()是挂起函数,在判断是否有下一个元素的时候就需要去 Channel 中读取元素了。当然也可以for ... in ..:

valconsumer=GlobalScope.launch{for(elementinchannel){println(element)delay(1000)}}

produce 和 actor

来看两个便捷的构造生产者和消费这的 api,我们可以通过produce来启动移动生产者协程,并返回一个ReceiveChannel,其他协程就可以通过这个 channel 来获取数据了。
同样的,我们可以通过actor来启动消费者协程,并返回一个SendChannel,其他协程就可以通过这个 channel 来发送数据了。

valreceiveChannel:ReceiveChannel<Int>=GlobalScope.produce{repeat(100){delay(100)send(it)}}valsendChannel:SendChannel<Int>=GlobalScope.actor{while(true){valelement=receive()println(element)}}

Channel 的关闭

以上面的produce方法为例,我们可以看到最终返回的是一个ProducerCoroutine,它的定义如下:

privateclassProducerCoroutine<E>(parentContext:CoroutineContext,channel:Channel<E>):ChannelCoroutine<E>(parentContext,channel,true,active=true),ProducerScope<E>{overridevalisActive:Booleanget()=super.isActiveoverridefunonCompleted(value:Unit){_channel.close()}overridefunonCancelled(cause:Throwable,handled:Boolean){valprocessed=_channel.close(cause)if(!processed&&!handled)handleCoroutineException(context,cause)}}

我们发现在它的完成取消方法中都会调用_channel.close的方法。也正是这样,Channel 才被称为热数据流。这里有一点需要注意:对千一个Channe如果我们调用了它的close方法,它会立即停止接收新元素,也就是说这时候它的isClosedForSend会立即返回true,而由Channel缓冲区的存在, 这时候可能还有一些元素没有被处理完,因此要等所有的元素都被读取之后isClosedForReceive才会返回true.

BroadcastChannel

在实际环境中,经常会出现一个发送对应多个接收的情况。这里我们就需要 BroadcastChannel 了。

valbroadcastChannel=BroadcastChannel<Int>(Channel.BUFFERED)valproducer=GlobalScope.launch{List(3){delay(1000)broadcastChannel.send(it)}}List(3){index->GlobalScope.launch{valreceiveChannel=broadcastChannel.openSubscription()for(iinreceiveChannel){println("[#$index] received$i")}}}.joinAll()

这里有个细节需要注意一下,如果把发送端的dealy(100)去掉,可能会出现部分元素收不到或者完全收不到的情况,这是因为BroadcastChannel在发送的时候没有订阅者,这条消息就被丢弃了。
我们也可以通过普通的 Channel 进行转换:

valchannel=Channel<Int>()channel.broadcast(3)

这里需要注意一下,BroadcastChannel被标记为过时了,可以使用SharedFlowStateFlow代替。channel.broadcast()方法也被标记为过时,也是使用SharedFlow来代替


以上

版权声明: 本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若内容造成侵权/违法违规/事实不符,请联系邮箱:809451989@qq.com进行投诉反馈,一经查实,立即删除!
网站建设 2026/6/10 15:21:33

微振动隔离

振动隔离工程是一门专注于振动控制的工程学科&#xff0c;核心在于设计并实施各类技术与设备&#xff0c;削弱机械系统或建筑结构受到的外部振动干扰。其核心目标是降低振动对设备精度、结构安全及人员舒适度的不利影响&#xff0c;进而保障系统稳定运行、延长设备使用寿命、保…

作者头像 李华
网站建设 2026/6/3 3:39:18

强烈安利10个AI论文软件,专科生轻松搞定毕业论文!

强烈安利10个AI论文软件&#xff0c;专科生轻松搞定毕业论文&#xff01; AI 工具如何助力论文写作&#xff0c;让专科生轻松应对毕业挑战 在当前的学术环境中&#xff0c;越来越多的学生开始借助 AI 工具来提升论文写作效率。尤其是在面对毕业论文这一重要任务时&#xff0c…

作者头像 李华
网站建设 2026/6/9 21:18:43

重新理解身份:为什么 SAP Cloud Identity Services 是 SAP BTP 时代的入口与底座

在 SAP BTP 里做 ABAP 或 CAP 开发时,很多人一开始把注意力放在服务编排、数据建模、扩展点、CI/CD,直到某一天系统上线,才发现真正让业务跑不起来的,往往不是代码,而是人怎么进来、进来能做什么、离职后怎么立刻收回权限。在云时代,身份不再只是登录页的一次用户名密码校…

作者头像 李华
网站建设 2026/6/10 15:22:49

EtherCAT分析仪:原理与使用全解析

先辑半导体EtherCAT分析仪 的原理和使用方法。第一部分&#xff1a;原理与架构EtherCAT分析仪的核心原理是利用芯片强大的硬件外设和软件协议栈&#xff0c;实现对EtherCAT数据帧的监听、捕获、解析和调试。它工作在 “监听模式”或“透明网关模式” &#xff0c;本身不主动作为…

作者头像 李华
网站建设 2026/5/26 19:34:16

【Java毕设源码分享】基于springboot+vue的沉浸式戏曲文化体验系统的设计与实现(程序+文档+代码讲解+一条龙定制)

博主介绍&#xff1a;✌️码农一枚 &#xff0c;专注于大学生项目实战开发、讲解和毕业&#x1f6a2;文撰写修改等。全栈领域优质创作者&#xff0c;博客之星、掘金/华为云/阿里云/InfoQ等平台优质作者、专注于Java、小程序技术领域和毕业项目实战 ✌️技术范围&#xff1a;&am…

作者头像 李华
网站建设 2026/5/29 9:37:03

CI_CD Pipeline趋势:加速集成与交付

在追求更快、更可靠、更精准的软件开发周期压力下&#xff0c;许多开发者正被迫摒弃过时的单体式开发方法&#xff0c;转而采用更简化的流程。为了实现真正的敏捷性&#xff0c;团队们纷纷为其软件交付流程实施了持续集成与持续交付策略。一个成熟的CI/CD Pipeline策略&#xf…

作者头像 李华