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

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

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

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

chan 是什么

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

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

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

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

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

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

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

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

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

GO 的 Chan 的源碼實現(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 是實現(xiàn)通道的核心數(shù)據(jù)結(jié)構(gòu)，對應(yīng)的成員也是不少，咱們根據(jù)源碼注釋一個參數(shù)一個參數(shù)的來看看

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

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

我們順便再來看看上述成員的協(xié)程隊列 waitq 對應(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，例如在一個通道上發(fā)送/接收

Sudog是很必要的，因為g↔synchronization對象關(guān)系是多對多

一個 g 可能在很多等候隊列上，所以一個 g 可能有很多sudogs

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

咱們抓住主要矛盾

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

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

無論是讀通道還是寫通道，都會需要 elem

讀通道

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

寫通道

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

咱們來畫個圖看看

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

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

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

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

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

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

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

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

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

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

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

咱們畫個圖清醒一下

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

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

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

寫 sendq和 讀 recvq 等待隊列是啥樣的

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

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

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

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

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

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

我們在來回顧一下，這篇文章的表格，通道會存在的異常情況：

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

• 因為讀阻塞的協(xié)程，會被通道中的寫入數(shù)據(jù)的協(xié)程喚醒，反之亦然
• 因為寫阻塞的協(xié)程，也會被通道中讀取數(shù)據(jù)的協(xié)程喚醒

elemtype元素類型信息又是啥

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

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

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

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

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

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

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

我們再來瞅瞅 chan.go 的源碼實現(xiàn) ，看到源碼中的 makechan 具體實現(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
}

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

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

// 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ù)，我們會涉及sendq 、 recvq隊列，和循環(huán)隊列的資源問題

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

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

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

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

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

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

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

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

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

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

我們再來梳理一下，什么情況下對通道操作，會報panic，咱們現(xiàn)在對之前提到的表格再來補充一波

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

你以為這就完了嗎？

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

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

• SELECT
• POLL
• EPOLL

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

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

SELECT 簡單使用

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

package main

import (
   "log"
   "time"
)

func main() {

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

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

   // 分別向通道中各自寫入數(shù)據(jù)，咱默認寫1吧
   // 直接寫一個匿名函數(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)
      }
   }
}

運行效果

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

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

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

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

總結(jié)

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

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

最新評論