GoLang?channel底層代碼分析詳解
以下源碼都摘自 golang 1.16.15 版本。
1. channel 底層結(jié)構(gòu)
Golang 中的 channel 對(duì)應(yīng)的底層結(jié)構(gòu)為 hchan 結(jié)構(gòu)體(channel的源碼位置在Golang包的 runtime/chan.go):
type hchan struct { qcount uint // buf當(dāng)前元素的數(shù)量 dataqsiz uint // buf的容量 buf unsafe.Pointer // channel緩沖區(qū),一個(gè)循環(huán)數(shù)組 elemsize uint16 // 元素大小 closed uint32 // channel關(guān)閉標(biāo)記 elemtype *_type // element type sendx uint // 當(dāng)下一次發(fā)送數(shù)據(jù)到channel時(shí),數(shù)據(jù)存放到buf中的哪個(gè)index recvx uint // 當(dāng)下一次從channel接收數(shù)據(jù)時(shí),從buf的哪個(gè)index獲取數(shù)據(jù) recvq waitq // 等待接收數(shù)據(jù)的goroutine列表,雙向鏈表 sendq waitq // 等待發(fā)送數(shù)據(jù)的goroutine列表,雙向鏈表 lock mutex // 互斥鎖,發(fā)送和接收操作前需要獲取的鎖,所以channel的發(fā)送和接收操作是互斥的 }
如果 dataqsiz == 0 時(shí),則為無(wú)緩沖 channel,如果 dataqsiz > 0 時(shí),則為有緩沖 channel。
其中 recvq 和 sendq 是一個(gè)雙向鏈表結(jié)構(gòu),鏈表中的元素為 sudog 結(jié)構(gòu)體,其中該結(jié)構(gòu)體中保存了g,所以本質(zhì)上recvq 和 sendq 是保存了等待接收/發(fā)送數(shù)據(jù)的goroutine列表。
channel 中的 recvq 和 sendq 的使用場(chǎng)景如下所示:
在從 channel 接收數(shù)據(jù)時(shí) (data := <- ch),如果 sendq 中沒有等待發(fā)送數(shù)據(jù)的 goroutine,且 buf 中沒有數(shù)據(jù)時(shí),則需要把當(dāng)前 goroutine 保存到 recvq 列表中,并掛起。
在向 channel 發(fā)送數(shù)據(jù)時(shí) (ch <- data),如果 recvq 中沒有等待接收數(shù)據(jù)的 goroutine,且 buf 滿了的情況下,則需要把當(dāng)前 goroutine 保存到 sendq 列表中,并掛起。
type waitq struct { first *sudog last *sudog }
// sudog表示等待隊(duì)列中的一個(gè)g,例如在一個(gè)channel中的發(fā)送/接收。 // sudog是必要的,因?yàn)間和同步對(duì)象的關(guān)系是多對(duì)多的,一個(gè)g可以在多個(gè)等待隊(duì)列中,因此一個(gè)g會(huì)有很多個(gè)sudog, // 很多g可能在等待著同一個(gè)同步對(duì)象,因此一個(gè)對(duì)象可能有多個(gè)sudog。 // sudog是從一個(gè)特殊的池中分配的,使用acquireSudog和releaseSudog分配和釋放它們。 type sudog struct { // 以下字段受此sudog阻塞的channel的hchan.lock保護(hù) g *g next *sudog prev *sudog elem unsafe.Pointer // data element (may point to stack) // 以下字段永遠(yuǎn)不會(huì)被同時(shí)訪問 // 對(duì)于channel,waitlink只能被g訪問 // 對(duì)于信號(hào)量,所有字段(包括上述字段)只有在持有semaRoot鎖時(shí)才能訪問。 acquiretime int64 releasetime int64 ticket uint32 // isSelect表示g正在參與選擇,因此g.selectDone必須經(jīng)過CAS處理,才能被喚醒 isSelect bool // success表示通過channel c的通信是否成功。 // 如果goroutine因?yàn)橥ㄟ^channel c傳遞了一個(gè)值而被喚醒,則為true // 如果因?yàn)閏被關(guān)閉而喚醒,則為false success bool parent *sudog // semaRoot binary tree waitlink *sudog // g.waiting list or semaRoot waittail *sudog // semaRoot c *hchan // channel }
channel 結(jié)構(gòu)圖:
2. channel 的創(chuàng)建
// 無(wú)緩沖channel ch := make(chan int) // 緩沖大小為5的channel ch2 := make(chan int, 5)
創(chuàng)建 channel 的源碼為runtime/chan.go文件中的 makechan 函數(shù):
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 { throw("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: // 隊(duì)列或元素大小為0,即無(wú)緩沖channel c = (*hchan)(mallocgc(hchanSize, nil, true)) // Race detector uses this location for synchronization. c.buf = c.raceaddr() case elem.ptrdata == 0: // 元素不包含指針類型,只進(jìn)行一次 hchan 和 buf 的內(nèi)存分配 // 當(dāng)存儲(chǔ)在buf中的元素不包含指針時(shí),GC就不會(huì)掃描hchan中的元素 c = (*hchan)(mallocgc(hchanSize+mem, nil, true)) c.buf = add(unsafe.Pointer(c), hchanSize) default: // 元素中包含指針類型,進(jìn)行2次內(nèi)存分配操作 // 用new分配內(nèi)存返回的是指針 c = new(hchan) c.buf = mallocgc(mem, elem, true) } // 初始化channel數(shù)據(jù) 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") } // 返回 hchan 的指針類型 return c }
注意這里返回的是 hchan 的指針,因此我們?cè)诤瘮?shù)間可以直接傳遞 channel,而不用傳遞channel的指針了。
另外,因?yàn)閏hannel 的內(nèi)存分配都用到了 mallocgc 函數(shù),而 mallocgc 是負(fù)責(zé)堆內(nèi)存分配的關(guān)鍵函數(shù),因此可見 channel 是分配在堆內(nèi)存上的。
3. channel 的發(fā)送流程
channel 的發(fā)送:
ch <- data
channel 發(fā)送的源碼對(duì)應(yīng) runtime/chan.go 的 chansend 函數(shù):
func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool { // 如果當(dāng)前channel是nil if c == nil { // 如果不阻塞,則直接返回false if !block { return false } // 掛起當(dāng)前goroutine gopark(nil, nil, waitReasonChanSendNilChan, traceEvGoStop, 2) throw("unreachable") } if debugChan { print("chansend: chan=", c, "\n") } if raceenabled { racereadpc(c.raceaddr(), callerpc, funcPC(chansend)) } // 這里訪問了hchan結(jié)構(gòu)中的closed, full函數(shù)內(nèi)部訪問了dataqsiz,recvq,qcount字段,這里沒有加鎖,是為什么呢? // 先說說這里判斷的含義:如果不阻塞,且channel沒有被關(guān)閉,且buf已滿,則快速返回false,表示數(shù)據(jù)發(fā)送失敗。 // 因?yàn)闆]有加鎖,假如在判斷c.closed == 0之后結(jié)果為true,在判斷full之前,這時(shí)channel被其他goroutine關(guān)閉了, // 然后full函數(shù)返回了true,那么它會(huì)直接return false,這樣子會(huì)有什么影響呢? // 其實(shí)并沒有什么影響,在這種情況下返回false也是合理的,因?yàn)槎际潜硎驹诓蛔枞那闆r下發(fā)送數(shù)據(jù)失敗。 // 所以這里訪問hchan里面的數(shù)據(jù)就沒有加鎖了 if !block && c.closed == 0 && full(c) { return false } var t0 int64 if blockprofilerate > 0 { t0 = cputicks() } // 鎖住channel,可見channel是并發(fā)安全的 lock(&c.lock) // 如果channel已關(guān)閉,則panic if c.closed != 0 { unlock(&c.lock) panic(plainError("send on closed channel")) } // 如果recvq等待接收隊(duì)列中有值,則直接把值傳給等待接收的goroutine,這樣可以減少一次內(nèi)存拷貝 if sg := c.recvq.dequeue(); sg != nil { send(c, sg, ep, func() { unlock(&c.lock) }, 3) return true } // 如果recvq等待接收隊(duì)列中沒有值,且為有緩沖channel,則把數(shù)據(jù)copy到buf中 if c.qcount < c.dataqsiz { // Space is available in the channel buffer. Enqueue the element to send. qp := chanbuf(c, c.sendx) if raceenabled { racenotify(c, c.sendx, nil) } typedmemmove(c.elemtype, qp, ep) c.sendx++ // 因?yàn)閎uf是環(huán)形數(shù)組,所以如果sendx超出了最大index,就要?dú)w0 if c.sendx == c.dataqsiz { c.sendx = 0 } c.qcount++ unlock(&c.lock) return true } // 如果recvq等待接收隊(duì)列中沒有值,且為無(wú)緩沖channel,且不阻塞,則直接返回false if !block { unlock(&c.lock) return false } // 接下來做阻塞當(dāng)前goroutine的一些準(zhǔn)備工作,構(gòu)造一個(gè)sudog // 獲取當(dāng)前goroutine的指針 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 // 把構(gòu)建好的 sudog 加到 sendq 發(fā)送等待隊(duì)列中 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) // 掛起當(dāng)前goroutine gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanSend, traceEvGoBlockSend, 2) // 如果當(dāng)前 goroutine 被喚醒后,會(huì)在這里繼續(xù)執(zhí)行 // 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 closed := !mysg.success gp.param = nil if mysg.releasetime > 0 { blockevent(mysg.releasetime-t0, 2) } mysg.c = nil releaseSudog(mysg) if closed { if c.closed == 0 { throw("chansend: spurious wakeup") } // 如果喚醒后,發(fā)現(xiàn) channel 被關(guān)閉,則關(guān)閉 panic(plainError("send on closed channel")) } return true }
full 函數(shù),用于判斷當(dāng)前channel是否還有坑位接收待發(fā)送的數(shù)據(jù):
// 判斷channel中是否還有位置存放數(shù)據(jù) func full(c *hchan) bool { // 如果是非緩沖channel if c.dataqsiz == 0 { // 如果 recvq 中沒有等待接收數(shù)據(jù)的 goroutine,則返回 true,表示已滿,否則返回 false return c.recvq.first == nil } // 如果是有緩沖 channel,則判斷buf是否已滿 return c.qcount == c.dataqsiz }
send 函數(shù),在recvq中有等待接收數(shù)據(jù)的goroutine時(shí)會(huì)被調(diào)用:
// 在一個(gè)空的 channel c 中完成發(fā)送操作 // 把數(shù)據(jù) ep 從發(fā)送者復(fù)制到接收者 sg 中 // 最后接收的 goroutine 會(huì)被喚醒 // channel c 一定是空的且被鎖住的 // sg 一定是已經(jīng)從 c 的 recvq 中出隊(duì)了 // eq 一定是不等于 nil 的,且指向堆或者是調(diào)用者的棧 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. racenotify(c, c.recvx, nil) racenotify(c, c.recvx, sg) c.recvx++ if c.recvx == c.dataqsiz { c.recvx = 0 } c.sendx = c.recvx // c.sendx = (c.sendx+1) % c.dataqsiz } } // sg.elem 指向接收者存放接收數(shù)據(jù)的存放的位置 if sg.elem != nil { // 直接內(nèi)存拷貝,從發(fā)送者拷貝到接收者內(nèi)存 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() } // 喚醒接收數(shù)據(jù)的goroutine goready(gp, skip+1) }
總結(jié) channel 的發(fā)送流程:
判斷 channel 是否是 nil,如果是,則會(huì)永久阻塞導(dǎo)致死鎖報(bào)錯(cuò)
如果 channel 中 recvq 存在接收者 goroutine,則直接把需要發(fā)送的數(shù)據(jù)拷貝到接收 goroutine,這里其實(shí)是有sodog 的結(jié)構(gòu),里面保存了接受者goroutine的指針。
如果 recvq 中不存在接收者:
a. 如果 buf 沒有滿,則直接把數(shù)據(jù)拷貝到 buf 的 sendx 位置
b. 如果 channel 為無(wú)緩沖 channel 或 buf 已滿,則把當(dāng)前 goroutine 保存到 sendq 等待隊(duì)列中,阻塞當(dāng)前 goroutine
4. channel 的接收流程
channel 的接收:
data := <- ch
data2, ok := <- ch
channel 的接收分別有2個(gè)函數(shù),其中一種是帶”ok“返回值的,另外一種是不帶"ok"返回值的。
- 帶”ok"返回值的函數(shù),該返回的布爾值為 true 時(shí),并不表示當(dāng)前通道還沒有關(guān)閉,而是僅僅表示當(dāng)前獲取到的值是通道的正常生產(chǎn)出來的數(shù)據(jù),而不是零值;當(dāng)該布爾值為 false 時(shí),表示當(dāng)前的通道已經(jīng)被關(guān)閉,并且獲取到的值是零值。
- 不帶"ok"返回值的函數(shù),當(dāng) channel 被關(guān)閉時(shí),就不能判斷當(dāng)前獲取到的值是 channel 正常生產(chǎn)的值,還是零值了。
// 無(wú)返回值 func chanrecv1(c *hchan, elem unsafe.Pointer) { chanrecv(c, elem, true) } // 返回 bool 類型,如果返回false,表示 channel 已經(jīng)被關(guān)閉,否則返回false。 func chanrecv2(c *hchan, elem unsafe.Pointer) (received bool) { _, received = chanrecv(c, elem, true) return }
不管是否返回 received,channel 的接收都調(diào)用了 chanrecv 函數(shù):
// 從 channel c 中接收數(shù)據(jù),并把數(shù)據(jù)復(fù)制到 ep 中。 // 在忽略接收數(shù)據(jù)的情況下,eq 可能是 nil,例如:<- ch // 如果不阻塞,且 channel 中沒有元素的情況下,直接快速返回(false, false) // 如果 c 已經(jīng)被關(guān)閉,*ep 為零值,怎返回(true, false) // 如果 *ep 中有元素,則返回(true, true) // 一個(gè)不等于 nil 的 eq 一定指向堆或者調(diào)用者的棧 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 } // 如果 c 為 nil,掛起當(dāng)前 goroutine gopark(nil, nil, waitReasonChanReceiveNilChan, traceEvGoStop, 2) throw("unreachable") } // Fast path: check for failed non-blocking operation without acquiring the lock. // 在非阻塞模式下,快速檢測(cè)接收失敗的情況 if !block && empty(c) { // 發(fā)現(xiàn) channel 沒有準(zhǔn)備好要接收數(shù)據(jù)后,我們觀察通道是否已經(jīng)關(guān)閉。 // 重新排序這些檢查可能會(huì)導(dǎo)致在關(guān)閉時(shí)不正確的行為。 // 例如,如果通道是open,且not empty,然后被關(guān)閉,接著排空->empty, // 重新排序的讀取可能會(huì)錯(cuò)誤地表示成”open和empty“。 // 為了防止重排序,我們對(duì)這2個(gè)檢查都使用原子加載,并依靠清空和關(guān)閉發(fā)生在同一個(gè)鎖下的不同臨界區(qū)。 // 當(dāng)關(guān)閉帶有阻塞發(fā)送的非緩沖channel,此假設(shè)失敗,但這無(wú)論如何都是錯(cuò)誤的條件。 if atomic.Load(&c.closed) == 0 { // 因?yàn)?channel 不能重新打開,所以在后面這里觀察到 channel 沒有被關(guān)閉,意味著它在第一次判斷 empty 的時(shí)候也沒有關(guān)閉。 // 這樣就表現(xiàn)得像在第一次判斷 empty 時(shí),通道也沒有關(guān)閉:if empty(c) && atomic.Load(&c.closed) == 0 {...} return } // 當(dāng)執(zhí)行到這里的時(shí)候,說明 channel 已經(jīng)被關(guān)閉了。 // 這時(shí)重新檢查通道是否還有其他待接收的數(shù)據(jù),這些數(shù)據(jù)可能在第一次 empty 檢查和通道關(guān)閉檢查之間到達(dá)。 // 在這種情況下發(fā)送時(shí),也需要按照連貫的順序。 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) // 如果 channel c 已經(jīng)被關(guān)閉,且 buf 中無(wú)元素,將獲取到零值 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 } // 如果 sendq 中有元素 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). // 找到一個(gè)正在等待的發(fā)送者。 // 1.如果是無(wú)緩沖 channel,則直接把從發(fā)送者那里接收數(shù)據(jù)。 // 2.如果是有緩沖 channel,這時(shí) sendq 中有元素,說明 buf 滿了,發(fā)送者需要等待消費(fèi)者消費(fèi) buf 數(shù)據(jù)后才能繼續(xù)發(fā)送數(shù)據(jù)。 // 這時(shí)當(dāng)前的 goroutine 會(huì)從 buf 的 recvx 位置接收數(shù)據(jù),并且把剛剛獲取到的發(fā)送者 sg 的發(fā)送數(shù)據(jù)拷貝到 buf 的 sendx 位置中。 recv(c, sg, ep, func() { unlock(&c.lock) }, 3) return true, true } // sendq 中沒有等待的發(fā)送者,且 buf 中有數(shù)據(jù),則直接從 buf 中接收數(shù)據(jù) if c.qcount > 0 { // Receive directly from queue qp := chanbuf(c, c.recvx) if raceenabled { racenotify(c, c.recvx, nil) } 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 } // 如果代碼運(yùn)行到這里,說明 channel 中沒有數(shù)據(jù)可以接收了,接下來就要準(zhǔn)備阻塞當(dāng)前 goroutine 了 // 如果不阻塞,則快速返回 if !block { // 解鎖 unlock(&c.lock) return false, false } // no sender available: block on this channel. // 構(gòu)造sudog // 獲取當(dāng)前 goroutine 指針 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 // 把構(gòu)造好的 sudog 入隊(duì) recvq 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) // 掛起當(dāng)前 goroutine gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanReceive, traceEvGoBlockRecv, 2) // 如果 goroutine 被喚醒,會(huì)從這里開始繼續(xù)執(zhí)行 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) } success := mysg.success gp.param = nil mysg.c = nil releaseSudog(mysg) return true, success }
empty 函數(shù)用于判斷從 channel c 中讀取數(shù)據(jù)是否會(huì)阻塞:
func empty(c *hchan) bool { // c.dataqsiz 是不會(huì)被改變的. if c.dataqsiz == 0 { return atomic.Loadp(unsafe.Pointer(&c.sendq.first)) == nil } return atomic.Loaduint(&c.qcount) == 0 }
recv 函數(shù)在 channel c 的 buf 是滿的,且 sendq 中有等待發(fā)送的 goroutine 時(shí)會(huì)被調(diào)用:
// 這里分為 2 個(gè)部分: // 1.發(fā)送者 sg 待發(fā)送的值會(huì)被放入通道 buf 中,發(fā)送者被喚醒繼續(xù)執(zhí)行 // 2.接收方(當(dāng)前 goroutine)接收的值寫入 ep // 對(duì)于同步 channel(無(wú)緩沖),2 個(gè)值都是一樣的 // 對(duì)于異步 channel(有緩沖),接收方從 channel buf 獲取數(shù)據(jù),發(fā)送方的數(shù)據(jù)放入 channel buf // channel c 一定是滿的,且已被鎖定,recv 用 unlockf 解鎖 channel c。 // sg 一定已經(jīng)從 sendq 出隊(duì) // 不等于 nil 的 ep 一定指向堆或調(diào)用者的棧 func recv(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) { if c.dataqsiz == 0 { if raceenabled { racesync(c, sg) } if ep != nil { // 非緩沖 channel,直接從發(fā)送方接收數(shù)據(jù) recvDirect(c.elemtype, sg, ep) } } else { // 緩沖 channel,buf 已滿 // 先從 buf 隊(duì)列頭部接收數(shù)據(jù),然后把獲取出來的發(fā)送方數(shù)據(jù)入隊(duì) qp := chanbuf(c, c.recvx) if raceenabled { racenotify(c, c.recvx, nil) racenotify(c, c.recvx, sg) } // 從 buf 中復(fù)制數(shù)據(jù)到接收方 if ep != nil { typedmemmove(c.elemtype, ep, qp) } // 把發(fā)送方 sg 的數(shù)據(jù)復(fù)制到 buf 中 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() } // 喚醒發(fā)送方 goroutine goready(gp, skip+1) }
總結(jié) channel 的接收流程:
判斷 channel 是否是 nil,如果是,則會(huì)永久阻塞導(dǎo)致死鎖報(bào)錯(cuò)如果 channel 中 sendq 有等待發(fā)送數(shù)據(jù)的 goroutine:
a. 如果是無(wú)緩存 channel,則直接把要發(fā)送的數(shù)據(jù)拷貝到接收者的 goroutine 中,并喚醒發(fā)送方 goroutine;
b. 如果是有緩存的 channel(說明此時(shí)recvd滿了),則把 buf 中的 recvx 位置的數(shù)據(jù)拷貝到當(dāng)前接收的goroutine,然后把 sendq 中第一個(gè)等待發(fā)送goroutine的數(shù)據(jù)拷貝到buf 中的 sendx 位置,并喚醒發(fā)送的goroutine如果 channel 中 sendq 沒有等待發(fā)送數(shù)據(jù)的 goroutine:
a. 如果 buf 有數(shù)據(jù),則把 buf 中的 recvx 位置的數(shù)據(jù)拷貝到當(dāng)前的接收goroutine
b. 如果 buf 沒有數(shù)據(jù),則把當(dāng)前 goroutine 加入 recvd 等待隊(duì)列中,并掛起
5. channel 使用注意事項(xiàng)
最后啰嗦一下 channel 使用的注意事項(xiàng),這也是在我們平常開發(fā)中容易忽略的:
- 一個(gè) channel 不能多次 close,否則會(huì)導(dǎo)致 panic。
- 關(guān)閉一個(gè) nil 的 channel,會(huì)導(dǎo)致 panic。
- 向一個(gè)已經(jīng) close 的 channel 發(fā)送數(shù)據(jù),會(huì)導(dǎo)致 panic。
- 不要從一個(gè) receiver 測(cè)關(guān)閉 channel,也不要在有多個(gè) sender 時(shí)關(guān)閉 channel。在go語(yǔ)言中,對(duì)于一個(gè) channel,如果最終沒有任何 goroutine 引用它,不管 channel 有沒有被關(guān)閉,最終都會(huì)被 gc 回收。
- 如果監(jiān)聽的channel 已經(jīng)關(guān)閉,還可以獲取到 channel buf 中剩余的值,當(dāng)接收完 buf 中的數(shù)據(jù)后,才會(huì)獲取到零值。
到此這篇關(guān)于GoLang channel底層代碼分析詳解的文章就介紹到這了,更多相關(guān)GoLang channel內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
Go語(yǔ)言并發(fā)處理效率響應(yīng)能力及在現(xiàn)代軟件開發(fā)中的重要性
這篇文章主要為大家介紹了Go語(yǔ)言并發(fā)處理的效率及響應(yīng)能力以及在現(xiàn)代軟件開發(fā)中的重要性實(shí)例詳解,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪2023-12-12深入解析Go語(yǔ)言中HTTP請(qǐng)求處理的底層實(shí)現(xiàn)
本文將詳細(xì)介紹?Go?語(yǔ)言中?HTTP?請(qǐng)求處理的底層機(jī)制,包括工作流程、創(chuàng)建?Listen?Socket?監(jiān)聽端口、接收客戶端請(qǐng)求并建立連接以及處理客戶端請(qǐng)求并返回響應(yīng)等,需要的朋友可以參考下2023-05-05golang判斷chan channel是否關(guān)閉的方法
這篇文章主要介紹了golang判斷chan channel是否關(guān)閉的方法,結(jié)合實(shí)例形式對(duì)比分析了Go語(yǔ)言判斷chan沒有關(guān)閉的后果及關(guān)閉的方法,需要的朋友可以參考下2016-07-07golang中判斷請(qǐng)求是http還是https獲取當(dāng)前訪問地址
這篇文章主要為大家介紹了golang中判斷請(qǐng)求是http還是https獲取當(dāng)前訪問地址示例詳解,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪2023-10-10golang 函數(shù)以及函數(shù)和方法的詳解及區(qū)別
這篇文章主要介紹了golang 函數(shù)以及函數(shù)和方法的區(qū)別的相關(guān)資料,需要的朋友可以參考下2017-05-05