一、核心接口分工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()时实际上是从这个内部队列的头部取出数据元素。如果队列为空接收协程会被挂起直到有新数据入队或通道关闭。