chan是我們學習golang繞不開的一個話題，今天我就不講基礎(chǔ)的使用了，因為太多這種文章了，我講一下channel底層的實現(xiàn)和它的數(shù)據(jù)結(jié)構(gòu)

必須了解的數(shù)據(jù)結(jié)構(gòu)

type hchan struct {
    qcount   uint           // 所有數(shù)據(jù)
    dataqsiz uint           // 數(shù)據(jù)size
    buf      unsafe.Pointer // 指向真實數(shù)據(jù)的指針
    elemsize uint16 
    closed   uint32
    elemtype *_type // 數(shù)據(jù)類型
    sendx    uint   // send index
    recvx    uint   // receive index
    recvq    waitq  // list of recv waiters
    sendq    waitq  // list of send waiters
    // 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.
    lock mutex
}
//qcount：代表 chan 中已經(jīng)接收但還沒被取走的元素的個數(shù)。=len（ch）
//dataqsiz：隊列的大小。chan 使用一個循環(huán)隊列來存放元素
// buf：存放元素的循環(huán)隊列的 buffer。
//elemtype 和 elemsize：chan 元素的類型和元素大小，一般分為普通類型和指針類型，在makechan函數(shù)中會判斷是否為指針類型
//sendx：處理發(fā)送數(shù)據(jù)的指針在 buf 中的位置。一旦接收了新的數(shù)據(jù)，指針就會加上 elemsize，移向下一個位置。buf 的總大小是 elemsize 的整數(shù)倍，而且 
//buf 是一個循環(huán)列表。
//recvx：處理接收請求時的指針在 buf 中的位置。一旦取出數(shù)據(jù)，此指針會移動到下一個位置。
//recvq：chan 是多生產(chǎn)者多消費者的模式，如果消費者因為沒有數(shù)據(jù)可讀而阻塞了，就會被加入到 recvq 隊列中。
//sendq：如果生產(chǎn)者因為 buf 滿了而阻塞，會被加入到 sendq 隊列中。

type waitq struct {
    first *sudog
    last  *sudog
}
// 就是gopark掉了的goroutine隊列

//單個節(jié)點結(jié)構(gòu)體
type sudog struct {
    g *g
    isSelect bool
    next     *sudog
    prev     *sudog
    elem     unsafe.Pointer // data element (may point to stack)
    acquiretime int64
    releasetime int64
    ticket      uint32
    parent      *sudog // semaRoot binary tree
    waitlink    *sudog // g.waiting list or semaRoot
    waittail    *sudog // semaRoot
    c           *hchan // channel
}

在說channel之前你應(yīng)該了解的

用通俗一點的話來說，channel實際底層并發(fā)是靠鎖lock實現(xiàn)的，數(shù)據(jù)寫入buf和從buf讀出來都有加鎖的動作，
存儲數(shù)據(jù)是靠一個循環(huán)隊列來保存的，隊列的大小就是屬性buf指向的隊列長度,隊列長度（len）就是qcount,
實際存儲數(shù)據(jù)是buf的指向的循環(huán)隊列，然后sendx和recvx是來控制數(shù)據(jù)讀取和刪除的，順序就是按照buf的
索引順序來的。然后還會保存兩個隊列：sendq，recvq,這兩個參數(shù)實際上是綁定wait狀態(tài)下的g，也就是
被park掉的goroutine,要理解這句話就要有點gmp模型的概念。

1.我們創(chuàng)建一個buffer channel

image.png

2.此時的結(jié)構(gòu)體應(yīng)該是這樣的

image.png

3.放入一個數(shù)據(jù)到chan

image.png

4.取走一個數(shù)據(jù)

image.png

5.那么簡單的存放數(shù)據(jù)知道了，它是如何將goroutine阻塞的呢---先將一下GMP

因為這個圖是網(wǎng)上扣下來的，不知道原圖是誰的，感覺畫的很不錯

image.png

image.png

這里在做個簡單的解釋：后面專門會出一個文章來講解一下GMP調(diào)度模型。M是一個內(nèi)核線程的一個映射（實際上的原理比這個復雜許多，涉及到線程模型，后面寫篇文章細講），P是我們的調(diào)度器，一般又稱為schedule，正常情況下，我們在代碼執(zhí)行寫了一個go func,那么這個go在代碼編譯的時候就會被丟入一個P的隊列中：就像一堆排隊的地鼠G去領(lǐng)錢，P就是包工頭，第一個地鼠把自己的挖的洞的照片給P看，P看完就給地鼠拍拍灰（準備環(huán)境），就讓G去M里面去領(lǐng)錢，（領(lǐng)錢就是執(zhí)行），M呢是個小金庫，但是每個地鼠只有10ms時間去領(lǐng)錢，過了就會被p強制的給退出來，有的地鼠不用10ms就能自己退出來

chan如何將goroutine阻塞的

image.png

這里需要了解一下gopark：將goroutine進行休眠

此時GMP應(yīng)該是這個樣子

image.png

此時G就沒有人執(zhí)行了，所以就看起來像卡住了一樣，那么就有個問題了：怎么讓它繼續(xù)運行呢，我們不是將chan讀出來goroutine又能繼續(xù)執(zhí)行了嗎？

此時Chan的結(jié)構(gòu)體變成了這個樣子

image.png

哪些被游離的gorouine 就被chan的隊列給抓住了，放到它的發(fā)送或者等待隊列中。

此時這里sendq/recvq長這個樣子

image.png

這里怎么讀呢：我就簡單解釋一下，沒這個圖，我想自己畫一個發(fā)現(xiàn)太丑了
此時buf的被別人取走了一個元素，那么就從sendqpop一個出來，是從頭部開始pop，然后元素放到了buf，goroutine被設(shè)置成runable狀態(tài)，然后放到P的runable隊列中去繼續(xù)執(zhí)行下文。
上面是說的發(fā)送chan被占滿

如果先來讀chan被阻塞，chan內(nèi)部長什么樣子呢

image.png

那么chan怎么處理先讀后寫的這種場景

按照上面的套路，其實我們可以想，寫的時候，先寫到buf，然后recvq檢測到buf有值了，將buf 的數(shù)據(jù)pop出來，將goroutine喚醒，數(shù)據(jù)寫入elem指針指向的地址。但是有沒有更好的套路呢？

其實根據(jù)場景：先有等待者，然后發(fā)送者來了，就像去銀行存錢一樣，取錢的人來了，但是銀行沒錢，后面存錢的人來了。最大的不同就是gorotuine不需要到chan，就想存錢取錢的人可以不用在銀行操作，那么是不是存錢的人就可以直接把錢給取錢的人

實際上的代碼也是這么來處理的，就是send 的goroutine直接寫了recv的stack，就減少了lock的啟動和釋放，提高了性能。但是也只有only operations in runtime where this happen：運行時發(fā)生這樣的情況。

• 基本上所有的流程就是上面這個圖片，下面是比較重要的一些方法，可以按照上面理解去下面看看，不過最好能用goland調(diào)試一下，看代碼很難看懂，我在看源碼的時候，確認chan 數(shù)據(jù)類型是指針還是普通數(shù)據(jù)就踩坑了，初次創(chuàng)建的時候調(diào)用下面的makechan，我是直接跑到runtime里面去看，發(fā)現(xiàn)斷點沒有走預(yù)期路線，我還調(diào)試了好久，才發(fā)現(xiàn)它的上層是反復調(diào)用這個方法，調(diào)用了3次，才返回實例，這里面我就不細細的講了，大家有什么建議也可以評論一下，大家共同學習進步

創(chuàng)建chan:makechan

func makechan(t *chantype, size int) *hchan {
    elem := t.elem
    // compiler checks this but be safe.
    if elem.size >= 1<<16 { //判斷元素的大小
        throw("makechan: invalid channel element type")
    }
    if hchanSize%maxAlign != 0 || elem.align > maxAlign {
        thro("makechan: bad alignment")
    }
    mem, overflow := math.MulUintptr(elem.size, uintptr(size))
    if overflow || mem > maxAlloc-hchanSize || size < 0 {
        panic(plainError("makechan: size out of range"))
    }
    // Hchan does not contain pointers interesting for GC when elements stored in buf do not contain pointers.
    // buf points into the same allocation, elemtype is persistent.
    // SudoG's are referenced from their owning thread so they can't be collected.
    // TODO(dvyukov,rlh): Rethink when collector can move allocated objects.
    var c *hchan
    switch {
    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()
    case elem.ptrdata == 0:
        // Elements do not contain pointers.
        // Allocate hchan and buf in one call.
        c = (*hchan)(mallocgc(hchanSize+mem, nil, true))
        c.buf = add(unsafe.Pointer(c), hchanSize)
    default:
        // Elements contain pointers.
        c = new(hchan)
        c.buf = mallocgc(mem, elem, true)
    }
    c.elemsize = uint16(elem.size)
    c.elemtype = elem
    c.dataqsiz = uint(size)
    lockInit(&c.lock, lockRankHchan)
    if debugChan {
        print("makechan: chan=", c, "; elemsize=", elem.size, "; dataqsiz=", size, "\n")

    }
    return c
}

chansend

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
    if c == nil {
        if !block {
            return false
        }
        gopark(nil, nil, waitReasonChanSendNilChan, traceEvGoStop, 2)
        throw("unreachable")
    }
    if debugChan {
        print("chansend: chan=", c, "\n")
    }
    if raceenabled {
        racereadpc(c.raceaddr(), callerpc, funcPC(chansend))
    }
    if !block && c.closed == 0 && full(c) {
        return false
    }
    var t0 int64
    if blockprofilerate > 0 {
        t0 = cputicks()
    }
    lock(&c.lock)
    if c.closed != 0 {
        unlock(&c.lock)
        panic(plainError("send on closed channel"))
    }
    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
    }
    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
    }
    // Block on the channel. Some receiver will complete our operation for us.
    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
    c.sendq.enqueue(mysg)
    // Signal to anyone trying to shrink our stack that we're about
    // to park on a channel. The window between when this G's status
    // changes and when we set gp.activeStackChans is not safe for
    // stack shrinking.
    atomic.Store8(&gp.parkingOnChan, 1)
    gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanSend, traceEvGoBlockSend, 2)
    // 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
    gp.activeStackChans = false
    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
    releaseSudog(mysg)
    return true
}

close chan

func closechan(c *hchan) {
    if c == nil {
        panic(plainError("close of nil channel"))
    }
    lock(&c.lock)
    if c.closed != 0 {
        unlock(&c.lock)
        panic(plainError("close of closed channel"))
    }
    if raceenabled {
        callerpc := getcallerpc()
        racewritepc(c.raceaddr(), callerpc, funcPC(closechan))
        racerelease(c.raceaddr())
    }
    c.closed = 1
    var glist gList
    // release all readers
    for {
        sg := c.recvq.dequeue()
        if sg == nil {
            break
        }
        if sg.elem != nil {
            typedmemclr(c.elemtype, sg.elem)
            sg.elem = nil
        }
        if sg.releasetime != 0 {
            sg.releasetime = cputicks()
        }
        gp := sg.g
        gp.param = nil
        if raceenabled {
            raceacquireg(gp, c.raceaddr())
        }
        glist.push(gp)
    }
    // release all writers (they will panic)
    for {
        sg := c.sendq.dequeue()
        if sg == nil {
            break
        }
        sg.elem = nil
        if sg.releasetime != 0 {
            sg.releasetime = cputicks()
        }
        gp := sg.g
        gp.param = nil
        if raceenabled {
            raceacquireg(gp, c.raceaddr())
        }
        glist.push(gp)
    }
    unlock(&c.lock)
    // Ready all Gs now that we've dropped the channel lock.
    for !glist.empty() {
        gp := glist.pop()
        gp.schedlink = 0
        goready(gp, 3)
    }
}

chanrecv方法

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")
    }
    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.
    if !block && empty(c) {
        if atomic.Load(&c.closed) == 0 {
            return
        }
        if empty(c) {
            // The channel is irreversibly closed and empty.
            if raceenabled {
                raceacquire(c.raceaddr())
            }
            if ep != nil {
                typedmemclr(c.elemtype, ep)
            }
            return true, false
        }
    }
    var t0 int64
    if blockprofilerate > 0 {
        t0 = cputicks()
    }
    lock(&c.lock)
    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
    }
    if sg := c.sendq.dequeue(); sg != nil {
        recv(c, sg, ep, func() { unlock(&c.lock) }, 3)
        return true, true
    }
    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)
        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
    }
    // 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)
    atomic.Store8(&gp.parkingOnChan, 1)
    gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanReceive, traceEvGoBlockRecv, 2)
    // someone woke us up
    if mysg != gp.waiting {
        throw("G waiting list is corrupted")
    }
    gp.waiting = nil
    gp.activeStackChans = false
    if mysg.releasetime > 0 {
        blockevent(mysg.releasetime-t0, 2)
    }
    closed := gp.param == nil
    gp.param = nil
    mysg.c = nil
    releaseSudog(mysg)
    return true, !closed
}