Go底層channel實現原理及範例詳解

2022-08-08 14:01:25

概念：

Go中的channel 是一個佇列，遵循先進先出的原則，負責協程之間的通訊（Go 語言提倡不要通過共用記憶體來通訊，而要通過通訊來實現記憶體共用，CSP(Communicating Sequential Process)並行模型，就是通過 goroutine 和 channel 來實現的）

使用場景：

停止訊號監聽

定時任務

生產方和消費方解耦

控制並行數

底層資料結構：

通過var宣告或者make函數建立的channel變數是一個儲存在函數棧幀上的指標，佔用8個位元組，指向堆上的hchan結構體

原始碼包中src/runtime/chan.go定義了hchan的資料結構：

hchan結構體：

type hchan struct {
 closed   uint32   // channel是否關閉的標誌
 elemtype *_type   // channel中的元素型別
 // channel分為無緩衝和有緩衝兩種。
 // 對於有緩衝的channel儲存資料，使用了 ring buffer（環形緩衝區) 來快取寫入的資料，本質是迴圈陣列
 // 為啥是迴圈陣列？普通陣列不行嗎，普通陣列容量固定更適合指定的空間，彈出元素時，普通陣列需要全部都前移
 // 當下標超過陣列容量後會回到第一個位置，所以需要有兩個欄位記錄當前讀和寫的下標位置
 buf      unsafe.Pointer // 指向底層迴圈陣列的指標（環形緩衝區）
 qcount   uint           // 迴圈陣列中的元素數量
 dataqsiz uint           // 迴圈陣列的長度
 elemsize uint16                 // 元素的大小
 sendx    uint           // 下一次寫下標的位置
 recvx    uint           // 下一次讀下標的位置
 // 嘗試讀取channel或向channel寫入資料而被阻塞的goroutine
 recvq    waitq  // 讀等待佇列
 sendq    waitq  // 寫等待佇列
 lock mutex //互斥鎖，保證讀寫channel時不存在並行競爭問題
}

等待佇列：

雙向連結串列，包含一個頭結點和一個尾結點

每個節點是一個sudog結構體變數，記錄哪個協程在等待，等待的是哪個channel，等待傳送/接收的資料在哪裡

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

操作：

建立

使用 make(chan T, cap) 來建立 channel，make 語法會在編譯時，轉換為 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))
}

建立channel 有兩種，一種是帶緩衝的channel，一種是不帶緩衝的channel

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

建立時會做一些檢查:

元素大小不能超過 64K
元素的對齊大小不能超過 maxAlign 也就是 8 位元組
計算出來的記憶體是否超過限制

建立時的策略:

如果是無緩衝的 channel，會直接給 hchan 分配記憶體
如果是有緩衝的 channel，並且元素不包含指標，那麼會為 hchan 和底層陣列分配一段連續的地址
如果是有緩衝的 channel，並且元素包含指標，那麼會為 hchan 和底層陣列分別分配地址

傳送

傳送操作，編譯時轉換為runtime.chansend函數

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

阻塞式：

呼叫chansend函數，並且block=true

ch <- 10

非阻塞式：

呼叫chansend函數，並且block=false

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

向 channel 中傳送資料時大概分為兩大塊：檢查和資料傳送，資料傳送流程如下：

如果 channel 的讀等待佇列存在接收者goroutine

將資料直接傳送給第一個等待的 goroutine， 喚醒接收的 goroutine

如果 channel 的讀等待佇列不存在接收者goroutine

如果迴圈陣列buf未滿，那麼將會把資料傳送到迴圈陣列buf的隊尾
如果迴圈陣列buf已滿，這個時候就會走阻塞傳送的流程，將當前 goroutine 加入寫等待佇列，並掛起等待喚醒

接收

傳送操作，編譯時轉換為runtime.chanrecv函數

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

阻塞式：

呼叫chanrecv函數，並且block=true

<ch
v := <ch
v, ok := <ch
// 當channel關閉時，for迴圈會自動退出，無需主動監測channel是否關閉，可以防止讀取已經關閉的channel,造成讀到資料為通道所儲存的資料型別的零值
for i := range ch {
    fmt.Println(i)
}

非阻塞式：

呼叫chanrecv函數，並且block=false

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

向 channel 中接收資料時大概分為兩大塊，檢查和資料傳送，而資料接收流程如下：

如果 channel 的寫等待佇列存在傳送者goroutine

如果是無緩衝 channel，直接從第一個傳送者goroutine那裡把資料拷貝給接收變數，喚醒傳送的 goroutine
如果是有緩衝 channel（已滿），將回圈陣列buf的隊首元素拷貝給接收變數，將第一個傳送者goroutine的資料拷貝到 buf迴圈陣列隊尾，喚醒傳送的 goroutine

如果 channel 的寫等待佇列不存在傳送者goroutine

如果迴圈陣列buf非空，將回圈陣列buf的隊首元素拷貝給接收變數
如果迴圈陣列buf為空，這個時候就會走阻塞接收的流程，將當前 goroutine 加入讀等待佇列，並掛起等待喚醒

關閉

關閉操作，呼叫close函數，編譯時轉換為runtime.closechan函數

close(ch)

func closechan(c *hchan)

案例分析：

package main
import (
    "fmt"
    "time"
    "unsafe"
)
func main() {
  // ch是長度為4的帶緩衝的channel
  // 初始hchan結構體重的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是傳送者
// 當G1向ch裡傳送資料時，首先會對buf加鎖，然後將task儲存的資料copy到buf中，然後sendx++，然後釋放對buf的鎖
func sendTask(ch chan string) {
    taskList := []string{"this", "is", "a", "demo"}
    for _, task := range taskList {
        ch <- task //傳送任務到channel
    }
}
// G2是接收者
// 當G2消費ch的時候，會首先對buf加鎖，然後將buf中的資料copy到task變數對應的記憶體裡，然後recvx++,並釋放鎖
func receiveTask(ch chan string) {
    for {
        task := <-ch                  //接收任務
        fmt.Println("received", task) //處理任務
    }
}

總結hchan結構體的主要組成部分有四個：