Go channel發(fā)送方和接收方如何相互阻塞等待源碼解讀
并發(fā)編程的可見性
在 Go 官網(wǎng)上的內(nèi)存模型一文中，介紹了在 Go 并發(fā)編程下數(shù)據(jù)可見性問題，可見性是并發(fā)編程中一個重要概念，指的是在哪些條件下，可以保證一個線程中讀取某個變量時，可以觀察到另一個線程對該變量的寫入后的值，Go 語言中的 goroutine 也適用。
一般來說可見性屬于偏硬件和底層，因為涉及到多核 CPU 的 cache 讀寫和同步問題，開發(fā)者不需要關心細節(jié)，高級編程語言要么屏蔽掉了這些細節(jié)，要么會給出一些保證，承諾在確定的條件下就會得到確定的結(jié)果。
Go channel 有一個特性是在一個無緩沖的 channel 上發(fā)送和接收必須等待對方準備好，才可以執(zhí)行，否則會被阻塞。實際上這就是一個同步保證，那么這個同步保證是如何實現(xiàn)的？下面看看官方文章中是如何解釋的。
先 send 后 receive
文中對 channel 的描述有幾個原則，第一個是
A send on a channel is synchronized before the completion of the corresponding receive from that channel.
意思是：在一個 channel 上的發(fā)送操作應該發(fā)生在對應的接收操作完成之前。說人話就是：要先發(fā)送數(shù)據(jù)，然后才能接收數(shù)據(jù)，否則就會阻塞。這也比較符合一般的認知。
并用下面一段代碼舉例說明，這段代碼確保一定會輸出 "hello, world”。
var c = make(chan int, 10)
var a string
func f() {
a = "hello, world"
c <- 0
}
func main() {
go f()
<-c
print(a)
}f 函數(shù)負責給變量 a 賦值，main 函數(shù)負責打印變量 a。main 函數(shù)阻塞等待在 <- c 處，直到 f 函數(shù)對 a 賦值之后并寫入數(shù)據(jù)到 c 中，main 函數(shù)才被喚醒繼續(xù)執(zhí)行，所以此時打印 a 必然會得到結(jié)果。
先 receive 后 send？
而下面這段描述有點反直覺
A receive from an unbuffered channel is synchronized before the completion of the corresponding send on that channel.
意思是在無緩沖 channel 上的接收操作發(fā)生在對應的發(fā)送操作完成之前，說人話就是：要先接收數(shù)據(jù)，之后才可以發(fā)送數(shù)據(jù)，否則就會阻塞。這句話看上去與第一條相悖，因為第一條強調(diào)發(fā)送操作要在接收完成之前發(fā)生，而這一條強調(diào)接收操作要在發(fā)送完成之前發(fā)生，這樣相互等待對方的情況，不會陷入死鎖狀態(tài)嗎？
下面的示例代碼與前一個類似，區(qū)別是將 c 換成了無緩沖 channel，并把 c 的寫入和讀取調(diào)換了位置，這段代碼同樣可以保證輸出 "hello, world”。
var c = make(chan int)
var a string
func f() {
a = "hello, world"
<-c
}
func main() {
go f()
c <- 0
print(a)
}這兩段話到底是什么意思？為什么要相互等待但又不會死鎖？
接下來看看 runtime/chan.go 中是怎么實現(xiàn) channel 的發(fā)送和接收的。
channel 的結(jié)構(gòu)
首先看看 channel 的數(shù)據(jù)結(jié)構(gòu)
type hchan struct {
qcount uint // 緩沖區(qū)元素數(shù)量
dataqsiz uint // 緩沖區(qū)大小
buf unsafe.Pointer // 緩沖區(qū)起始指針
elemsize uint16
closed uint32
elemtype *_type
sendx uint // 下一次發(fā)送的元素在隊列中的索引
recvx uint // 下一個接收的元素在隊列中的索引
recvq waitq // 當隊列無數(shù)據(jù)時，receiver 阻塞等待的隊列
sendq waitq // 當隊列無空間時，sender 阻塞等待的隊列
lock mutex
}channel 內(nèi)部實現(xiàn)了一個環(huán)形隊列，通過 qcount dataqsiz buf sendx recvx 幾個部分組成。
另外 channel 還維護了兩個等待隊列，如果在執(zhí)行 <-c receive 操作時，此時 channel 不滿足接收條件，receiver 會進入 recvq 等待隊列；同樣的如果執(zhí)行 c<- send 操作時，此時 channel 不滿足發(fā)送條件，sender 會進入 sendq 等待隊列。
具體看代碼：
var c = make(chan int)
var a string
func f() {
a = "hello, world"
x := <-c // 3
fmt.Println("\nx:", x)
}
func main() {
go f() // 1
c <- 123456 // 2
print(a)
}send 具體干了什么
當 main 函數(shù)執(zhí)行到 c<-123456 是，會執(zhí)行 runtime/chan.go 中的 chansend 函數(shù)，該函數(shù)首先會判斷當前 channel c 的等待接收隊列是否有阻塞的 receiver
func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
// ...省略部分代碼...
// 是否有等待的 receiver 存在
if sg := c.recvq.dequeue(); sg != nil {
send(c, sg, ep, func() { unlock(&c.lock) }, 3)
return true
}
// ...省略部分代碼...
}如果有等待的 receiver 則彈出隊列，調(diào)用 send 函數(shù)，其中 sg 就表示 receiver，sg.elem 表示將數(shù)據(jù)接收到哪里去，這個地址也就對應示例代碼中的變量 x 的地址。
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
}
// ...省略部分代碼...
// 將 goroutine 置為可執(zhí)行狀態(tài)
}sendDirect 函數(shù)就是直接從 src 里面將數(shù)據(jù)復制到 dst 中。
// 直接拷貝數(shù)據(jù)
func sendDirect(t *_type, sg *sudog, src unsafe.Pointer) {
dst := sg.elem
typeBitsBulkBarrier(t, uintptr(dst), uintptr(src), t.size)
memmove(dst, src, t.size)
}回到 chansend 函數(shù)，如果沒有等待的 receiver，那么會查看當前 buf 中是否有空間，如果有空間，則數(shù)據(jù)緩存到 buf 中。
func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
// ...省略部分代碼...
// 將數(shù)據(jù)緩存到 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++
if c.sendx == c.dataqsiz {
c.sendx = 0
}
c.qcount++
unlock(&c.lock)
return true
}
// ...省略部分代碼...
}如果也沒有 buf 空間，那么就將 sender 本身放入到 sendq 等待隊列中。
func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
// ...省略部分代碼...
// 進入 sendq 等待隊列
gp := getg()
mysg := acquireSudog()
mysg.releasetime = 0
if t0 != 0 {
mysg.releasetime = -1
}
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)
// ...省略部分代碼...
}總結(jié)起來 send 操作分三部分：
- 如果當前 channel 上有等待的 receiver，則直接 copy 數(shù)據(jù)過去
- 否則如果當前 buf 有空閑空間，則將數(shù)據(jù)存在 buf 中
- 否則將 sender 本身加入到 sendq 等待隊列中
receive 具體干了什么
相應的與發(fā)送類似，執(zhí)行到示例代碼中第 (3) 步接收數(shù)據(jù)時，會調(diào)用 runtime/chan.go 中的 chanrecv 函數(shù)來處理接收，同樣是先看 sender 等待隊列是否有阻塞的 sender
func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
// ...省略部分代碼...
// 從等待的 sender 取一個出來
if sg := c.sendq.dequeue(); sg != nil {
recv(c, sg, ep, func() { unlock(&c.lock) }, 3)
return true, true
}
// ...省略部分代碼...
}如果有的等待的 sender，那么將 sender 取出來，并復制數(shù)據(jù)。
func recv(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {
// ...省略部分代碼...
if ep != nil {
// copy data from sender
recvDirect(c.elemtype, sg, ep)
}
// ...省略部分代碼...
}
func recvDirect(t *_type, sg *sudog, dst unsafe.Pointer) {
src := sg.elem
typeBitsBulkBarrier(t, uintptr(dst), uintptr(src), t.size)
memmove(dst, src, t.size)
}如果沒有等待的 sender，那么看 buf 中有沒有緩存的數(shù)據(jù)
func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
// ...省略部分代碼...
if c.qcount > 0 {
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
}
// ...省略部分代碼...
}最后如果也沒有 buf 數(shù)據(jù)，那么久把自己加入到 receiver 等待隊列中 recvq
func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
// ...省略部分代碼...
gp := getg()
mysg := acquireSudog()
mysg.releasetime = 0
if t0 != 0 {
mysg.releasetime = -1
}
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)
// ...省略部分代碼...
}總結(jié)起來 receive 操作分三部分：
- 如果當前 channel 上有等待的 sender，則直接 copy 數(shù)據(jù)過去
- 否則如果當前 buf 有緩存的數(shù)據(jù)，則將讀取該數(shù)據(jù)
- 否則將 receiver 本身加入到 recvq 等待隊列中
小結(jié)
這樣一來就能夠理解前面的兩個原則了，在一個無緩沖的 channel 中，無論是 sender 先執(zhí)行，還是 receiver 先執(zhí)行，都會因為找不到對方，并且沒有 buf 空間的情況下，將自己加入到等待隊列；當對方開始執(zhí)行時就會檢查到已經(jīng)有對端正在阻塞，進而拷貝數(shù)據(jù)，并喚醒阻塞的對象最終走完整個流程。
有一種說法是：sender 必須在 receiver 準備好才能執(zhí)行，否則就會阻塞；而 receiver 必須在 sender 準備好才能執(zhí)行，否則就會阻塞；這個說法沒錯，但是太籠統(tǒng)了，什么叫準備好？怎么算是準備好？這是比較模糊的。而看過 send 和 receive 的流程之后，就更能理解整個過程了。
為什么要有無緩沖 channel
實際上兩個 goroutine 相互等待對方到達某個狀態(tài)的效果，非常類似操作系統(tǒng)中的一種同步機制：屏障 barrier，同步屏障要求只有當所有進程都到達屏障后，才能一起執(zhí)行下一狀態(tài)，否則就阻塞在屏障處。
回到 channel 操作，即 sender 和 receiver 無論誰先執(zhí)行，都必須等待對方也已經(jīng)執(zhí)行，兩者才可以繼續(xù)執(zhí)行。就像一塊電路板串聯(lián)有兩個開關，要想電路聯(lián)通，必須兩個開關都被打開才可以，而不管哪一個先打開，都必須等待另一個開關也打開，之后電流才可以接通電路也才聯(lián)通。
可以將無緩沖 channel 看做是一種同步屏障，同步屏障能夠讓多個 goroutine 都達到某種狀態(tài)之后才可以繼續(xù)執(zhí)行，這是帶緩沖 channel 無法做到的。另外在無緩沖 channel 數(shù)據(jù)的交換更加簡單快速，因為不需要維護緩存 buf，實現(xiàn)邏輯也更簡單，運行更可靠。
以上就是Go channel發(fā)送方和接收方如何相互阻塞等待源碼解讀的詳細內(nèi)容，更多關于Go channel相互阻塞等待的資料請關注腳本之家其它相關文章！
相關文章
這篇文章主要為大家介紹了go基礎語法50問及方法詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪2023-01-01
這篇文章主要介紹了golang post請求常用的幾種方式小結(jié)，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2021-04-04
strconv包提供了字符串和基本數(shù)據(jù)類型之間的相互轉(zhuǎn)換功能，本文將帶大家深入了解Go語言標準庫中的strconv包，掌握其常用的函數(shù)和用法，希望對大家有所幫助2023-06-06
Go 數(shù)據(jù)庫查詢與結(jié)構(gòu)體映射的示例詳解
本文主要介紹了如何使用Go語言進行數(shù)據(jù)庫查詢并將查詢結(jié)果映射到結(jié)構(gòu)體中,文章詳細講解了結(jié)構(gòu)體字段導出和db標簽的使用方法,并通過示例代碼展示了如何正確地進行數(shù)據(jù)庫查詢和結(jié)果映射2024-11-11
這篇文章主要介紹了Golang 端口復用測試的實現(xiàn)，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2021-03-03
數(shù)列趨勢圖是用來表示數(shù)列中各項之間的變化趨勢的圖形,它可以幫助我們觀察和分析數(shù)列的規(guī)律和特點,一般來說,數(shù)列趨勢圖可以分為兩種類型：折線圖和散點圖,本文給大家介紹了Golang繪制數(shù)列趨勢圖的操作步驟,文中有詳細的代碼示例,需要的朋友可以參考下2024-04-04
本文介紹了如何在Go語言項目中導入同級目錄和不同目錄的包,詳細解釋了創(chuàng)建文件結(jié)構(gòu)、編寫主函數(shù)、同級目錄和不同目錄方法的調(diào)用,適合初學者參考,幫助理解Go項目的基本構(gòu)建和包管理2024-09-09
golang高并發(fā)系統(tǒng)限流策略漏桶和令牌桶算法源碼剖析
這篇文章主要介紹了golang高并發(fā)系統(tǒng)限流策略漏桶和令牌桶算法源碼剖析，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪2022-06-06
這篇文章主要給大家介紹了關于Golang讀寫Excel的方法教程，golang操作excel利用的是Excelize，Excelize是Golang編寫的一個用來操作 Office Excel 文檔類庫，基于微軟的Office OpenXML標準，文中介紹的非常詳細，需要的朋友可以參考借鑒，下面來一起看看吧。2017-05-05
Go語言web快速開發(fā)框架Gin的HttpRouter路由的使用
in框架內(nèi)部使用了高性能的路由器庫httprouter,支持動態(tài)參數(shù)匹配和簡潔的接口,本文主要介紹了Go語言web快速開發(fā)框架Gin的HttpRouter路由的使用,感興趣的可以了解一下2025-03-03