[toc]
基礎(chǔ)概念
Channel 是 Golang 的核心類型，常用于多個(gè) Goroutine 之間的通信。可以把 Channel 理解成是一個(gè)單向的管道，具有 FIFO 特性。
image.png
Channel 是有容量限制的
- 當(dāng)容量是 0 時(shí)，稱為無緩沖 Channel。發(fā)送和接收只有一方就緒時(shí)，就緒方會被阻塞直到另一方也就緒。
- 當(dāng)容量大于 0 時(shí)，稱為有緩沖 Channel。當(dāng)傳輸中的元素個(gè)數(shù)超過容量時(shí)，發(fā)送方將會被阻塞直到有可用的緩沖空間出現(xiàn)；當(dāng)傳輸中的元素個(gè)數(shù)為 0 時(shí)，消費(fèi)方將會被阻塞直到緩沖空間出現(xiàn)新的數(shù)據(jù)。
數(shù)據(jù)結(jié)構(gòu)
type hchan struct {
qcount uint // total data in the queue
dataqsiz uint // size of the circular queue
buf unsafe.Pointer // points to an array of dataqsiz elements
elemsize uint16
closed uint32
elemtype *_type // element type
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，緩沖隊(duì)列的大小，記錄實(shí)際元素?cái)?shù)量
- dataqsiz，緩沖隊(duì)列的容量，記錄最大可存儲元素?cái)?shù)量
- buf，指向環(huán)形緩沖隊(duì)列的指針
- elemsize，每個(gè)元素的大小
- closed，記錄 channel 的關(guān)閉狀態(tài)
- elemtype，元素的類型
- sendx，緩沖隊(duì)列中即將發(fā)送的數(shù)據(jù)下標(biāo)
- recvx，緩沖隊(duì)列中即將接收的數(shù)據(jù)下標(biāo)
- recvq，等待從 channel 接收數(shù)據(jù)的 goroutine 雙向鏈表
- sendq，等待向 channel 發(fā)送數(shù)據(jù)的 goroutine 雙向鏈表
- lock，多 goroutine 讀寫的并發(fā)保護(hù)鎖
圖解發(fā)送數(shù)據(jù)
image.png
注：無緩沖 Channel 原理類似不做贅述
圖解接收數(shù)據(jù)
image.png
注：無緩沖 Channel 原理類似不做贅述
源碼解讀
發(fā)送數(shù)據(jù)
func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
...
lock(&c.lock)
if c.closed != 0 {
unlock(&c.lock)
panic(plainError("send on closed channel"))
}
if sg := c.recvq.dequeue(); sg != nil { // 關(guān)鍵點(diǎn)1
// 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 { // 關(guān)鍵點(diǎn)2
// Space is available in the channel buffer. Enqueue the element to send.
qp := chanbuf(c, c.sendx)
...
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
}
// 關(guān)鍵點(diǎn)3
// 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)
...
return true
}
func send(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {
...
if sg.elem != nil {
sendDirect(c.elemtype, sg, ep)
sg.elem = nil
}
gp := sg.g
unlockf()
gp.param = unsafe.Pointer(sg)
sg.success = true
if sg.releasetime != 0 {
sg.releasetime = cputicks()
}
goready(gp, skip+1)
}
關(guān)鍵點(diǎn)
- 當(dāng) recvq 有等待的接收者時(shí)，說明緩沖隊(duì)列是空的，則將數(shù)據(jù)直接發(fā)送給接收者，然后將接收者的 Goroutine 標(biāo)記成可運(yùn)行的狀態(tài)，并加入到本地可運(yùn)行隊(duì)列中。
- 當(dāng)緩沖隊(duì)列未滿時(shí)，則將數(shù)據(jù)直接寫入緩沖隊(duì)列。
- 當(dāng)緩沖隊(duì)列滿了或者無緩沖隊(duì)列時(shí)，則將發(fā)送數(shù)據(jù)的指針和當(dāng)前 Goroutine 等信息組裝成 sudog 并加入到 sendq 中，等待合適機(jī)會執(zhí)行。
接收數(shù)據(jù)
func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
...
lock(&c.lock)
...
if sg := c.sendq.dequeue(); sg != nil { // 關(guān)鍵點(diǎn)1
// 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
}
if c.qcount > 0 { // 關(guān)鍵點(diǎn)2
// Receive directly from queue
qp := chanbuf(c, c.recvx)
...
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
}
// 關(guān)鍵點(diǎn)3
// 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)
// 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), waitReasonChanReceive, traceEvGoBlockRecv, 2)
...
return true, success
}
func recv(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {
if c.dataqsiz == 0 {
...
if ep != nil {
// copy data from sender
recvDirect(c.elemtype, sg, ep)
}
} else {
// Queue is full. Take the item at the
// head of the queue. Make the sender enqueue
// its item at the tail of the queue. Since the
// queue is full, those are both the same slot.
qp := chanbuf(c, c.recvx)
...
// copy data from queue to receiver
if ep != nil {
typedmemmove(c.elemtype, ep, qp)
}
// copy data from sender to queue
typedmemmove(c.elemtype, qp, sg.elem)
c.recvx++
if c.recvx == c.dataqsiz {
c.recvx = 0
}
c.sendx = c.recvx // c.sendx = (c.sendx+1) % c.dataqsiz
}
sg.elem = nil
gp := sg.g
unlockf()
gp.param = unsafe.Pointer(sg)
sg.success = true
if sg.releasetime != 0 {
sg.releasetime = cputicks()
}
goready(gp, skip+1)
}
關(guān)鍵點(diǎn)
- 當(dāng) sendq 有等待的發(fā)送者時(shí)，如果是無緩沖隊(duì)列，則直接從發(fā)送者獲取數(shù)據(jù)；如果是緩沖隊(duì)列滿了，則從緩沖隊(duì)列取出一個(gè)數(shù)據(jù)，然后將發(fā)送者的數(shù)據(jù)寫入緩沖隊(duì)列。最后將發(fā)送者的 Goroutine 標(biāo)記成可運(yùn)行的狀態(tài)，并加入到本地可運(yùn)行隊(duì)列中。
- 當(dāng)緩沖隊(duì)列有數(shù)據(jù)時(shí)，則直接從緩沖隊(duì)列讀取數(shù)據(jù)。
- 當(dāng)緩沖無數(shù)據(jù)時(shí)，則將接收數(shù)據(jù)的指針和當(dāng)前 Goroutine 等信息組裝成 sudog 并加入到 recvq 中，等待合適機(jī)會執(zhí)行。
