新闻详情

新闻详情

首页 / 资讯中心 / 详情

SendChannel 与 ReceiveChannel 通信机制

发布时间:2026/9/7 22:48:51来源:尧图网络
SendChannel 与 ReceiveChannel 通信机制
一、核心接口分工SendChannel作为通道的发送端仅暴露数据写入能力。其核心方法为挂起函数send(element: E)用于向通道提交数据。当缓冲区已满时当前协程会自动挂起而非阻塞线程同时支持trySend非阻塞尝试发送以及close()关闭通道等操作。ReceiveChannel作为通道的接收端仅暴露数据读取能力。其核心方法为挂起函数receive(): E用于从通道取出数据。当通道为空时当前协程会自动挂起等待新数据到达同时支持tryReceive非阻塞尝试接收亦可直接使用for-in循环遍历通道内所有元素直至通道关闭。二、使用示例fun main() runBlocking { // 创建默认无缓冲通道同时拥有发送端和接收端 val channel ChannelInt() // 取出 SendChannel 侧仅可用于发送数据 val sender: SendChannelInt channel // 取出 ReceiveChannel 侧仅可用于接收数据 val receiver: ReceiveChannelInt channel ​ // 生产者协程通过 SendChannel 发送数据 launch { repeat(5) { sender.send(it) println(Sent: $it) } sender.close() // 发送完成后关闭通道 } ​ // 消费者协程通过 ReceiveChannel 接收数据 launch { // for 循环自动遍历通道关闭后自动结束循环 for (element in receiver) { println(Received: $element) } println(通信结束) } ​ delay(1000) }三、编译器生成的迭代逻辑编译器实际生成的逻辑等价于val iterator receiver.iterator() // 获取迭代器 while (iterator.hasNext()) { // 判断是否有下一个元素 val element iterator.next() // 获取下一个元素 println(element) }四、Channel 实现类public fun E Channel( capacity: Int RENDEZVOUS, onBufferOverflow: BufferOverflow BufferOverflow.SUSPEND, onUndeliveredElement: ((E) - Unit)? null ): ChannelE { // 这里的 when 判断用于选择具体的实现类 return when (capacity) { RENDEZVOUS - RendezvousChannel(onUndeliveredElement) // 无缓冲容量为 0 CONFLATED - ConflatedChannel(onUndeliveredElement) // Conflated容量为 -1 UNLIMITED - LinkedListChannel(onUndeliveredElement) // 无限缓冲使用链表 else - ArrayChannel(capacity, onBufferOverflow, onUndeliveredElement) // 有界缓冲使用数组 } }上述实现类均继承自共同的基类 AbstractChannel。// AbstractChannel.kt 简化源码逻辑 abstract class AbstractChannelE( private val onUndeliveredElement: ((E) - Unit)? ) : ChannelE, SendChannelE, ReceiveChannelE { ​ // 1. 提供迭代器入口 public final override fun iterator(): ChannelIteratorE Itr(this) ​ // ... 其他 send/receive 逻辑 }五、迭代器实现原理private class ItrE(JvmField val channel: AbstractChannelE) : ChannelIteratorE { var result: Any? POLL_FAILED // E | POLL_FAILED | Closed ​ override suspend fun hasNext(): Boolean { // check for repeated hasNext if (result ! POLL_FAILED) return hasNextResult(result) // fast path -- try poll non-blocking result channel.pollInternal() if (result ! POLL_FAILED) return hasNextResult(result) // slow-path does suspend return hasNextSuspend() } ​ private fun hasNextResult(result: Any?): Boolean { if (result is Closed*) { if (result.closeCause ! null) throw recoverStackTrace(result.receiveException) return false } return true } ​ private suspend fun hasNextSuspend(): Boolean suspendCancellableCoroutineReusable sc { cont - val receive ReceiveHasNext(this, cont) while (true) { if (channel.enqueueReceive(receive)) { channel.removeReceiveOnCancel(cont, receive) returnsc } // hm... something is not right. try to poll val result channel.pollInternal() this.result result if (result is Closed*) { if (result.closeCause null) cont.resume(false) else cont.resumeWithException(result.receiveException) returnsc } if (result ! POLL_FAILED) { Suppress(UNCHECKED_CAST) cont.resume(true, channel.onUndeliveredElement?.bindCancellationFun(result as E, cont.context)) returnsc } } } ​ Suppress(UNCHECKED_CAST) override fun next(): E { val result this.result if (result is Closed*) throw recoverStackTrace(result.receiveException) if (result ! POLL_FAILED) { this.result POLL_FAILED return result as E } ​ throw IllegalStateException(hasNext should be called prior to next invocation) } }在pollInternal内部this.result被赋值为以下结果val result channel.pollInternal() this.result result六、pollInternal 函数protected open fun pollInternal(): Any? { while (true) { val send takeFirstSendOrPeekClosed() ?: return POLL_FAILED val token send.tryResumeSend(null) if (token ! null) { assert { token RESUME_TOKEN } send.completeResumeSend() return send.pollResult } // too late, already cancelled, but we removed it from the queue and need to notify on undelivered element send.undeliveredElement() } } ​ protected fun takeFirstSendOrPeekClosed(): Send? queue.removeFirstIfIsInstanceOfOrPeekIfSend { it is Closed* }七、内部队列机制Channel 的设计灵感来源于 Java 中的BlockingQueue但其专为非阻塞挂起协程而设计。生产者SendChannel调用send(element)时实际上是将数据元素封装成一个节点放入这个内部队列的尾部。如果队列已满对于有界 Channel发送协程会被挂起直到有空间可用。消费者ReceiveChannel调用receive()时实际上是从这个内部队列的头部取出数据元素。如果队列为空接收协程会被挂起直到有新数据入队或通道关闭。
网站建设高端定制企业官网
RELATED

相关资讯

更多精彩内容,欢迎继续阅读

较早相关资讯

最新相关资讯

西工大数电实验全解析:Verilog与FPGA设计与仿真验证实践 2026/9/8 1:58:23

西工大数电实验全解析:Verilog与FPGA设计与仿真验证实践

简介:这是一套面向西工大数字电子技术实验课程的完整资料包,覆盖逻辑门、组合逻辑、时序逻辑、存储器与FPGA等经典主题,包含多份实验报告、Verilog/VHDL源码与操作截图,并涉及布尔代数、卡诺图化简和状态机设计等核心知识点。包内…

阅读更多 →
吃透MSC.ADAMS/View高级教程:从建模到参数化仿真的实战指南 2026/9/8 1:58:23

吃透MSC.ADAMS/View高级教程:从建模到参数化仿真的实战指南

简介:面向ADAMS多体动力学仿真进阶用户的MSC.ADAMS/View高级培训光盘文件,覆盖高级建模、运动学与动力学分析、接触约束设置、参数化设计及优化等核心技能,适合需要系统提升View实操能力的工程师与科研人员。压缩包共477个文件,大…

阅读更多 →
Agent长任务执行机制:从模型决策到运行时循环 2026/9/8 1:58:23

Agent长任务执行机制:从模型决策到运行时循环

看到 Agent 在任务日志里一步一步执行到第 30 步,是一种有点神奇的体验。一开始它只是在读文档,接着调了一个 API,然后根据返回值生成 SQL,再接下来又去数据库中验证了一遍结果,最后告诉你哪一项是异常的。整个过程没有…

阅读更多 →
Agent 连续跑几十步背后的四大架构:模型、运行时、工具与编排 2026/9/8 1:58:23

Agent 连续跑几十步背后的四大架构:模型、运行时、工具与编排

Agent 为什么能连续跑几十步?新手第一次接触 Agent 时,很容易把它理解成“一个特别会聊天的模型”。但真正跑过 Agent 项目就会发现,单次模型推理根本不可能完成复杂任务。模型只是“大脑”,连续执行几十步靠的是模型、运行时、工…

阅读更多 →
拆解Nanobot与OpenClaw:从源码读懂AI Agent架构设计 2026/9/8 1:58:23

拆解Nanobot与OpenClaw:从源码读懂AI Agent架构设计

1. 项目概述:为什么要用 Nanobot 来“拆”框架先说结论:OpenClaw 是目前个人 AI 助理类开源项目里,把“agent 能力”和“消息平台接入”这两件事拆得最干净的项目之一。而 Nanobot 则是它早期核心模块的前身或同源实验项目(具体关…

阅读更多 →
AI短剧连载平台选型实战:从角色一致性到漫剧工作流 2026/9/8 1:55:22

AI短剧连载平台选型实战:从角色一致性到漫剧工作流

/* MD / 富文本中的 .toc(含博客园搬家等嵌套结构);.toc-box 在侧栏,不受影响 */#content_views .toc,/* 编辑器常在目录前后插入空 p(:empty 仍占 20px),一并去掉避免顶空隙 */#content_views.markdown_views > p:empty:has(+ .toc),#content_views.markdown_views …

阅读更多 →

今日资讯

本周资讯

本月资讯

看完文章仍有疑问?

联系尧图顾问,获取一对一建站咨询

立即免费咨询 📞 400-888-8888
📞