路徑為：./src/runtime/chan.go 文件中，先看channel結(jié)構(gòu)體:
type hchan struct {
qcount uint // total data in the queue 當(dāng)前隊列中的數(shù)據(jù)的個數(shù)
dataqsiz uint // size of the circular queue channel環(huán)形隊列的大小
buf unsafe.Pointer // points to an array of dataqsiz elements 存放數(shù)據(jù)的環(huán)形隊列的指針
elemsize uint16 // channel 中存放的數(shù)據(jù)類型的大小|即每個元素的大小
closed uint32 // channel 是否關(guān)閉的標(biāo)示
elemtype *_type // element type channel中存放的元素的類型
sendx uint // send index 當(dāng)前發(fā)送元素指向channel環(huán)形隊列的下標(biāo)指針
recvx uint // receive index 當(dāng)前接收元素指向channel環(huán)形隊列的下標(biāo)指針
recvq waitq // list of recv waiters 等待接收元素的goroutine隊列
sendq waitq // list of send waiters 等待發(fā)送元素的goroutine隊列
// 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.
// 保持此鎖定時不要更改另一個G的狀態(tài)（特別是，沒有準(zhǔn)備好G），因為這可能會因堆棧收縮而死鎖。
lock mutex
}
以及waitq的結(jié)構(gòu)體：
//等待發(fā)送及接收的等待接收元素的goroutine隊列的結(jié)構(gòu)體
type waitq struct {
first *sudog
last *sudog
}
等待發(fā)送或接受goroutine鏈表的結(jié)構(gòu)體sudog:
// sudog表示等待鏈表中的g，例如用于發(fā)送/接收在頻道上。
// 一個G可以出現(xiàn)在許多等待列表中，因此一個G有許多sudog；許多G可能在等待相同的結(jié)果,同步對象，因此一個對象可能有多個sudog。
// sudog是從一個特殊的池中分配的。使用AcquireDog和
// 釋放sudog來分配和釋放它們。
type sudog struct {
// 以下字段受hchan.lock的保護(hù)
g *g // 綁定的goroutine
isSelect bool // isSelect的布爾值表示該線程是否正在進(jìn)行操作channel
next *sudog // 指向下一個等待線程的指針地址
prev *sudog // 指向上一個等待線程的指針地址
elem unsafe.Pointer // data element (may point to stack) 數(shù)據(jù)對象（可能指向棧）
// 當(dāng)進(jìn)行channel的send操作時，elem代表將要保存進(jìn)channel的元素
// 當(dāng)進(jìn)行channel的recv操作時, elem代表從channel接受的元素
// G1執(zhí)行ch<-task4的時候，G1會創(chuàng)建一個sudog然后將elem保存進(jìn)入sendq隊列
// 從不同場景訪問以下字段。
// 對于channel，WaitLink只能由G訪問。
// 對于信號量，所有字段（包括上面的字段）只有在持有semaroot鎖時才能訪問。
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
}
從以上三個結(jié)構(gòu)體我們即可看出channel其實就是由一個環(huán)形數(shù)組實現(xiàn)的隊列用于在確定大小的連續(xù)內(nèi)存塊進(jìn)行數(shù)據(jù)元素的存儲，用waitq以及鏈表sudog共同實現(xiàn)goroutine的等待隊列，并在每個鏈表元素中存儲待從channel中取出或拷貝進(jìn)channel的數(shù)據(jù)元素,可以理解為每個等待線程都是channel的搬運工，負(fù)責(zé)運送數(shù)據(jù).
其中hchan中的lock是 recvq 是讀操作阻塞在 channel 的 goroutine 列表，sendq 是寫操作阻塞在 channel 的 goroutine 列表。
qcount 和 dataqsiz 分別描述了該channel的當(dāng)前使用量和最大容量。
接下來進(jìn)行channel的每一個函數(shù)方法進(jìn)行分析:
makechan:
func makechan(t *chantype, size int) *hchan {
elem := t.elem
// compiler checks this but be safe.
// 判斷定義的channel存儲的每個元素大小是否在范圍內(nèi)
if elem.size >= 1<<16 {
throw("makechan: invalid channel element type")
}
if hchanSize%maxAlign != 0 || elem.align > maxAlign {
throw("makechan: bad alignment")
}
// 計算channel所需要分配的內(nèi)存大小
mem, overflow := math.MulUintptr(elem.size, uintptr(size))
// 判斷內(nèi)存大小是否超過限制
if overflow || mem > maxAlloc-hchanSize || size < 0 {
panic(plainError("makechan: size out of range"))
}
var c *hchan
switch {
// 當(dāng)計算channel的內(nèi)存大小為0時創(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類型非指針
// 當(dāng)計算channel的內(nèi)存大小為0時創(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元素類型包含指針時分配離散的內(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ù)接收兩個參數(shù)，一個是channel里面保存的元素的數(shù)據(jù)類型，一個是緩沖的容量(如果為0表示是非緩沖buffer)，創(chuàng)建流程如下：
根據(jù)傳遞的緩沖大小size是否為零，分別創(chuàng)建不帶buffer的channel或則帶size大小的緩沖channel：
對于不帶緩沖channel，申請一個hchan數(shù)據(jù)結(jié)構(gòu)的內(nèi)存大小；
對于帶緩沖channel，new一個hchan對象，并初始化buffer內(nèi)存；
對于包含指針帶緩存的channel同樣申請一個hchan數(shù)據(jù)結(jié)構(gòu)的內(nèi)存大小;
以及設(shè)置channel的屬性。
帶指針以及不帶指針帶內(nèi)存申請區(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的實現(xiàn)很簡單，主要就是根據(jù)下標(biāo)（sendx或recvx）以及每一個元素的大小還有環(huán)形隊列的指針計算出該下標(biāo)槽點內(nèi)存地址并返回
chansend：
// 通用單通道發(fā)送/接收
// 如果阻塞不是nil，則將不會休眠，但如果無法完成則返回。
// 當(dāng)睡眠中涉及的通道關(guān)閉時，睡眠可以通過g.param == nil喚醒。 最簡單的循環(huán)和重新運行操作; 我們會
// 看到它現(xiàn)在已經(jīng)關(guān)閉了。
func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
// 當(dāng) channel 未初始化或為 nil 時，向其中發(fā)送數(shù)據(jù)將會永久阻塞
if c == nil {
if !block {
return false
}
// gopark 會使當(dāng)前 goroutine 休眠，并通過 unlockf 喚醒，但是此時傳入的 unlockf 為 nil, 因此，goroutine 會一直休眠
gopark(nil, nil, waitReasonChanSendNilChan, traceEvGoStop, 2)
throw("unreachable")
}
if debugChan {
print("chansend: chan=", c, "\n")
}
// 如果開啟了競爭檢測
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
//計時器
if blockprofilerate > 0 {
t0 = cputicks()
}
// 獲取同步鎖
lock(&c.lock)
// 向已經(jīng)關(guān)閉的 channel 發(fā)送消息會產(chǎn)生 panic
if c.closed != 0 {
unlock(&c.lock)
panic(plainError("send on closed channel"))
}
// CASE1: 當(dāng)有 goroutine 在 recv 隊列上等待時，跳過緩存隊列，將消息直接發(fā)給 reciever goroutine
// dequeue 從等待接受的線程隊列鏈表獲取一個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: 緩存隊列未滿，則將消息復(fù)制到緩存隊列上并移動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: 緩存隊列已滿，將goroutine 加入 send 隊列
// 創(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ā)送的線程隊列
c.sendq.enqueue(mysg)
// 休眠線程（即阻塞）
// 通過調(diào)用goready（gp），goroutine可以再次運行。
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 有以下四種情況：【都是對不為nil的chan的情況】
1.向已經(jīng)close的chan寫數(shù)據(jù)，拋panic。
2.有 goroutine 阻塞在 channel recv 隊列上，此時緩存隊列( hchan.buf)為空(即緩沖區(qū)內(nèi)無元素)，直接將消息發(fā)送給 reciever goroutine,只產(chǎn)生一次復(fù)制,從當(dāng)前 channel 的等待隊列中取出等待的 goroutine，然后調(diào)用 send。goready 負(fù)責(zé)喚醒 goroutine。
3.當(dāng) channel 緩存隊列( hchan.buf )有剩余空間時，將數(shù)據(jù)放到隊列里，等待接收，接收后總共產(chǎn)生兩次復(fù)制
4.當(dāng) channel 緩存隊列( hchan.buf )已滿時，將當(dāng)前 goroutine 加入 send 隊列并阻塞
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 隊列不為空,直接從channel隊列中獲取
// sg是sends 線程隊列
// 從sends 線程隊列獲取一個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: 緩存隊列不為空，此時只有可能是緩存隊列已滿，從隊列頭取出元素，
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)
//移動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: 緩存隊列為空，將 goroutine 加入 recv 隊列，并阻塞
// 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的四種情況相互配合相互對應(yīng)實現(xiàn)一存一拿的執(zhí)行順序
close channel 的工作
整個channel的流程結(jié)構(gòu)：
WechatIMG1513.jpeg
15141548494179_.pic_hd.jpg
