Go channel 實現原理分析

時間 2019-12-07

標籤 channel 實現原理分析简体版

原文原文鏈接

channel一個類型管道，經過它能夠在goroutine之間發送和接收消息。它是Golang在語言層面提供的goroutine間的通訊方式。git

衆所周知，Go依賴於稱爲CSP（Communicating Sequential Processes）的併發模型，經過Channel實現這種同步模式。Go併發的核心哲學是不要經過共享內存進行通訊; 相反，經過溝通分享記憶。github

下面以簡單的示例來演示Go如何經過channel來實現通訊。golang

package main
import (
    "fmt"
    "time"
)
func goRoutineA(a <-chan int) {
    val := <-a
    fmt.Println("goRoutineA received the data", val)
}
func goRoutineB(b chan int) {
    val := <-b
    fmt.Println("goRoutineB  received the data", val)
}
func main() {
    ch := make(chan int, 3)
    go goRoutineA(ch)
    go goRoutineB(ch)
    ch <- 3
    time.Sleep(time.Second * 1)
}

結果爲：數組

goRoutineA received the data 3併發

上面只是個簡單的例子，只輸出goRoutineA ，沒有執行goRoutineB，說明channel僅容許被一個goroutine讀寫。ide

接下來咱們經過源代碼分析程序執行過程，在講以前，若是不瞭解go 併發和調度相關知識。請閱讀這篇文章ui

https://github.com/guyan0319/...this

說道channel這裏不得不提通道的結構hchan。spa

hchan

源代碼在src/runtime/chan.go.net

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
}
type waitq struct {
    first *sudog
    last  *sudog
}

說明：

qcount uint // 當前隊列中剩餘元素個數
dataqsiz uint // 環形隊列長度，即緩衝區的大小，即make（chan T，N），N.
buf unsafe.Pointer // 環形隊列指針
elemsize uint16 // 每一個元素的大小
closed uint32 // 表示當前通道是否處於關閉狀態。建立通道後，該字段設置爲0，即通道打開; 經過調用close將其設置爲1，通道關閉。
elemtype *_type // 元素類型，用於數據傳遞過程當中的賦值；
sendx uint和recvx uint是環形緩衝區的狀態字段，它指示緩衝區的當前索引 - 支持數組，它能夠從中發送數據和接收數據。
recvq waitq // 等待讀消息的goroutine隊列
sendq waitq // 等待寫消息的goroutine隊列
lock mutex // 互斥鎖，爲每一個讀寫操做鎖定通道，由於發送和接收必須是互斥操做。

這裏sudog表明goroutine。

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

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

這裏咱們先討論帶緩衝

ch := make(chan int, 3)

建立通道後的緩衝通道結構

hchan struct {
    qcount uint : 0 
    dataqsiz uint : 3 
    buf unsafe.Pointer : 0xc00007e0e0 
    elemsize uint16 : 8 
    closed uint32 : 0 
    elemtype *runtime._type : &{
        size:8 
        ptrdata:0 
        hash:4149441018 
        tflag:7 
        align:8 
        fieldalign:8 
        kind:130 
        alg:0x55cdf0 
        gcdata:0x4d61b4 
        str:1055 
        ptrToThis:45152
        }
    sendx uint : 0 
    recvx uint : 0 
    recvq runtime.waitq : 
        {first:<nil> last:<nil>}
    sendq runtime.waitq : 
        {first:<nil> last:<nil>}
    lock runtime.mutex : 
        {key:0}
}

源代碼

func makechan(t *chantype, size int) *hchan {

   elem := t.elem
   ...
}

若是咱們建立一個帶buffer的channel，底層的數據模型以下圖：

向channel寫入數據

ch <- 3

底層hchan數據流程如圖

發送操做概要

一、鎖定整個通道結構。

二、肯定寫入。嘗試recvq從等待隊列中等待goroutine，而後將元素直接寫入goroutine。

三、若是recvq爲Empty，則肯定緩衝區是否可用。若是可用，從當前goroutine複製數據到緩衝區。

四、若是緩衝區已滿，則要寫入的元素將保存在當前正在執行的goroutine的結構中，而且當前goroutine將在sendq中排隊並從運行時掛起。

五、寫入完成釋放鎖。

這裏咱們要注意幾個屬性buf、sendx、lock的變化。

流程圖

從channel讀取操做

幾乎和寫入操做相同

代碼

func goRoutineA(a <-chan int) {
   val := <-a
   fmt.Println("goRoutineA received the data", val)
}

底層hchan數據流程如圖

這裏咱們要注意幾個屬性buf、sendx、recvx、lock的變化。

讀取操做概要

一、先獲取channel全局鎖

二、嘗試sendq從等待隊列中獲取等待的goroutine，

三、若有等待的goroutine，沒有緩衝區，取出goroutine並讀取數據，而後喚醒這個goroutine，結束讀取釋放鎖。

四、若有等待的goroutine，且有緩衝區（此時緩衝區已滿），從緩衝區隊首取出數據，再從sendq取出一個goroutine，將goroutine中的數據存入buf隊尾，結束讀取釋放鎖。

五、如沒有等待的goroutine，且緩衝區有數據，直接讀取緩衝區數據，結束讀取釋放鎖。

六、如沒有等待的goroutine，且沒有緩衝區或緩衝區爲空，將當前的goroutine加入sendq排隊，進入睡眠，等待被寫goroutine喚醒。結束讀取釋放鎖。

流程圖

recvq和sendq 結構

recvq和sendq基本上是鏈表，看起來基本以下

select

select就是用來監聽和channel有關的IO操做，當 IO 操做發生時，觸發相應的動做。

一個簡單的示例以下

package main

import (
   "fmt"
   "time"
)

func goRoutineD(ch chan int, i int) {
   time.Sleep(time.Second * 3)
   ch <- i
}
func goRoutineE(chs chan string, i string) {
   time.Sleep(time.Second * 3)
   chs <- i

}

func main() {
   ch := make(chan int, 5)
   chs := make(chan string, 5)

   go goRoutineD(ch, 5)
   go goRoutineE(chs, "ok")

    select {
    case msg := <-ch:
        fmt.Println(" received the data ", msg)
    case msgs := <-chs:
        fmt.Println(" received the data ", msgs)
    default:
        fmt.Println("no data received ")
        time.Sleep(time.Second * 1)
    }


}

運行程序，由於當前時間沒有到3s，因此select 選擇defult

no data received

修改程序，咱們註釋掉default，並多執行幾回結果爲

received the data 5

received the data ok

select語句會阻塞，直到監測到一個能夠執行的IO操做爲止，而這裏goRoutineD和goRoutineE睡眠時間是相同的，都是3s，從輸出可看出，從channel中讀出數據的順序是隨機的。

再修改代碼，goRoutineD睡眠時間改爲4s

func goRoutineD(ch chan int, i int) {
   time.Sleep(time.Second * 4)
   ch <- i
}

此時會先執行goRoutineE，select 選擇case msgs := <-chs。

range

能夠持續從channel讀取數據，一直到channel被關閉，當channel中沒有數據時會阻塞當前goroutine，與讀channel時阻塞處理機制同樣。

package main

import (
   "fmt"
   "time"
)

func goRoutineD(ch chan int, i int) {
   for   i := 1; i <= 5; i++{
      time.Sleep(time.Second * 1)
      ch <- i
   }

}
func chanRange(chanName chan int) {
   for e := range chanName {
      fmt.Printf("Get element from chan: %d\n", e)
      if len(chanName) <= 0 { // 若是現有數據量爲0，跳出循環
            break
      }
   }
}
func main() {
   ch := make(chan int, 5)
   go goRoutineD(ch, 5)
   chanRange(ch)

}

結果：
Get element from chan: 1
Get element from chan: 2
Get element from chan: 3
Get element from chan: 4
Get element from chan: 5

死鎖（deadlock）

指兩個或兩個以上的協程的執行過程當中，因爲競爭資源或因爲彼此通訊而形成的一種阻塞的現象。

在非緩衝信道若發生只流入不流出，或只流出不流入，就會發生死鎖。

下面是一些死鎖的例子

一、

package main

func main() {
   ch := make(chan int)
   ch <- 3
}

上面狀況，向非緩衝通道寫數據會發生阻塞，致使死鎖。解決辦法建立緩衝區 ch := make(chan int，3)

二、

package main

import (
   "fmt"
)

func main() {
   ch := make(chan int)
   fmt.Println(<-ch)
}

向非緩衝通道讀取數據會發生阻塞，致使死鎖。解決辦法開啓緩衝區，先向channel寫入數據。

三、

package main

func main() {
   ch := make(chan int, 3)
   ch <- 3
   ch <- 4
   ch <- 5
   ch <- 6
}

寫入數據超過緩衝區數量也會發生死鎖。解決辦法將寫入數據取走。

四、

package main

func main() {
    ch := make(chan int, 3)
    ch <- 1
    close(ch)
    ch <- 2
}

向關閉的channel寫入數據。解決辦法別向關閉的channel寫入數據。

死鎖的狀況有不少這裏再也不贅述。

參考：

https://codeburst.io/diving-d...

https://speakerdeck.com/kavya...

https://my.oschina.net/renhc/...

links

相關標籤/搜索

每日一句

每一个你不满意的现在，都有一个你没有努力的曾经。