一、核心接口分工SendChannel作為通道的發(fā)送端僅暴露數(shù)據(jù)寫入能力。其核心方法為掛起函數(shù)send(element: E)用于向通道提交數(shù)據(jù)。當(dāng)緩沖區(qū)已滿時當(dāng)前協(xié)程會自動掛起而非阻塞線程同時支持trySend非阻塞嘗試發(fā)送以及close()關(guān)閉通道等操作。ReceiveChannel作為通道的接收端僅暴露數(shù)據(jù)讀取能力。其核心方法為掛起函數(shù)receive(): E用于從通道取出數(shù)據(jù)。當(dāng)通道為空時當(dāng)前協(xié)程會自動掛起等待新數(shù)據(jù)到達同時支持tryReceive非阻塞嘗試接收亦可直接使用for-in循環(huán)遍歷通道內(nèi)所有元素直至通道關(guān)閉。二、使用示例fun main() runBlocking { // 創(chuàng)建默認無緩沖通道同時擁有發(fā)送端和接收端 val channel ChannelInt() // 取出 SendChannel 側(cè)僅可用于發(fā)送數(shù)據(jù) val sender: SendChannelInt channel // 取出 ReceiveChannel 側(cè)僅可用于接收數(shù)據(jù) val receiver: ReceiveChannelInt channel ? // 生產(chǎn)者協(xié)程通過 SendChannel 發(fā)送數(shù)據(jù) launch { repeat(5) { sender.send(it) println(Sent: $it) } sender.close() // 發(fā)送完成后關(guān)閉通道 } ? // 消費者協(xié)程通過 ReceiveChannel 接收數(shù)據(jù) launch { // for 循環(huán)自動遍歷通道關(guān)閉后自動結(jié)束循環(huán) for (element in receiver) { println(Received: $element) } println(通信結(jié)束) } ? delay(1000) }三、編譯器生成的迭代邏輯編譯器實際生成的邏輯等價于val iterator receiver.iterator() // 獲取迭代器 while (iterator.hasNext()) { // 判斷是否有下一個元素 val element iterator.next() // 獲取下一個元素 println(element) }四、Channel 實現(xiàn)類public fun E Channel( capacity: Int RENDEZVOUS, onBufferOverflow: BufferOverflow BufferOverflow.SUSPEND, onUndeliveredElement: ((E) - Unit)? null ): ChannelE { // 這里的 when 判斷用于選擇具體的實現(xiàn)類 return when (capacity) { RENDEZVOUS - RendezvousChannel(onUndeliveredElement) // 無緩沖容量為 0 CONFLATED - ConflatedChannel(onUndeliveredElement) // Conflated容量為 -1 UNLIMITED - LinkedListChannel(onUndeliveredElement) // 無限緩沖使用鏈表 else - ArrayChannel(capacity, onBufferOverflow, onUndeliveredElement) // 有界緩沖使用數(shù)組 } }上述實現(xiàn)類均繼承自共同的基類 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 邏輯 }五、迭代器實現(xiàn)原理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內(nèi)部this.result被賦值為以下結(jié)果val result channel.pollInternal() this.result result六、pollInternal 函數(shù)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* }七、內(nèi)部隊列機制Channel 的設(shè)計靈感來源于 Java 中的BlockingQueue但其專為非阻塞掛起協(xié)程而設(shè)計。生產(chǎn)者SendChannel調(diào)用send(element)時實際上是將數(shù)據(jù)元素封裝成一個節(jié)點放入這個內(nèi)部隊列的尾部。如果隊列已滿對于有界 Channel發(fā)送協(xié)程會被掛起直到有空間可用。消費者ReceiveChannel調(diào)用receive()時實際上是從這個內(nèi)部隊列的頭部取出數(shù)據(jù)元素。如果隊列為空接收協(xié)程會被掛起直到有新數(shù)據(jù)入隊或通道關(guān)閉。