GO 中 Chan 實(shí)現(xiàn)原理分享

嗨，我是小魔童哪吒，還記得咱們之前分享過GO 通道 和sync包的使用嗎？咱們來回顧一下

• 分享了通道是什么，通道的種類
• 無緩沖，有緩沖，單向通道具體對(duì)應(yīng)什么
• 對(duì)于通道的具體實(shí)踐
• 分享了關(guān)于通道的異常情況整理
• 簡單分享了sync包的使用

要是對(duì)上述內(nèi)容還有點(diǎn)興趣的話，歡迎查看文章 GO通道和 sync 包的分享

chan 是什么

是一種特殊的類型，是連接并發(fā)goroutine的管道

channel 通道是可以讓一個(gè) goroutine 協(xié)程發(fā)送特定值到另一個(gè) goroutine 協(xié)程的通信機(jī)制。

通道像一個(gè)傳送帶或者隊(duì)列，總是遵循先入先出（First In First Out）的規(guī)則，保證收發(fā)數(shù)據(jù)的順序，這一點(diǎn)和管道是一樣的

一個(gè)協(xié)程從通道的一頭放入數(shù)據(jù)，另一個(gè)協(xié)程從通道的另一頭讀出數(shù)據(jù)

每一個(gè)通道都是一個(gè)具體類型的導(dǎo)管，聲明 channel 的時(shí)候需要為其指定元素類型。

本篇文章主要是分享關(guān)于通道的實(shí)現(xiàn)原理，關(guān)于通道的使用，可以查看文章 GO通道和 sync 包的分享 ，這里有詳細(xì)的說明

GO 中 Chan 的底層數(shù)據(jù)結(jié)構(gòu)

了解每一個(gè)組件或者每一個(gè)數(shù)據(jù)類型的實(shí)現(xiàn)原理，咱們都會(huì)去看源碼中的數(shù)據(jù)結(jié)構(gòu)是如何設(shè)計(jì)的

同樣，我們一起來看看 GO 的 Chan 的數(shù)據(jù)結(jié)構(gòu)

GO 的 Chan 的源碼實(shí)現(xiàn)是在 ： 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
}

hchan 是實(shí)現(xiàn)通道的核心數(shù)據(jù)結(jié)構(gòu)，對(duì)應(yīng)的成員也是不少，咱們根據(jù)源碼注釋一個(gè)參數(shù)一個(gè)參數(shù)的來看看

根據(jù)上面的參數(shù)，我們或多或少就可以知道 GO 中的通道實(shí)現(xiàn)原理設(shè)計(jì)了哪些知識(shí)點(diǎn)：

• 指針
• 環(huán)形隊(duì)列
• 協(xié)程
• 互斥鎖

我們順便再來看看上述成員的協(xié)程隊(duì)列 waitq 對(duì)應(yīng)的是啥樣的數(shù)據(jù)結(jié)構(gòu)

type waitq struct {
   first *sudog
   last  *sudog
}

sudog 結(jié)構(gòu)是在 src/runtime/runtime2.go中 ，咱們順便多學(xué)一手

// sudog represents a g in a wait list, such as for sending/receiving
// on a channel.
type sudog struct {
   // The following fields are protected by the hchan.lock of the
   // channel this sudog is blocking on. shrinkstack depends on
   // this for sudogs involved in channel ops.

   g *g

   next *sudog
   prev *sudog
   elem unsafe.Pointer // data element (may point to stack)

   // The following fields are never accessed concurrently.
   // For channels, waitlink is only accessed by g.
   // For semaphores, all fields (including the ones above)
   // are only accessed when holding a semaRoot lock.

   acquiretime int64
   releasetime int64
   ticket      uint32

   // isSelect indicates g is participating in a select, so
   // g.selectDone must be CAS'd to win the wake-up race.
   isSelect bool

   // success indicates whether communication over channel c
   // succeeded. It is true if the goroutine was awoken because a
   // value was delivered over channel c, and false if awoken
   // because c was closed.
   success bool

   parent   *sudog // semaRoot binary tree
   waitlink *sudog // g.waiting list or semaRoot
   waittail *sudog // semaRoot
   c        *hchan // channel
}

根據(jù)源碼注釋，咱們大致知道sudog 是干啥的

Sudog表示等待列表中的 g，例如在一個(gè)通道上發(fā)送/接收

Sudog是很必要的，因?yàn)間↔synchronization對(duì)象關(guān)系是多對(duì)多

一個(gè) g 可能在很多等候隊(duì)列上，所以一個(gè) g 可能有很多sudogs

而且許多 g 可能在等待同一個(gè)同步對(duì)象，所以一個(gè)對(duì)象可能有許多sudogs

咱們抓住主要矛盾

Sudog的數(shù)據(jù)結(jié)構(gòu)，主要的東西就是一個(gè) g 和一個(gè) elem ，

g，上面有說到他和 Sudog的對(duì)應(yīng)關(guān)系

無論是讀通道還是寫通道，都會(huì)需要 elem

讀通道

數(shù)據(jù)會(huì)從hchan的隊(duì)列中，拷貝到sudog的elem中

寫通道

與讀通道類似，是將數(shù)據(jù)從 sudog 的elem處拷貝到hchan的隊(duì)列中

咱們來畫個(gè)圖看看

此處咱們畫一個(gè) hchan的結(jié)構(gòu)，主要畫一下 recvq等待讀取消息的協(xié)程隊(duì)列，此處的隊(duì)列，實(shí)際上就是用鏈表來實(shí)現(xiàn)的

recvq會(huì)對(duì)應(yīng)到 waitq結(jié)構(gòu)，waitq 分為first頭結(jié)點(diǎn) 和 last尾節(jié)點(diǎn) 結(jié)構(gòu)分別是 sudog

sudog里面 elem存放具體的數(shù)據(jù)，next 指針指向下一個(gè) sudog，直到指到last 的 sudog

通過上述的，應(yīng)該就能明白 GO 中的 chan 基本結(jié)構(gòu)了吧

咱來再來詳細(xì)看看 hchan 中其他參數(shù)都具體是啥意思

• dataqsiz 對(duì)應(yīng)的環(huán)形隊(duì)列是啥樣的
• 寫 sendq和 讀 recvq 等待隊(duì)列是啥樣的
• elemtype元素類型信息又是啥

dataqsiz 對(duì)應(yīng)的環(huán)形隊(duì)列是啥樣的

環(huán)形隊(duì)列，故名思議就是 一個(gè)首尾連接，成環(huán)狀的隊(duì)列

GO 中的 chan內(nèi)部的環(huán)形隊(duì)列，主要作用是作為緩沖區(qū)

這個(gè)環(huán)形隊(duì)列的長度，我們?cè)趧?chuàng)建隊(duì)列的時(shí)候， 也就是創(chuàng)建 hchan 結(jié)構(gòu)的時(shí)候，就已經(jīng)指定好了的

就是 dataqsiz ，環(huán)形隊(duì)列的長度

咱們畫個(gè)圖清醒一下

上圖需要表達(dá)的意思是這個(gè)樣子的，上述的隊(duì)列是循環(huán)隊(duì)列，默認(rèn)首尾連接哦：

• dataqsiz 表示 循環(huán)隊(duì)列的長度是 8 個(gè)
• qcount 表示 當(dāng)前隊(duì)列中有 5 個(gè)元素
• buf 是指針，指向循環(huán)隊(duì)列頭
• sendx 是發(fā)送隊(duì)列的下標(biāo)，這里為 1 ，則指向隊(duì)列的第 2 個(gè)區(qū)域 ，這個(gè)參數(shù)可選范圍是 [0 , 8)
• recvx 是接收隊(duì)列的下標(biāo)，這里為 4 ，則指向的是 隊(duì)列的第 5 個(gè)區(qū)域進(jìn)行讀取數(shù)據(jù)

這里順帶提一下，hchan 中讀取數(shù)據(jù)還是寫入數(shù)據(jù)，都是需要去拿 lock 互斥鎖的，同一個(gè)通道，在同一個(gè)時(shí)刻只能允許一個(gè)協(xié)程進(jìn)行讀寫

寫 sendq和 讀 recvq 等待隊(duì)列是啥樣的

hchan 結(jié)構(gòu)中的 2 個(gè)協(xié)程隊(duì)列，一個(gè)是用于讀取數(shù)據(jù)，一個(gè)是用于發(fā)送數(shù)據(jù)，他們都是等待隊(duì)列，我們來看看這個(gè)等待隊(duì)列都是咋放數(shù)據(jù)上去的，分別有啥特性需要注意

當(dāng)從通道中讀取 或者 發(fā)送數(shù)據(jù)：

• 若通道的緩沖區(qū)為空，或者沒有緩沖區(qū)，此時(shí)從通道中讀取數(shù)據(jù)，則協(xié)程是會(huì)被阻塞的
• 若通道緩沖區(qū)為滿，或者沒有緩沖區(qū)，此時(shí)從通道中寫數(shù)據(jù)，則協(xié)程仍然也會(huì)被阻塞

這些被阻塞的協(xié)程就會(huì)被放到等待隊(duì)列中，按照讀 和 寫 的動(dòng)作來進(jìn)行分類為寫 sendq和 讀 recvq 隊(duì)列

那么這些阻塞的協(xié)程，啥時(shí)候會(huì)被喚醒呢？

看過之前的文章 GO通道和 sync 包的分享，應(yīng)該就能知道

我們?cè)趤砘仡櫼幌拢@篇文章的表格，通道會(huì)存在的異常情況：

此時(shí)，我們就知道，具體什么時(shí)候被阻塞的協(xié)程會(huì)被喚醒了

• 因?yàn)樽x阻塞的協(xié)程，會(huì)被通道中的寫入數(shù)據(jù)的協(xié)程喚醒，反之亦然
• 因?yàn)閷懽枞膮f(xié)程，也會(huì)被通道中讀取數(shù)據(jù)的協(xié)程喚醒

elemtype元素類型信息又是啥

這個(gè)元素類型信息就不難理解了，對(duì)于我們使用通道，創(chuàng)建通道的時(shí)候我們需要填入通道中數(shù)據(jù)的類型，一個(gè)通道，只能寫一種數(shù)據(jù)類型，指的就是這里的elemtype

另外 hchan 還有一個(gè)成員是elemsize，代表上述元素類型的占用空間大小

那么這倆成員有啥作用呢？

elemtype和elemsize就可以計(jì)算指定類型的數(shù)據(jù)占用空間大小了

前者用于在數(shù)據(jù)傳遞的過程中進(jìn)行賦值

后者可以用來在環(huán)形隊(duì)列中定位具體的元素

創(chuàng)建 chan 是咋實(shí)現(xiàn)的

我們?cè)賮沓虺?chan.go 的源碼實(shí)現(xiàn) ，看到源碼中的 makechan 具體實(shí)現(xiàn)

func makechan(t *chantype, size int) *hchan {
   elem := t.elem

   // compiler checks this but be safe.
   if elem.size >= 1<<16 {
      throw("makechan: invalid channel element type")
   }
   if hchanSize%maxAlign != 0 || elem.align > maxAlign {
      throw("makechan: bad alignment")
   }

   mem, overflow := math.MulUintptr(elem.size, uintptr(size))
   if overflow || mem > maxAlloc-hchanSize || size < 0 {
      panic(plainError("makechan: size out of range"))
   }

   // Hchan does not contain pointers interesting for GC when elements stored in buf do not contain pointers.
   // buf points into the same allocation, elemtype is persistent.
   // SudoG's are referenced from their owning thread so they can't be collected.
   // TODO(dvyukov,rlh): Rethink when collector can move allocated objects.
   var c *hchan
   switch {
   case mem == 0:
      // Queue or element size is zero.
      c = (*hchan)(mallocgc(hchanSize, nil, true))
      // Race detector uses this location for synchronization.
      c.buf = c.raceaddr()
   case elem.ptrdata == 0:
      // Elements do not contain pointers.
      // Allocate hchan and buf in one call.
      c = (*hchan)(mallocgc(hchanSize+mem, nil, true))
      c.buf = add(unsafe.Pointer(c), hchanSize)
   default:
      // Elements contain pointers.
      c = new(hchan)
      c.buf = mallocgc(mem, elem, true)
   }

   c.elemsize = uint16(elem.size)
   c.elemtype = elem
   c.dataqsiz = uint(size)
   lockInit(&c.lock, lockRankHchan)

   if debugChan {
      print("makechan: chan=", c, "; elemsize=", elem.size, "; dataqsiz=", size, "\n")
   }
   return c
}

如上源碼實(shí)際上就是初始化 chan 對(duì)應(yīng)的成員，其中循環(huán)隊(duì)列 buf 的大小，是由 makechan 函數(shù)傳入的 類型信息和緩沖區(qū)長度決定的，也就是makechan 的入?yún)?/p>

可以通過上述代碼的 3 個(gè)位置就可以知道

// 1
func makechan(t *chantype, size int) *hchan
// 2
mem, overflow := math.MulUintptr(elem.size, uintptr(size))
// 3
var c *hchan
   switch {
   case mem == 0:
      // Queue or element size is zero.
      c = (*hchan)(mallocgc(hchanSize, nil, true))
      // Race detector uses this location for synchronization.
      c.buf = c.raceaddr()
   case elem.ptrdata == 0:
      // Elements do not contain pointers.
      // Allocate hchan and buf in one call.
      c = (*hchan)(mallocgc(hchanSize+mem, nil, true))
      c.buf = add(unsafe.Pointer(c), hchanSize)
   default:
      // Elements contain pointers.
      c = new(hchan)
      c.buf = mallocgc(mem, elem, true)
   }

讀寫 chan 的基本流程

第一張圖說明白向 chan 寫入數(shù)據(jù)的流程

向通道中寫入數(shù)據(jù)，我們會(huì)涉及sendq 、 recvq隊(duì)列，和循環(huán)隊(duì)列的資源問題

根據(jù)圖示可以看出向通道中寫入數(shù)據(jù)分為 3 種情況：

• 寫入數(shù)據(jù)的時(shí)候，若recvq 隊(duì)列為空，且循環(huán)隊(duì)列有空位，那么就直接將數(shù)據(jù)寫入到 循環(huán)隊(duì)列的隊(duì)尾 即可
• 若recvq 隊(duì)列為空，且循環(huán)隊(duì)列無空位，則將當(dāng)前的協(xié)程放到sendq等待隊(duì)列中進(jìn)行阻塞，等待被喚醒，當(dāng)被喚醒的時(shí)候，需要寫入的數(shù)據(jù)，已經(jīng)被讀取出來，且已經(jīng)完成了寫入操作
• 若recvq 隊(duì)列為不為空，那么可以說明循環(huán)隊(duì)列中沒有數(shù)據(jù)，或者循環(huán)隊(duì)列是空的，即沒有緩沖區(qū)（向無緩沖的通道寫入數(shù)據(jù)），此時(shí)，直接將recvq等待隊(duì)列中取出一個(gè)G，寫入數(shù)據(jù)，喚醒G，完成寫入操作

第二張圖說明白向 chan 讀取數(shù)據(jù)的流程

向通道中讀取數(shù)據(jù)，我們會(huì)涉及sendq 、 recvq隊(duì)列，和循環(huán)隊(duì)列的資源問題

根據(jù)圖示可以看出向通道中讀取數(shù)據(jù)分為 4 種情況：

• 若sendq為空，且循環(huán)隊(duì)列無元素的時(shí)候，那就將當(dāng)前的協(xié)程加入recvq等待隊(duì)列，把recvq等待隊(duì)列對(duì)頭的一個(gè)協(xié)程取出來，喚醒，讀取數(shù)據(jù)
• 若sendq為空，且循環(huán)隊(duì)列有元素的時(shí)候，直接讀取循環(huán)隊(duì)列中的數(shù)據(jù)即可
• 若sendq有數(shù)據(jù)，且循環(huán)隊(duì)列有元素的時(shí)候，直接讀取循環(huán)隊(duì)列中的數(shù)據(jù)即可，且把sendq隊(duì)列取一個(gè)G放到循環(huán)隊(duì)列中，進(jìn)行補(bǔ)充
• 若sendq有數(shù)據(jù)，且循環(huán)隊(duì)列無元素的時(shí)候，則從sendq取出一個(gè)G，并且喚醒他，進(jìn)行數(shù)據(jù)讀取操作

上面說了通道的創(chuàng)建，讀寫，那么通道咋關(guān)閉？

通道的關(guān)閉，我們?cè)趹?yīng)用的時(shí)候直接 close 就搞定了，那么對(duì)應(yīng)close的時(shí)候，底層的隊(duì)列都是做了啥呢？

若關(guān)閉了當(dāng)前的通道，那么系統(tǒng)會(huì)把recvq 讀取數(shù)據(jù)的等待隊(duì)列里面的所有協(xié)程，全部喚醒，這里面的每一個(gè)G 寫入的數(shù)據(jù) 默認(rèn)就寫個(gè) nil，因?yàn)橥ǖ狸P(guān)閉了，從關(guān)閉的通道里面讀取數(shù)據(jù)，讀到的是nil

系統(tǒng)還會(huì)把sendq寫數(shù)據(jù)的等待隊(duì)列里面的每一個(gè)協(xié)程喚醒，但是此時(shí)就會(huì)有問題了，向已經(jīng)關(guān)閉的協(xié)程里面寫入數(shù)據(jù)，會(huì)報(bào)panic

我們?cè)賮硎崂硪幌?，什么情況下對(duì)通道操作，會(huì)報(bào)panic，咱們現(xiàn)在對(duì)之前提到的表格再來補(bǔ)充一波

• 關(guān)閉一個(gè)已經(jīng)被關(guān)閉了的通道，會(huì)報(bào)panic
• 關(guān)閉一個(gè)未初始化的通道，即為nil的通道，也會(huì)報(bào)panic
• 向一個(gè)已經(jīng)關(guān)閉的通道寫入數(shù)據(jù)，會(huì)報(bào)panic

你以為這就完了嗎？

GO 里面Chan 一般會(huì)和 select 搭配使用，我們最后來簡單說一下GO 的 通道咋和select使用吧

GO 里面select 就和 C/C++里面的多路IO復(fù)用類似，在C/C++中多路IO復(fù)用有如下幾種方式

• SELECT
• POLL
• EPOLL

都可以自己去模擬實(shí)現(xiàn)多路IO復(fù)用，各有利弊，一般使用的最多的是 EPOLL，且C/C++也有對(duì)應(yīng)的網(wǎng)絡(luò)庫

當(dāng)我們寫GO 的多路IO復(fù)用的時(shí)候，那就相當(dāng)爽了，GO 默認(rèn)支持select 關(guān)鍵字

SELECT 簡單使用

我們就來看看都是咋用的，不廢話，咱直接上DEMO

package main

import (
   "log"
   "time"
)

func main() {

   // 簡單設(shè)置log參數(shù)
   log.SetFlags(log.Lshortfile | log.LstdFlags)

   // 創(chuàng)建 2 個(gè)通道，元素?cái)?shù)據(jù)類型為 int，緩沖區(qū)大小為 5
   var ch1 = make(chan int, 5)
   var ch2 = make(chan int, 5)

   // 分別向通道中各自寫入數(shù)據(jù)，咱默認(rèn)寫1吧
   // 直接寫一個(gè)匿名函數(shù) 向通道中添加數(shù)據(jù)
   go func (){
      var num = 1
      for {
         ch1 <- num
         num += 1
         time.Sleep(1 * time.Second)
      }
   }()

   go func (){
      var num = 1
      for {
         ch2 <- num
         num += 1
         time.Sleep(1 * time.Second)
      }
   }()

   for {
      select {// 讀取數(shù)據(jù)
      case num := <-ch1:
         log.Printf("read ch1 data is  %d\n", num)

      case num := <-ch2:
         log.Printf("read ch2 data is: %d\n", num)

      default:
         log.Printf("ch1 and ch2 is empty\n")
          // 休息 1s 再讀
         time.Sleep(1 * time.Second)
      }
   }
}

運(yùn)行效果

2021/06/18 17:43:06 main.go:54: ch1 and ch2 is empty
2021/06/18 17:43:07 main.go:48: read ch1 data is  1
2021/06/18 17:43:07 main.go:48: read ch1 data is  2
2021/06/18 17:43:07 main.go:51: read ch2 data is: 1
2021/06/18 17:43:07 main.go:51: read ch2 data is: 2
2021/06/18 17:43:07 main.go:54: ch1 and ch2 is empty
2021/06/18 17:43:08 main.go:48: read ch1 data is  3
2021/06/18 17:43:08 main.go:51: read ch2 data is: 3
2021/06/18 17:43:08 main.go:54: ch1 and ch2 is empty
2021/06/18 17:43:09 main.go:48: read ch1 data is  4
2021/06/18 17:43:09 main.go:51: read ch2 data is: 4
2021/06/18 17:43:09 main.go:54: ch1 and ch2 is empty
2021/06/18 17:43:10 main.go:51: read ch2 data is: 5
2021/06/18 17:43:10 main.go:48: read ch1 data is  5

從運(yùn)行結(jié)果來看，select 監(jiān)控的 2個(gè) 通道，讀取到的數(shù)據(jù)是隨機(jī)的

可是我們看到case這個(gè)關(guān)鍵字，是不是會(huì)想到 switch ... case...，此處的的case 是順序運(yùn)行的（GO 中沒有switch），select 里面的 case 應(yīng)該也是順序運(yùn)行才對(duì)呀，為啥結(jié)果是隨機(jī)的?

大家要是感興趣的話，可以深入研究一下，咱們今天就先到這里了。

總結(jié)

• 分享了 GO 中通道是什么
• 通道的底層數(shù)據(jù)結(jié)構(gòu)詳細(xì)解析
• 通道在GO源碼中是如何實(shí)現(xiàn)的
• Chan 讀寫的基本原理
• 關(guān)閉通道會(huì)出現(xiàn)哪些異常，panic
• select 的簡單應(yīng)用

到此這篇關(guān)于GO語言中Chan實(shí)現(xiàn)原理的示例詳解的文章就介紹到這了,更多相關(guān)GO語言Chan內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論