您好,登錄后才能下訂單哦!
這篇文章主要介紹“Golang Channel的底層結(jié)構(gòu)是什么”的相關(guān)知識(shí),小編通過(guò)實(shí)際案例向大家展示操作過(guò)程,操作方法簡(jiǎn)單快捷,實(shí)用性強(qiáng),希望這篇“Golang Channel的底層結(jié)構(gòu)是什么”文章能幫助大家解決問(wèn)題。
Golang 使用 Groutine 和 channels 實(shí)現(xiàn)了 CSP(Communicating Sequential Processes)
模型,channles
在 goroutine
的通信和同步中承擔(dān)著重要的角色。
在GopherCon 2017
中,Golang 專家 Kavya 深入介紹了 Go Channels 的內(nèi)部機(jī)制,以及運(yùn)行時(shí)調(diào)度器和內(nèi)存管理系統(tǒng)是如何支持 Channel 的,本文根據(jù) Kavya 的 ppt 學(xué)習(xí)和分析一下 go channels 的原理,希望能夠?qū)σ院笳_高效使用 golang 的并發(fā)帶來(lái)一些啟發(fā)。
以一個(gè)簡(jiǎn)單的 channel
應(yīng)用開(kāi)始,使用 goroutine
和 channel
實(shí)現(xiàn)一個(gè)任務(wù)隊(duì)列,并行處理多個(gè)任務(wù)。
func main(){ //帶緩沖的 channel ch := make(chan Task, 3) //啟動(dòng)固定數(shù)量的 worker for i := 0; i< numWorkers; i++ { go worker(ch) } //發(fā)送任務(wù)給 worker hellaTasks := getTaks() for _, task := range hellaTasks { ch <- task } ... } func worker(ch chan Task){ for { //接受任務(wù) task := <- ch process(task) } }
從上面的代碼可以看出,使用 golang
的 goroutine
和 channel
可以很容易的實(shí)現(xiàn)一個(gè)生產(chǎn)者-消費(fèi)者模式的任務(wù)隊(duì)列,相比 Java, c++簡(jiǎn)潔了很多。
channel 可以天然的實(shí)現(xiàn)了下面四個(gè)特性:
goroutine 安全
在不同的 goroutine 之間存儲(chǔ)和傳輸值 – 提供 FIFO 語(yǔ)義 (buffered channel 提供)
可以讓 goroutine block/unblock
那么 channel
是怎么實(shí)現(xiàn)這些特性的呢?下面我們看看當(dāng)我們調(diào)用 make 來(lái)生成一個(gè) channel 的時(shí)候都做了些什么。
上述任務(wù)隊(duì)列的例子第三行,使用 make 創(chuàng)建了一個(gè)長(zhǎng)度為 3 的帶緩沖的 channel,channel 在底層是一個(gè) hchan 結(jié)構(gòu)體,位于 src/runtime/chan.go
里。
其定義如下:
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 }
make 函數(shù)在創(chuàng)建 channel
的時(shí)候會(huì)在該進(jìn)程的 heap 區(qū)申請(qǐng)一塊內(nèi)存,創(chuàng)建一個(gè) hchan 結(jié)構(gòu)體,返回執(zhí)行該內(nèi)存的指針,所以獲取的的 ch 變量本身就是一個(gè)指針,在函數(shù)之間傳遞的時(shí)候是同一個(gè) channel。
hchan 結(jié)構(gòu)體使用一個(gè)環(huán)形隊(duì)列來(lái)保存 groutine
之間傳遞的數(shù)據(jù)(如果是緩存 channel 的話),使用**兩個(gè) list **保存像該 chan 發(fā)送和從該 chan 接收數(shù)據(jù)的 goroutine
,還有一個(gè) mutex 來(lái)保證操作這些結(jié)構(gòu)的安全。
向 channel
發(fā)送和從 channel 接收數(shù)據(jù)主要涉及 hchan 里的四個(gè)成員變量,借用 Kavya ppt 里的圖示,來(lái)分析發(fā)送和接收的過(guò)程。
還是以前面的任務(wù)隊(duì)列為例:
//G1 func main(){ ... for _, task := range hellaTasks { ch <- task //sender } ... } //G2 func worker(ch chan Task){ for { //接受任務(wù) task := <- ch //recevier process(task) } }
其中 G1 是發(fā)送者,G2 是接收,因?yàn)?ch 是長(zhǎng)度為 3 的帶緩沖 channel,初始的時(shí)候 hchan 結(jié)構(gòu)體的 buf 為空,sendx 和 recvx 都為 0,當(dāng) G1 向 ch 里發(fā)送數(shù)據(jù)的時(shí)候,會(huì)首先對(duì) buf 加鎖,然后將要發(fā)送的數(shù)據(jù) copy 到 buf 里,并增加 sendx 的值,最后釋放 buf 的鎖。然后 G2 消費(fèi)的時(shí)候首先對(duì) buf 加鎖,然后將 buf 里的數(shù)據(jù) copy 到 task 變量對(duì)應(yīng)的內(nèi)存里,增加 recvx,最后釋放鎖。整個(gè)過(guò)程,G1 和 G2 沒(méi)有共享的內(nèi)存,底層通過(guò) hchan 結(jié)構(gòu)體的 buf,使用 copy 內(nèi)存的方式進(jìn)行通信,最后達(dá)到了共享內(nèi)存的目的,這完全符合 CSP 的設(shè)計(jì)理念
Do not comminute by sharing memory;instead, share memory by communicating
一般情況下,G2 的消費(fèi)速度應(yīng)該是慢于 G1 的,所以 buf 的數(shù)據(jù)會(huì)越來(lái)越多,這個(gè)時(shí)候 G1 再向 ch 里發(fā)送數(shù)據(jù),這個(gè)時(shí)候 G1 就會(huì)阻塞,那么阻塞到底是發(fā)生了什么呢?
goroutine 是 Golang 實(shí)現(xiàn)的用戶空間的輕量級(jí)的線程,有 runtime 調(diào)度器調(diào)度,與操作系統(tǒng)的 thread 有多對(duì)一的關(guān)系,相關(guān)的數(shù)據(jù)結(jié)構(gòu)如下圖:
其中 M 是操作系統(tǒng)的線程,G 是用戶啟動(dòng)的 goroutine,P 是與調(diào)度相關(guān)的 context,每個(gè) M 都擁有一個(gè) P,P 維護(hù)了一個(gè)能夠運(yùn)行的 goutine 隊(duì)列,用于該線程執(zhí)行。
當(dāng) G1 向 buf 已經(jīng)滿了的 ch 發(fā)送數(shù)據(jù)的時(shí)候,當(dāng) runtine 檢測(cè)到對(duì)應(yīng)的 hchan 的 buf 已經(jīng)滿了,會(huì)通知調(diào)度器,調(diào)度器會(huì)將 G1 的狀態(tài)設(shè)置為 waiting, 移除與線程 M 的聯(lián)系,然后從 P 的 runqueue 中選擇一個(gè) goroutine 在線程 M 中執(zhí)行,此時(shí) G1 就是阻塞狀態(tài),但是不是操作系統(tǒng)的線程阻塞,所以這個(gè)時(shí)候只用消耗少量的資源。
那么 G1 設(shè)置為 waiting 狀態(tài)后去哪了?怎們?nèi)?resume 呢?我們?cè)倩氐?hchan 結(jié)構(gòu)體,注意到 hchan 有個(gè) sendq 的成員,其類型是 waitq,查看源碼如下:
type hchan struct { ... recvq waitq // list of recv waiters sendq waitq // list of send waiters ... } // type waitq struct { first *sudog last *sudog }
實(shí)際上,當(dāng) G1 變?yōu)?waiting
狀態(tài)后,會(huì)創(chuàng)建一個(gè)代表自己的 sudog 的結(jié)構(gòu),然后放到 sendq 這個(gè) list 中,sudog 結(jié)構(gòu)中保存了 channel 相關(guān)的變量的指針(如果該 Goroutine 是 sender,那么保存的是待發(fā)送數(shù)據(jù)的變量的地址,如果是 receiver 則為接收數(shù)據(jù)的變量的地址,之所以是地址,前面我們提到在傳輸數(shù)據(jù)的時(shí)候使用的是 copy 的方式)
當(dāng) G2 從 ch 中接收一個(gè)數(shù)據(jù)時(shí),會(huì)通知調(diào)度器,設(shè)置 G1 的狀態(tài)為 runnable
,然后將加入 P 的 runqueue 里,等待線程執(zhí)行。
前面我們是假設(shè) G1 先運(yùn)行,如果 G2 先運(yùn)行會(huì)怎么樣呢?如果 G2 先運(yùn)行,那么 G2 會(huì)從一個(gè) empty 的 channel 里取數(shù)據(jù),這個(gè)時(shí)候 G2 就會(huì)阻塞,和前面介紹的 G1 阻塞一樣,G2 也會(huì)創(chuàng)建一個(gè) sudog 結(jié)構(gòu)體,保存接收數(shù)據(jù)的變量的地址,但是該 sudog 結(jié)構(gòu)體是放到了 recvq 列表里,當(dāng) G1 向 ch 發(fā)送數(shù)據(jù)的時(shí)候,runtime
并沒(méi)有對(duì) hchan 結(jié)構(gòu)體題的 buf 進(jìn)行加鎖,而是直接將 G1 里的發(fā)送到 ch 的數(shù)據(jù) copy 到了 G2 sudog
里對(duì)應(yīng)的 elem 指向的內(nèi)存地址!
關(guān)于“Golang Channel的底層結(jié)構(gòu)是什么”的內(nèi)容就介紹到這里了,感謝大家的閱讀。如果想了解更多行業(yè)相關(guān)的知識(shí),可以關(guān)注億速云行業(yè)資訊頻道,小編每天都會(huì)為大家更新不同的知識(shí)點(diǎn)。
免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng),如果涉及侵權(quán)請(qǐng)聯(lián)系站長(zhǎng)郵箱:is@yisu.com進(jìn)行舉報(bào),并提供相關(guān)證據(jù),一經(jīng)查實(shí),將立刻刪除涉嫌侵權(quán)內(nèi)容。