路徑為：./src/runtime/chan.go 文件中生真，先看channel結(jié)構(gòu)體:

type hchan struct {
    qcount   uint           // total data in the queue 當(dāng)前隊(duì)列中的數(shù)據(jù)的個(gè)數(shù)
    dataqsiz uint           // size of the circular queue   channel環(huán)形隊(duì)列的大小
    buf      unsafe.Pointer // points to an array of dataqsiz elements  存放數(shù)據(jù)的環(huán)形隊(duì)列的指針
    elemsize uint16     // channel 中存放的數(shù)據(jù)類(lèi)型的大小|即每個(gè)元素的大小
    closed   uint32     // channel 是否關(guān)閉的標(biāo)示
    elemtype *_type // element type channel中存放的元素的類(lèi)型
    sendx    uint   // send index   當(dāng)前發(fā)送元素指向channel環(huán)形隊(duì)列的下標(biāo)指針
    recvx    uint   // receive index 當(dāng)前接收元素指向channel環(huán)形隊(duì)列的下標(biāo)指針
    recvq    waitq  // list of recv waiters 等待接收元素的goroutine隊(duì)列
    sendq    waitq  // list of send waiters  等待發(fā)送元素的goroutine隊(duì)列

    // lock protects all fields in hchan, as well as several
    // fields in sudogs blocked on this channel.
    //
    // Do not change another G's status while holding this lock
    // (in particular, do not ready a G), as this can deadlock
    // with stack shrinking.
    // 保持此鎖定時(shí)不要更改另一個(gè)G的狀態(tài)（特別是脖咐，沒(méi)有準(zhǔn)備好G），因?yàn)檫@可能會(huì)因堆棧收縮而死鎖汇歹。
    lock mutex
}

以及waitq的結(jié)構(gòu)體：

//等待發(fā)送及接收的等待接收元素的goroutine隊(duì)列的結(jié)構(gòu)體
type waitq struct {
    first *sudog
    last  *sudog
}

等待發(fā)送或接受goroutine鏈表的結(jié)構(gòu)體sudog:

// sudog表示等待鏈表中的g屁擅，例如用于發(fā)送/接收在頻道上。
// 一個(gè)G可以出現(xiàn)在許多等待列表中产弹，因此一個(gè)G有許多sudog派歌；許多G可能在等待相同的結(jié)果,同步對(duì)象弯囊，因此一個(gè)對(duì)象可能有多個(gè)sudog。
// sudog是從一個(gè)特殊的池中分配的胶果。使用AcquireDog和
// 釋放sudog來(lái)分配和釋放它們匾嘱。

type sudog struct {
    // 以下字段受hchan.lock的保護(hù)
    g *g // 綁定的goroutine
    isSelect bool   // isSelect的布爾值表示該線程是否正在進(jìn)行操作channel
    next     *sudog // 指向下一個(gè)等待線程的指針地址
    prev     *sudog // 指向上一個(gè)等待線程的指針地址
    elem     unsafe.Pointer // data element (may point to stack) 數(shù)據(jù)對(duì)象（可能指向棧）
    // 當(dāng)進(jìn)行channel的send操作時(shí)，elem代表將要保存進(jìn)channel的元素
    // 當(dāng)進(jìn)行channel的recv操作時(shí), elem代表從channel接受的元素
    // G1執(zhí)行ch<-task4的時(shí)候早抠，G1會(huì)創(chuàng)建一個(gè)sudog然后將elem保存進(jìn)入sendq隊(duì)列

    // 從不同場(chǎng)景訪問(wèn)以下字段霎烙。
    // 對(duì)于channel，WaitLink只能由G訪問(wèn)蕊连。
    // 對(duì)于信號(hào)量悬垃，所有字段（包括上面的字段）只有在持有semaroot鎖時(shí)才能訪問(wèn)。
    acquiretime int64 // 獲取時(shí)間
    releasetime int64 // 釋放時(shí)間
    ticket      uint32
    parent      *sudog // semaRoot binary tree
    waitlink    *sudog // g.waiting list or semaRoot
    waittail    *sudog // semaRoot
    c           *hchan // channel // 綁定channel
}

從以上三個(gè)結(jié)構(gòu)體我們即可看出channel其實(shí)就是由一個(gè)環(huán)形數(shù)組實(shí)現(xiàn)的隊(duì)列用于在確定大小的連續(xù)內(nèi)存塊進(jìn)行數(shù)據(jù)元素的存儲(chǔ)甘苍，用waitq以及鏈表sudog共同實(shí)現(xiàn)goroutine的等待隊(duì)列尝蠕，并在每個(gè)鏈表元素中存儲(chǔ)待從channel中取出或拷貝進(jìn)channel的數(shù)據(jù)元素,可以理解為每個(gè)等待線程都是channel的搬運(yùn)工，負(fù)責(zé)運(yùn)送數(shù)據(jù).
其中hchan中的lock是 recvq 是讀操作阻塞在 channel 的 goroutine 列表载庭，sendq 是寫(xiě)操作阻塞在 channel 的 goroutine 列表看彼。
qcount 和 dataqsiz 分別描述了該channel的當(dāng)前使用量和最大容量。

接下來(lái)進(jìn)行channel的每一個(gè)函數(shù)方法進(jìn)行分析:

makechan:

func makechan(t *chantype, size int) *hchan {
    elem := t.elem

    // compiler checks this but be safe.
    // 判斷定義的channel存儲(chǔ)的每個(gè)元素大小是否在范圍內(nèi)
    if elem.size >= 1<<16 {
        throw("makechan: invalid channel element type")
    }

    if hchanSize%maxAlign != 0 || elem.align > maxAlign {
        throw("makechan: bad alignment")
    }

    // 計(jì)算channel所需要分配的內(nèi)存大小
    mem, overflow := math.MulUintptr(elem.size, uintptr(size))
    // 判斷內(nèi)存大小是否超過(guò)限制
    if overflow || mem > maxAlloc-hchanSize || size < 0 {
        panic(plainError("makechan: size out of range"))
    }

    var c *hchan
    switch {
    // 當(dāng)計(jì)算channel的內(nèi)存大小為0時(shí)創(chuàng)建不帶buffer的channel
    case mem == 0:
        // Queue or element size is zero.
        c = (*hchan)(mallocgc(hchanSize, nil, true))
        // Race detector uses this location for synchronization.
        c.buf = c.raceaddr()
    // elem類(lèi)型非指針
    // 當(dāng)計(jì)算channel的內(nèi)存大小為0時(shí)創(chuàng)建帶buffer的channel
    // 分配連續(xù)的內(nèi)存 （連續(xù)內(nèi)存有利于提高內(nèi)存使用效率）
    // 直接從棧中分配內(nèi)存
    case elem.kind&kindNoPointers != 0:
        // 分配內(nèi)存
        c = (*hchan)(mallocgc(hchanSize+mem, nil, true))
        c.buf = add(unsafe.Pointer(c), hchanSize)
    // 當(dāng)channel元素類(lèi)型包含指針時(shí)分配離散的內(nèi)存
    default:
        // Elements contain pointers.
        c = new(hchan)
        // 分配內(nèi)存
        c.buf = mallocgc(mem, elem, true)
    }

    c.elemsize = uint16(elem.size)
    c.elemtype = elem
    c.dataqsiz = uint(size)

    if debugChan {
        print("makechan: chan=", c, "; elemsize=", elem.size, "; elemalg=", elem.alg, "; dataqsiz=", size, "\n")
    }
    return c
}

函數(shù)接收兩個(gè)參數(shù)囚聚，一個(gè)是channel里面保存的元素的數(shù)據(jù)類(lèi)型靖榕，一個(gè)是緩沖的容量(如果為0表示是非緩沖buffer)，創(chuàng)建流程如下：

根據(jù)傳遞的緩沖大小size是否為零顽铸，分別創(chuàng)建不帶buffer的channel或則帶size大小的緩沖channel：
對(duì)于不帶緩沖channel茁计，申請(qǐng)一個(gè)hchan數(shù)據(jù)結(jié)構(gòu)的內(nèi)存大小跋破；
對(duì)于帶緩沖channel，new一個(gè)hchan對(duì)象瓶蝴，并初始化buffer內(nèi)存毒返；
對(duì)于包含指針帶緩存的channel同樣申請(qǐng)一個(gè)hchan數(shù)據(jù)結(jié)構(gòu)的內(nèi)存大小;
以及設(shè)置channel的屬性。
帶指針以及不帶指針帶內(nèi)存申請(qǐng)區(qū)別可以看內(nèi)存管理相關(guān)源碼舷手。

chanbuf:

 //chanbuf(c, i) is pointer to the i'th slot in the buffer.
func chanbuf(c *hchan, i uint) unsafe.Pointer {
    return add(c.buf, uintptr(i)*uintptr(c.elemsize))
}

chanbuf的實(shí)現(xiàn)很簡(jiǎn)單拧簸，主要就是根據(jù)下標(biāo)（sendx或recvx）以及每一個(gè)元素的大小還有環(huán)形隊(duì)列的指針計(jì)算出該下標(biāo)槽點(diǎn)內(nèi)存地址并返回

chansend：

// 通用單通道發(fā)送/接收
// 如果阻塞不是nil，則將不會(huì)休眠男窟，但如果無(wú)法完成則返回盆赤。
// 當(dāng)睡眠中涉及的通道關(guān)閉時(shí)，睡眠可以通過(guò)g.param == nil喚醒歉眷。 最簡(jiǎn)單的循環(huán)和重新運(yùn)行操作; 我們會(huì) 
// 看到它現(xiàn)在已經(jīng)關(guān)閉了牺六。   

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {

    // 當(dāng) channel 未初始化或?yàn)?nil 時(shí)，向其中發(fā)送數(shù)據(jù)將會(huì)永久阻塞
    if c == nil {
        if !block {
            return false
        }
        // gopark 會(huì)使當(dāng)前 goroutine 休眠汗捡，并通過(guò) unlockf 喚醒淑际，但是此時(shí)傳入的 unlockf 為 nil, 因此，goroutine 會(huì)一直休眠
        gopark(nil, nil, waitReasonChanSendNilChan, traceEvGoStop, 2)
        throw("unreachable")
    }

    if debugChan {
        print("chansend: chan=", c, "\n")
    }

    // 如果開(kāi)啟了競(jìng)爭(zhēng)檢測(cè)
    if raceenabled {
        racereadpc(c.raceaddr(), callerpc, funcPC(chansend))
    }

    // Fast path: check for failed non-blocking operation without acquiring the lock.
    //
    // After observing that the channel is not closed, we observe that the channel is
    // not ready for sending. Each of these observations is a single word-sized read
    // (first c.closed and second c.recvq.first or c.qcount depending on kind of channel).
    // Because a closed channel cannot transition from 'ready for sending' to
    // 'not ready for sending', even if the channel is closed between the two observations,
    // they imply a moment between the two when the channel was both not yet closed
    // and not ready for sending. We behave as if we observed the channel at that moment,
    // and report that the send cannot proceed.
    //
    // It is okay if the reads are reordered here: if we observe that the channel is not
    // ready for sending and then observe that it is not closed, that implies that the
    // channel wasn't closed during the first observation.
    if !block && c.closed == 0 && ((c.dataqsiz == 0 && c.recvq.first == nil) ||
        (c.dataqsiz > 0 && c.qcount == c.dataqsiz)) {
        return false
    }

    var t0 int64
    //計(jì)時(shí)器
    if blockprofilerate > 0 {
        t0 = cputicks()
    }

    // 獲取同步鎖
    lock(&c.lock)


    // 向已經(jīng)關(guān)閉的 channel 發(fā)送消息會(huì)產(chǎn)生 panic
    if c.closed != 0 {
        unlock(&c.lock)
        panic(plainError("send on closed channel"))
    }

    // CASE1: 當(dāng)有 goroutine 在 recv 隊(duì)列上等待時(shí)，跳過(guò)緩存隊(duì)列春缕，將消息直接發(fā)給 reciever goroutine
    // dequeue 從等待接受的線程隊(duì)列鏈表獲取一個(gè)sudog
    if sg := c.recvq.dequeue(); sg != nil {
        // Found a waiting receiver. We pass the value we want to send
        // directly to the receiver, bypassing the channel buffer (if any).
        send(c, sg, ep, func() { unlock(&c.lock) }, 3)
        return true
    }

     // CASE2: 緩存隊(duì)列未滿盗胀，則將消息復(fù)制到緩存隊(duì)列上并移動(dòng)sendx下標(biāo)
    if c.qcount < c.dataqsiz {
        // Space is available in the channel buffer. Enqueue the element to send.
        qp := chanbuf(c, c.sendx)
        if raceenabled {
            raceacquire(qp)
            racerelease(qp)
        }
        typedmemmove(c.elemtype, qp, ep)
        c.sendx++
        if c.sendx == c.dataqsiz {
            c.sendx = 0
        }
        c.qcount++
        unlock(&c.lock)
        return true
    }

    if !block {
        unlock(&c.lock)
        return false
    }
     // CASE3: 緩存隊(duì)列已滿，將goroutine 加入 send 隊(duì)列
     // 創(chuàng)建 sudo
    // Block on the channel. Some receiver will complete our operation for us.
    //獲取當(dāng)前線程并綁定到sudog
    gp := getg()
    mysg := acquireSudog()
    mysg.releasetime = 0
    if t0 != 0 {
        mysg.releasetime = -1
    }
    // No stack splits between assigning elem and enqueuing mysg
    // on gp.waiting where copystack can find it.
    mysg.elem = ep
    mysg.waitlink = nil
    mysg.g = gp
    mysg.isSelect = false
    mysg.c = c
    gp.waiting = mysg
    gp.param = nil
    // 講當(dāng)前sudog放入等待發(fā)送的線程隊(duì)列
    c.sendq.enqueue(mysg)
    // 休眠線程（即阻塞）
    // 通過(guò)調(diào)用goready（gp）锄贼，goroutine可以再次運(yùn)行票灰。
    goparkunlock(&c.lock, waitReasonChanSend, traceEvGoBlockSend, 3)
    // Ensure the value being sent is kept alive until the
    // receiver copies it out. The sudog has a pointer to the
    // stack object, but sudogs aren't considered as roots of the
    // stack tracer.
    KeepAlive(ep)

    // someone woke us up.
    if mysg != gp.waiting {
        throw("G waiting list is corrupted")
    }
    gp.waiting = nil
    if gp.param == nil {
        if c.closed == 0 {
            throw("chansend: spurious wakeup")
        }
        panic(plainError("send on closed channel"))
    }
    gp.param = nil
    if mysg.releasetime > 0 {
        blockevent(mysg.releasetime-t0, 2)
    }
    mysg.c = nil

    //釋放sudog
    releaseSudog(mysg)
    return true
}

// send processes a send operation on an empty channel c.
// The value ep sent by the sender is copied to the receiver sg.
// The receiver is then woken up to go on its merry way.
// Channel c must be empty and locked.  send unlocks c with unlockf.
// sg must already be dequeued from c.
// ep must be non-nil and point to the heap or the caller's stack.
func send(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {
    if raceenabled {
        if c.dataqsiz == 0 {
            racesync(c, sg)
        } else {
            // Pretend we go through the buffer, even though
            // we copy directly. Note that we need to increment
            // the head/tail locations only when raceenabled.
            qp := chanbuf(c, c.recvx)
            raceacquire(qp)
            racerelease(qp)
            raceacquireg(sg.g, qp)
            racereleaseg(sg.g, qp)
            c.recvx++
            if c.recvx == c.dataqsiz {
                c.recvx = 0
            }
            c.sendx = c.recvx // c.sendx = (c.sendx+1) % c.dataqsiz
        }
    }
    if sg.elem != nil {
        sendDirect(c.elemtype, sg, ep)
        sg.elem = nil
    }
    gp := sg.g
    unlockf()
    gp.param = unsafe.Pointer(sg)
    if sg.releasetime != 0 {
        sg.releasetime = cputicks()
    }
    goready(gp, skip+1)
}

send 有以下四種情況：【都是對(duì)不為nil的chan的情況】
1.向已經(jīng)close的chan寫(xiě)數(shù)據(jù)，拋panic宅荤。

2.有 goroutine 阻塞在 channel recv 隊(duì)列上屑迂，此時(shí)緩存隊(duì)列( hchan.buf)為空(即緩沖區(qū)內(nèi)無(wú)元素)，直接將消息發(fā)送給 reciever goroutine,只產(chǎn)生一次復(fù)制,從當(dāng)前 channel 的等待隊(duì)列中取出等待的 goroutine膘侮，然后調(diào)用 send屈糊。goready 負(fù)責(zé)喚醒 goroutine。

3.當(dāng) channel 緩存隊(duì)列( hchan.buf )有剩余空間時(shí)琼了，將數(shù)據(jù)放到隊(duì)列里逻锐，等待接收，接收后總共產(chǎn)生兩次復(fù)制

4.當(dāng) channel 緩存隊(duì)列( hchan.buf )已滿時(shí)雕薪，將當(dāng)前 goroutine 加入 send 隊(duì)列并阻塞

receive：

// chanrecv receives on channel c and writes the received data to ep.
// ep may be nil, in which case received data is ignored.
// If block == false and no elements are available, returns (false, false).
// Otherwise, if c is closed, zeros *ep and returns (true, false).
// Otherwise, fills in *ep with an element and returns (true, true).
// A non-nil ep must point to the heap or the caller's stack.
func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
    // raceenabled: don't need to check ep, as it is always on the stack
    // or is new memory allocated by reflect.

    if debugChan {
        print("chanrecv: chan=", c, "\n")
    }

    // 從 nil 的 channel 中接收消息昧诱，永久阻塞
    if c == nil {
        if !block {
            return
        }
        gopark(nil, nil, waitReasonChanReceiveNilChan, traceEvGoStop, 2)
        throw("unreachable")
    }

    // Fast path: check for failed non-blocking operation without acquiring the lock.
    //
    // After observing that the channel is not ready for receiving, we observe that the
    // channel is not closed. Each of these observations is a single word-sized read
    // (first c.sendq.first or c.qcount, and second c.closed).
    // Because a channel cannot be reopened, the later observation of the channel
    // being not closed implies that it was also not closed at the moment of the
    // first observation. We behave as if we observed the channel at that moment
    // and report that the receive cannot proceed.
    //
    // The order of operations is important here: reversing the operations can lead to
    // incorrect behavior when racing with a close.
    if !block && (c.dataqsiz == 0 && c.sendq.first == nil ||
        c.dataqsiz > 0 && atomic.Loaduint(&c.qcount) == 0) &&
        atomic.Load(&c.closed) == 0 {
        return
    }

    var t0 int64
    if blockprofilerate > 0 {
        t0 = cputicks()
    }

    lock(&c.lock)

     // CASE1: 從已經(jīng) close 且為空的 channel recv 數(shù)據(jù)蝌以，返回空值
    if c.closed != 0 && c.qcount == 0 {
        if raceenabled {
            raceacquire(c.raceaddr())
        }
        unlock(&c.lock)
        if ep != nil {
            typedmemclr(c.elemtype, ep)
        }
        return true, false
    }

    // CASE2: send 隊(duì)列不為空,直接從channel隊(duì)列中獲取
    // sg是sends 線程隊(duì)列
    // 從sends 線程隊(duì)列獲取一個(gè)sudog并喚醒讓其將元素推入channel
    if sg := c.sendq.dequeue(); sg != nil {
        // Found a waiting sender. If buffer is size 0, receive value
        // directly from sender. Otherwise, receive from head of queue
        // and add sender's value to the tail of the queue (both map to
        // the same buffer slot because the queue is full).
        recv(c, sg, ep, func() { unlock(&c.lock) }, 3)
        return true, true
    }
    // CASE3: 緩存隊(duì)列不為空敬尺，此時(shí)只有可能是緩存隊(duì)列已滿笼踩，從隊(duì)列頭取出元素阵子，
    if c.qcount > 0 {
        // Receive directly from queue
        qp := chanbuf(c, c.recvx)
        if raceenabled {
            raceacquire(qp)
            racerelease(qp)
        }
        if ep != nil {
            typedmemmove(c.elemtype, ep, qp)
        }
        typedmemclr(c.elemtype, qp)
        //移動(dòng)channel的recvx下標(biāo)
        c.recvx++
        if c.recvx == c.dataqsiz {
            c.recvx = 0
        }
        c.qcount--
        unlock(&c.lock)
        return true, true
    }

    if !block {
        unlock(&c.lock)
        return false, false
    }


    // CASE4: 緩存隊(duì)列為空荞胡，將 goroutine 加入 recv 隊(duì)列惑惶，并阻塞
    // no sender available: block on this channel.
    gp := getg()
    mysg := acquireSudog()
    mysg.releasetime = 0
    if t0 != 0 {
        mysg.releasetime = -1
    }
    // No stack splits between assigning elem and enqueuing mysg
    // on gp.waiting where copystack can find it.
    mysg.elem = ep
    mysg.waitlink = nil
    gp.waiting = mysg
    mysg.g = gp
    mysg.isSelect = false
    mysg.c = c
    gp.param = nil
    c.recvq.enqueue(mysg)
    goparkunlock(&c.lock, waitReasonChanReceive, traceEvGoBlockRecv, 3)

    // someone woke us up
    if mysg != gp.waiting {
        throw("G waiting list is corrupted")
    }
    gp.waiting = nil
    if mysg.releasetime > 0 {
        blockevent(mysg.releasetime-t0, 2)
    }
    closed := gp.param == nil
    gp.param = nil
    mysg.c = nil
    releaseSudog(mysg)
    return true, !closed
}

從代碼上可以很明顯的看出
receive和send的四種情況相互配合相互對(duì)應(yīng)實(shí)現(xiàn)一存一拿的執(zhí)行順序
close channel 的工作

整個(gè)channel的流程結(jié)構(gòu)：

WechatIMG1513.jpeg

15141548494179_.pic_hd.jpg