詳解Go channel管道的運行原理
更新時間：2023年05月06日 11:35:00 作者：unitiny
Go推薦通過通信來共享內存，而channel就實現了這一理念。那channel是怎么運行的呢？本文將帶你搞懂Go channel管道的運行原理，感興趣的同學可以參考一下
前言
Go推薦通過通信來共享內存，而channel就實現了這一理念。那channel是怎么運行的呢？
功能
舉個例子看下channel的使用效果：
package main
import (
"fmt"
"math/rand"
"time"
)
func write(c chan int) {
for {
num := rand.Intn(100)
c <- num
}
}
func read(c chan int) {
for {
num := <-c
fmt.Println("讀取管道的隨機數：", num)
time.Sleep(time.Second)
}
}
func main() {
var c = make(chan int, 8)
go read(c)
for i := 0; i < 5; i++ {
go write(c)
}
time.Sleep(time.Minute)
}以上代碼新建了一個緩沖區(qū)為8的管道，然后開啟read和五個write讀寫協程。寫協程寫入一個隨機數，讀協程每隔一秒讀取并打印，效果如下：
說明協程間可以通過管道來互相通信。接著了解下channel的結構。
channel結構
channel結構體位于GOROOT/src/runtime/chan.go下的hchan，源碼如下：
type hchan struct {
qcount uint // 隊列中元素總數
dataqsiz uint // 環(huán)型隊列大小
buf unsafe.Pointer // 指向dataqsize的數組（即緩沖區(qū)）
elemsize uint16
closed uint32
elemtype *_type // 元素類型
sendx uint // 發(fā)送到緩沖區(qū)的位置索引
recvx uint // 接收到緩沖區(qū)的位置索引
recvq waitq // 接收者隊列
sendq waitq // 發(fā)送者隊列
lock mutex // 鎖，用于保護channel數據
}其中發(fā)送者和接收者隊列是一個waitq類型，具體如下：
type waitq struct {
first *sudog
last *sudog
}waitq里有隊頭first，隊尾last的指針，指向sudog結構體。
也就是說，waitq是一個列表隊列，隊列里每個元素都是一個sudog結構體，sudog中包裝著一個協程。
解析一個hchan各部分結構：
- 頭部
type hchan struct {
qcount uint // 隊列中元素總數
dataqsiz uint // 環(huán)型隊列大小
buf unsafe.Pointer // 指向dataqsize的數組（即緩沖區(qū)）
elemsize uint16
closed uint32
elemtype *_type // 元素類型
...
}這部分表示一個環(huán)型緩沖區(qū)。圖解如下：
- 尾部
type hchan struct {
...
sendx uint // 發(fā)送到緩沖區(qū)的位置索引
recvx uint // 接收到緩沖區(qū)的位置索引
recvq waitq // 接收者隊列
sendq waitq // 發(fā)送者隊列
...
}這部分把協程分為兩個身份，使用chan <- 語法的協程為發(fā)送者，使用<- chan 語法的協程為接收者，并放到各自隊列中。圖解如下：
結合示例代碼。運行結構如下：
由于寫協程一直寫，讀協程每隔一秒才讀一次，因此很快將緩沖區(qū)寫滿了，這時：
- 寫協程被裝入sudog進行休眠等待
- 讀協程每隔一秒從緩沖區(qū)讀取數據
運行原理
使用chan <- 為發(fā)送者，對發(fā)送者來說：
- 先查看是否有接收者，有則優(yōu)先喚醒并拷貝數據給接收者，然后結束
- 無接收者再查看緩沖區(qū)，數據未滿則將數據放入緩沖區(qū)，然后結束
- 緩沖區(qū)也滿了，則封裝成sudog，休眠等待
使用<- chan 為接收者，對接收者來說：
- 優(yōu)先接收緩沖區(qū)的值
- 再接收發(fā)送者的值
- 否則休眠等待
思考下：
有休眠的接收者，且緩沖區(qū)數據已滿的情況是否存在？為什么？
有休眠的發(fā)送者，且緩沖區(qū)為空的情況是否存在？為什么？
以上答案：
有休眠的接收者，緩沖區(qū)不會出現數據已滿情況。因為接收者要休眠，得緩沖區(qū)沒數據才行。
有休眠的發(fā)送者，緩沖區(qū)不會出現為空情況。因為發(fā)送者要休眠，得緩沖區(qū)數據已滿才行。
源碼分析
使用chan <-后，會調用GOROOT\src\runtime\chan.go下的chansend1方法
func chansend1(c *hchan, elem unsafe.Pointer) {
chansend(c, elem, true, getcallerpc())
}然后調用chansend方法
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 {
send(c, sg, ep, func() { unlock(&c.lock) }, 3)
return true
}
// 判斷緩沖區(qū)是否還有空余
if c.qcount < c.dataqsiz {
// Space is available in the channel buffer. Enqueue the element to send.
qp := chanbuf(c, c.sendx) // 有的話獲得緩沖區(qū)要存放數據的地址
if raceenabled {
racenotify(c, c.sendx, nil)
}
typedmemmove(c.elemtype, qp, ep) // 將數據拷貝到緩沖區(qū)擴容地址qp上
c.sendx++
if c.sendx == c.dataqsiz {
c.sendx = 0
}
c.qcount++
unlock(&c.lock)
return true
}
...
// 否則封裝成sodug休眠自己，加入發(fā)送者等待隊列
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.
gp.parkingOnChan.Store(true)
// 主動掛起
gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanSend, traceEvGoBlockSend, 2)
...
// 被喚醒后釋放sudog
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) // 釋放sudog
if closed {
if c.closed == 0 {
throw("chansend: spurious wakeup")
}
panic(plainError("send on closed channel"))
}
return true
}使用<- chan后，會調用GOROOT\src\runtime\chan.go下的chanrecv1方法
func chanrecv1(c *hchan, elem unsafe.Pointer) {
chanrecv(c, elem, true)
}然后調用chanrecv方法
func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
...
lock(&c.lock)
if c.closed != 0 {
if c.qcount == 0 {
if raceenabled {
raceacquire(c.raceaddr())
}
unlock(&c.lock)
if ep != nil {
typedmemclr(c.elemtype, ep)
}
return true, false
}
// The channel has been closed, but the channel's buffer have data.
} else {
// 如果有發(fā)送者在休眠，則調用recv
if sg := c.sendq.dequeue(); sg != nil {
recv(c, sg, ep, func() { unlock(&c.lock) }, 3)
return true, true
}
}
// 無發(fā)送者，但緩沖區(qū)有數據
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
}
if !block {
unlock(&c.lock)
return false, false
}
// 休眠自己
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) // 封裝成sudog入隊
gp.parkingOnChan.Store(true)
// 主動掛起
gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanReceive, traceEvGoBlockRecv, 2)
// 被喚醒后釋放sudog
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
}當有發(fā)送者，會調用recv
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 {
// copy data from sender
recvDirect(c.elemtype, sg, ep)
}
} else {
// 獲取緩沖區(qū)數據的位置
qp := chanbuf(c, c.recvx)
if raceenabled {
racenotify(c, c.recvx, nil)
racenotify(c, c.recvx, sg)
}
// copy data from queue to receiver
if ep != nil {
// 將緩沖區(qū)數據拷貝到
typedmemmove(c.elemtype, ep, qp)
}
// 將發(fā)送者的數據拷貝到緩沖區(qū)
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) // 喚醒發(fā)送者協程
}因此，接收者還是先接收緩沖區(qū)數據，再接收發(fā)送者的數據。其實就是按隊列的先進先出順序。
總結
留下兩個問題：
發(fā)送者分別遇到無有休眠接收協程，有休眠接收協程，無接收協程且緩沖區(qū)沒滿，緩沖區(qū)滿了四種情況該如何處理？
接收者分別遇到無休眠發(fā)送協程且緩沖區(qū)為空，無發(fā)送協程且緩沖區(qū)有數據，有休眠發(fā)送協程且緩沖區(qū)已滿，緩沖區(qū)滿了四種情況該如何處理？
以上就是詳解Go channel管道的運行原理的詳細內容，更多關于Go channel的資料請關注腳本之家其它相關文章！
相關文章
這篇文章主要介紹了Golang使用ini庫讀取配置詳情，go-ini是一個非常方便、高效的go配置文件操作庫。使用它在項目中讀取和修改配置文件，下文相關資料需要的小伙伴可可以參考一下2022-04-04
這篇文章主要介紹了Go語言接口用法,實例分析了Go語言接口的功能、定義及使用技巧,需要的朋友可以參考下2015-02-02
這篇文章主要為大家詳細介紹了Go調用opencv實現圖片矯正的代碼示例,文中的示例代碼簡潔易懂,感興趣的小伙伴可以跟隨小編一起學習一下2023-09-09
這篇文章主要為大家介紹了Go程序員踩過的defer坑錯誤處理，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪2022-06-06
這篇文章主要為大家介紹了Go語言IO輸入輸出底層原理及文件操作API示例詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪2023-01-01
goioc?是一個基于?GO?語言編寫的依賴注入框架，基于反射來進行編寫。本文主要為大家介紹了goioc框架的原理與使用，需要的可以參考一下2022-11-11
本文主要介紹了Go語言利用Unmarshal解析json字符串的實現，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2022-05-05
本文主要介紹了Golang實現HTTP編程請求和響應，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2022-08-08
本文詳細介紹了Go語言中的strings庫的使用方法,包括字符串的查找、替換、分割、比較、大小寫轉換等操作,strings庫是Go語言中非常重要且功能豐富的標準庫,幾乎涵蓋了所有字符串處理的需求2024-09-09
golang多次讀取http request body的問題分析
這篇文章主要給大家分析了golang多次讀取http request body的問題,文中通過代碼示例和圖文介紹的非常詳細,對大家的學習或工作有一定的幫助,需要的朋友可以參考下2024-01-01