快捷導(dǎo)航

Go底層channel實(shí)現(xiàn)原理及示例詳解

更新時(shí)間：2022年08月08日 10:58:51 作者：阿甘與阿Q

這篇文章主要介紹了Go底層channel實(shí)現(xiàn)原理及示例詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

概念：

Go中的channel 是一個(gè)隊(duì)列，遵循先進(jìn)先出的原則，負(fù)責(zé)協(xié)程之間的通信（Go 語(yǔ)言提倡不要通過(guò)共享內(nèi)存來(lái)通信，而要通過(guò)通信來(lái)實(shí)現(xiàn)內(nèi)存共享，CSP(Communicating Sequential Process)并發(fā)模型，就是通過(guò) goroutine 和 channel 來(lái)實(shí)現(xiàn)的）

使用場(chǎng)景：

停止信號(hào)監(jiān)聽(tīng)

定時(shí)任務(wù)

生產(chǎn)方和消費(fèi)方解耦

控制并發(fā)數(shù)

底層數(shù)據(jù)結(jié)構(gòu)：

通過(guò)var聲明或者make函數(shù)創(chuàng)建的channel變量是一個(gè)存儲(chǔ)在函數(shù)棧幀上的指針，占用8個(gè)字節(jié)，指向堆上的hchan結(jié)構(gòu)體

源碼包中src/runtime/chan.go定義了hchan的數(shù)據(jù)結(jié)構(gòu)：

hchan結(jié)構(gòu)體：

type hchan struct {
 closed   uint32   // channel是否關(guān)閉的標(biāo)志
 elemtype *_type   // channel中的元素類型
 // channel分為無(wú)緩沖和有緩沖兩種。
 // 對(duì)于有緩沖的channel存儲(chǔ)數(shù)據(jù)，使用了 ring buffer（環(huán)形緩沖區(qū)) 來(lái)緩存寫入的數(shù)據(jù)，本質(zhì)是循環(huán)數(shù)組
 // 為啥是循環(huán)數(shù)組？普通數(shù)組不行嗎，普通數(shù)組容量固定更適合指定的空間，彈出元素時(shí)，普通數(shù)組需要全部都前移
 // 當(dāng)下標(biāo)超過(guò)數(shù)組容量后會(huì)回到第一個(gè)位置，所以需要有兩個(gè)字段記錄當(dāng)前讀和寫的下標(biāo)位置
 buf      unsafe.Pointer // 指向底層循環(huán)數(shù)組的指針（環(huán)形緩沖區(qū)）
 qcount   uint           // 循環(huán)數(shù)組中的元素?cái)?shù)量
 dataqsiz uint           // 循環(huán)數(shù)組的長(zhǎng)度
 elemsize uint16                 // 元素的大小
 sendx    uint           // 下一次寫下標(biāo)的位置
 recvx    uint           // 下一次讀下標(biāo)的位置
 // 嘗試讀取channel或向channel寫入數(shù)據(jù)而被阻塞的goroutine
 recvq    waitq  // 讀等待隊(duì)列
 sendq    waitq  // 寫等待隊(duì)列
 lock mutex //互斥鎖，保證讀寫channel時(shí)不存在并發(fā)競(jìng)爭(zhēng)問(wèn)題
}

等待隊(duì)列：

雙向鏈表，包含一個(gè)頭結(jié)點(diǎn)和一個(gè)尾結(jié)點(diǎn)

每個(gè)節(jié)點(diǎn)是一個(gè)sudog結(jié)構(gòu)體變量，記錄哪個(gè)協(xié)程在等待，等待的是哪個(gè)channel，等待發(fā)送/接收的數(shù)據(jù)在哪里

type waitq struct {
   first *sudog
   last  *sudog
}
type sudog struct {
    g *g
    next *sudog
    prev *sudog
    elem unsafe.Pointer 
    c        *hchan 
    ...
}

操作：

創(chuàng)建

使用 make(chan T, cap) 來(lái)創(chuàng)建 channel，make 語(yǔ)法會(huì)在編譯時(shí)，轉(zhuǎn)換為 makechan64 和 makechan

func makechan64(t *chantype, size int64) *hchan {
    if int64(int(size)) != size {
        panic(plainError("makechan: size out of range"))
    }
    return makechan(t, int(size))
}

創(chuàng)建channel 有兩種，一種是帶緩沖的channel，一種是不帶緩沖的channel

// 帶緩沖
ch := make(chan int, 3)
// 不帶緩沖
ch := make(chan int)

創(chuàng)建時(shí)會(huì)做一些檢查:

元素大小不能超過(guò) 64K
元素的對(duì)齊大小不能超過(guò) maxAlign 也就是 8 字節(jié)
計(jì)算出來(lái)的內(nèi)存是否超過(guò)限制

創(chuàng)建時(shí)的策略:

如果是無(wú)緩沖的 channel，會(huì)直接給 hchan 分配內(nèi)存
如果是有緩沖的 channel，并且元素不包含指針，那么會(huì)為 hchan 和底層數(shù)組分配一段連續(xù)的地址
如果是有緩沖的 channel，并且元素包含指針，那么會(huì)為 hchan 和底層數(shù)組分別分配地址

發(fā)送

發(fā)送操作，編譯時(shí)轉(zhuǎn)換為runtime.chansend函數(shù)

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool

阻塞式：

調(diào)用chansend函數(shù)，并且block=true

ch <- 10

非阻塞式：

調(diào)用chansend函數(shù)，并且block=false

select {
    case ch <- 10:
    ...
  default
}

向 channel 中發(fā)送數(shù)據(jù)時(shí)大概分為兩大塊：檢查和數(shù)據(jù)發(fā)送，數(shù)據(jù)發(fā)送流程如下：

如果 channel 的讀等待隊(duì)列存在接收者goroutine

將數(shù)據(jù)直接發(fā)送給第一個(gè)等待的 goroutine， 喚醒接收的 goroutine

如果 channel 的讀等待隊(duì)列不存在接收者goroutine

如果循環(huán)數(shù)組buf未滿，那么將會(huì)把數(shù)據(jù)發(fā)送到循環(huán)數(shù)組buf的隊(duì)尾
如果循環(huán)數(shù)組buf已滿，這個(gè)時(shí)候就會(huì)走阻塞發(fā)送的流程，將當(dāng)前 goroutine 加入寫等待隊(duì)列，并掛起等待喚醒

接收

發(fā)送操作，編譯時(shí)轉(zhuǎn)換為runtime.chanrecv函數(shù)

func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool)

阻塞式：

調(diào)用chanrecv函數(shù)，并且block=true

<ch
v := <ch
v, ok := <ch
// 當(dāng)channel關(guān)閉時(shí)，for循環(huán)會(huì)自動(dòng)退出，無(wú)需主動(dòng)監(jiān)測(cè)channel是否關(guān)閉，可以防止讀取已經(jīng)關(guān)閉的channel,造成讀到數(shù)據(jù)為通道所存儲(chǔ)的數(shù)據(jù)類型的零值
for i := range ch {
    fmt.Println(i)
}

非阻塞式：

調(diào)用chanrecv函數(shù)，并且block=false

select {
    case <-ch:
    ...
  default
}

向 channel 中接收數(shù)據(jù)時(shí)大概分為兩大塊，檢查和數(shù)據(jù)發(fā)送，而數(shù)據(jù)接收流程如下：

如果 channel 的寫等待隊(duì)列存在發(fā)送者goroutine

如果是無(wú)緩沖 channel，直接從第一個(gè)發(fā)送者goroutine那里把數(shù)據(jù)拷貝給接收變量，喚醒發(fā)送的 goroutine
如果是有緩沖 channel（已滿），將循環(huán)數(shù)組buf的隊(duì)首元素拷貝給接收變量，將第一個(gè)發(fā)送者goroutine的數(shù)據(jù)拷貝到 buf循環(huán)數(shù)組隊(duì)尾，喚醒發(fā)送的 goroutine

如果 channel 的寫等待隊(duì)列不存在發(fā)送者goroutine

如果循環(huán)數(shù)組buf非空，將循環(huán)數(shù)組buf的隊(duì)首元素拷貝給接收變量
如果循環(huán)數(shù)組buf為空，這個(gè)時(shí)候就會(huì)走阻塞接收的流程，將當(dāng)前 goroutine 加入讀等待隊(duì)列，并掛起等待喚醒

關(guān)閉

關(guān)閉操作，調(diào)用close函數(shù)，編譯時(shí)轉(zhuǎn)換為runtime.closechan函數(shù)

close(ch)

func closechan(c *hchan)

案例分析：

package main
import (
    "fmt"
    "time"
    "unsafe"
)
func main() {
  // ch是長(zhǎng)度為4的帶緩沖的channel
  // 初始hchan結(jié)構(gòu)體重的buf為空，sendx和recvx均為0
    ch := make(chan string, 4)
    fmt.Println(ch, unsafe.Sizeof(ch))
    go sendTask(ch)
    go receiveTask(ch)
    time.Sleep(1 * time.Second)
}
// G1是發(fā)送者
// 當(dāng)G1向ch里發(fā)送數(shù)據(jù)時(shí)，首先會(huì)對(duì)buf加鎖，然后將task存儲(chǔ)的數(shù)據(jù)copy到buf中，然后sendx++，然后釋放對(duì)buf的鎖
func sendTask(ch chan string) {
    taskList := []string{"this", "is", "a", "demo"}
    for _, task := range taskList {
        ch <- task //發(fā)送任務(wù)到channel
    }
}
// G2是接收者
// 當(dāng)G2消費(fèi)ch的時(shí)候，會(huì)首先對(duì)buf加鎖，然后將buf中的數(shù)據(jù)copy到task變量對(duì)應(yīng)的內(nèi)存里，然后recvx++,并釋放鎖
func receiveTask(ch chan string) {
    for {
        task := <-ch                  //接收任務(wù)
        fmt.Println("received", task) //處理任務(wù)
    }
}

總結(jié)hchan結(jié)構(gòu)體的主要組成部分有四個(gè)：