go中的數據結構-通道channel

1. channel的使用

　　很多文章介紹channel的時候都和並發揉在一起，這里我想把它當做一種數據結構來單獨介紹它的實現原理。

　　channel，通道。golang中用於數據傳遞的一種數據結構。是golang中一種傳遞數據的方式，也可用作事件通知。

1.1 聲明、傳值、關閉

　　使用chan關鍵字聲明一個通道，在使用前必須先創建，操作符 <- 用於指定通道的方向，發送或接收。如果未指定方向，則為雙向通道。

//聲明和創建
var ch chan int      // 聲明一個傳遞int類型的channel
ch := make(chan int) // 使用內置函數make()定義一個channel
ch2 := make(chan interface{})         // 創建一個空接口類型的通道, 可以存放任意格式

type Equip struct{ /* 一些字段 */ }
ch2 := make(chan *Equip)             // 創建Equip指針類型的通道, 可以存放*Equip

//傳值
ch <- value          // 將一個數據value寫入至channel，這會導致阻塞，直到有其他goroutine從這個channel中讀取數據
value := <-ch        // 從channel中讀取數據，如果channel之前沒有寫入數據，也會導致阻塞，直到channel中被寫入數據為止

ch := make(chan interface{})  // 創建一個空接口通道
ch <- 0 // 將0放入通道中
ch <- "hello"  // 將hello字符串放入通道中

//關閉
close(ch)            // 關閉channel

　　把數據往通道中發送時，如果接收方一直都沒有接收，那么發送操作將持續阻塞。Go 程序運行時能智能地發現一些永遠無法發送成功的語句並報錯：

fatal error: all goroutines are asleep - deadlock! 
//運行時發現所有的 goroutine（包括main）都處於等待 goroutine。

1.2 四種重要的通道使用方式

無緩沖通道

　　通道默認是無緩沖的，無緩沖通道上的發送操作將會被阻塞，直到有其他goroutine從對應的通道上執行接收操作，數據傳送完成，通道繼續工作。

package main

import (
    "fmt"
    "time"
)
var done chan bool
func HelloWorld() {
    fmt.Println("Hello world goroutine")
    time.Sleep(1*time.Second)
    done <- true
}
func main() {
    done = make(chan bool)  // 創建一個channel
    go HelloWorld()
    <-done
}

//輸出
//Hello world goroutine

　　由於main不會等goroutine執行結束才返回，前文專門加了sleep輸出為了可以看到goroutine的輸出內容，那么在這里由於是阻塞的，所以無需sleep。

　　將代碼中”done <- true”和”<-done”，去掉再執行,沒有上面的輸出內容。

管道

　　通道可以用來連接goroutine，一邊的輸入是另一邊輸出。這就叫做管道：

 

package main

import (
    "fmt"
    "time"
)
var echo chan string
var receive chan string

// 定義goroutine 1 
func Echo() {
    time.Sleep(1*time.Second)
    echo <- "這是一次測試"
}

// 定義goroutine 2
func Receive() {
    temp := <- echo // 阻塞等待echo的通道的返回
    receive <- temp
}


func main() {
    echo = make(chan string)
    receive = make(chan string)

    go Echo()
    go Receive()

    getStr := <-receive   // 接收goroutine 2的返回

    fmt.Println(getStr)
}

　　輸出字符串："這是一次測試"。

　　在這里不一定要去關閉channel，因為底層的垃圾回收機制會根據它是否可以訪問來決定是否自動回收它。(這里不是根據channel是否關閉來決定的)

單向通道類型

package main

import (
    "fmt"
    "time"
)

// 定義goroutine 1
func Echo(out chan<- string) {   // 定義輸出通道類型
    time.Sleep(1*time.Second)
    out <- "這又是一次測試"
    close(out)
}

// 定義goroutine 2
func Receive(out chan<- string, in <-chan string) { // 定義輸出通道類型和輸入類型
    temp := <-in // 阻塞等待echo的通道的返回
    out <- temp
    close(out)
}


func main() {
    echo := make(chan string)
    receive := make(chan string)

    go Echo(echo)
    go Receive(receive, echo)

    getStr := <-receive   // 接收goroutine 2的返回

    fmt.Println(getStr)
}

　　輸出：這又是一次測試。

緩沖管道

　　goroutine的通道默認是是阻塞的，那么有什么辦法可以緩解阻塞？ 答案是：加一個緩沖區。

　　創建一個緩沖通道：

ch := make(chan string, 3) // 創建了緩沖區為3的通道

//==
len(ch)   // 長度計算
cap(ch)   // 容量計算

　　緩沖通道傳遞數據示意圖：

 

2. 內部結構 

　　Go語言channel是first-class的，意味着它可以被存儲到變量中，可以作為參數傳遞給函數，也可以作為函數的返回值返回。作為Go語言的核心特征之一，雖然channel看上去很高端，但是其實它僅僅就是一個數據結構而已，具體定義在 $GOROOT/src/runtime/chan.go里。如下：

type hchan struct {
  qcount uint   // 隊列中的總數據
  dataqsiz uint   // 循環隊列的大小
  buf unsafe.Pointer // 指向dataqsiz元素數組 指向環形隊列
  elemsize uint16  // 
  closed uint32 
  elemtype *_type // 元素類型
  sendx uint // 發送索引
  recvx uint // 接收索引
  recvq waitq // 接待員名單， 因recv而阻塞的等待隊列。
  sendq waitq // 發送服務員列表， 因send而阻塞的等待隊列。
  //鎖定保護hchan中的所有字段，以及幾個在此通道上阻止的sudogs中的字段。
  //按住此鎖定時不要更改另一個G的狀態（尤其是不要准備G），因為這可能會導致死鎖堆棧縮小。
  lock mutex 
}

 　　其核心是存放channel數據的環形隊列，由qcount和elemsize分別指定了隊列的容量和當前使用量。dataqsize是隊列的大小。elemalg是元素操作的一個Alg結構體，記錄下元素的操作，如copy函數，equal函數，hash函數等。

　　如果是帶緩沖區的chan，則緩沖區數據實際上是緊接着Hchan結構體中分配的。不帶緩沖的 channel ，環形隊列 size 則為 0。

c = (Hchan*)runtime.mal(n + hint*elem->size);

　　另一重要部分是recvq和sendq兩個雙向鏈表，前者是等待讀通道(<-channel)的goroutine隊列，后者是等待寫通道(channel <- xxx)的goroutine隊列。若一個goroutine阻塞於channel了，那么它就被掛在recvq或sendq隊列中。WaitQ是鏈表的定義，包含一個頭結點和一個尾結點：

struct    WaitQ
{
    SudoG*    first;
    SudoG*    last;
};

　　隊列中的每個成員是一個SudoG結構體變量：

struct    SudoG
{
    G*    g;        // g和selgen構成
    uint32    selgen;        // 指向g的弱指針
    SudoG*    link;
    int64    releasetime;
    byte*    elem;        // 數據元素
};

　　SudoG里主要結構是一個g和一個elem。elem用於存儲goroutine的數據。讀通道時，數據會從Hchan的buf隊列中拷貝到SudoG的elem域。寫通道時，數據則是由SudoG的elem域拷貝到Hchan的隊列中。

 

• buf是有緩沖的channel所特有的結構，用來存儲緩存數據。是個循環鏈表
• sendx和recvx用於記錄buf這個循環鏈表中的發送或者接收的index
• lock是個互斥鎖。

從最基本的開始-創建channel

創建一個緩沖channel

ch := make(chan int, 3)  //

底層操作就是從Heap中分配一塊內存，在內存中實例化一個hchan的結構體，並返回一個ch指針，使用 channel時，在函數之間的傳遞就是這個指針，這就是為什么函數傳遞中無需使用channel的指針，而直接用channel就行了，因為channel本身就是一個指針。　　

基本的寫channel操作，在底層運行時庫中對應的是一個runtime.chansend函數。

chan <- value

　　在運行時庫中會執行：

void runtime·chansend(ChanType *t, Hchan *c, byte *ep, bool *pres, void *pc)

　　其中c就是channel，ep是取變量v的地址。這里的傳值約定是調用者負責分配好ep的空間，僅需要簡單的取變量地址就夠了。pres參數是在select中的通道操作使用的。

　　這個函數首先判斷是同步或異步。同步chan不帶緩沖區，可能寫阻塞，而異步chan帶緩沖區，只有緩沖區滿才阻塞。在同步的情況下，首先查看Hchan結構體中的recvq鏈表時否為空，即是否有因為讀該管道而阻塞的goroutine。如果有則可以正常寫channel，否則操作會阻塞。

　　recvq不為空時，將一個SudoG結構體出隊列，將傳給通道的數據(函數參數ep)拷貝到SudoG結構體中的elem域，並將SudoG中的g放到就緒隊列中，狀態置為ready，然后函數返回。如果recvq為空，將當前goroutine阻塞。此時將一個SudoG結構體掛到通道的sendq鏈表中，這個SudoG中的elem域是參數eq，SudoG中的g是當前的goroutine。當前goroutine會被設置為waiting狀態並掛到等待隊列中。

　　異步時，如果緩沖區滿，要將當前goroutine和數據一起作為SudoG結構體掛在sendq隊列中。在channel緩沖區不滿的情況，直接將數據放到channel的緩沖區中，調用者返回。

實現細節：

• 當使用send (ch <- xx)或者recv ( <-ch)的時候，首先要鎖住hchan這個結構體。（lock字段）；
• 向緩沖區寫數據，按鏈表順序存放在buf中，直到緩沖區滿；
• 取數據的時候按鏈表順序讀取，符合FIFO的原則。

讀寫操作的細節都可以細化為：

• 第一，加鎖
• 第二，把數據從goroutine中copy到“隊列”中(或者從隊列中copy到goroutine中）。
• 第三，釋放鎖

　　讀channel操作也是類似的，對應的函數是runtime.chansend。基本過程類似。

　　當協程嘗試從未關閉的 channel 中讀取數據時，內部的操作如下：

• 當 buf 非空時，此時 recvq 必為空，buf 彈出一個元素給讀協程，讀協程獲得數據后繼續執行，此時若 sendq 非空，則從 sendq 中彈出一個寫協程轉入 running 狀態，待寫數據入隊列 buf ，此時讀取操作 <- ch 未阻塞；
• 當 buf 為空但 sendq 非空時(不帶緩沖的 channel)，則從 sendq 中彈出一個寫協程轉入 running 狀態，待寫數據直接傳遞給讀協程，讀協程繼續執行，此時讀取操作 <- ch 未阻塞；
• 當 buf 為空並且 sendq 也為空時，讀協程入隊列 recvq 並轉入 blocking 狀態，當后續有其他協程往 channel 寫數據時，讀協程才會重新轉入 running 狀態，此時讀取操作 <- ch 阻塞。

　　類似的，當協程嘗試往未關閉的 channel 中寫入數據時，內部的操作如下：

• 當隊列 recvq 非空時，此時隊列 buf 必為空，從 recvq 彈出一個讀協程接收待寫數據，此讀協程此時結束阻塞並轉入 running 狀態，寫協程繼續執行，此時寫入操作 ch <- 未阻塞；
• 當隊列 recvq 為空但 buf 未滿時，此時 sendq 必為空，寫協程的待寫數據入 buf 然后繼續執行，此時寫入操作 ch <- 未阻塞；
• 當隊列 recvq 為空並且 buf 為滿時，此時寫協程入隊列 sendq 並轉入 blokcing 狀態，當后續有其他協程從 channel 中讀數據時，寫協程才會重新轉入 running 狀態，此時寫入操作 ch <- 阻塞。

　　當關閉 non-nil channel 時，內部的操作如下：

• 當隊列 recvq 非空時，此時 buf 必為空，recvq 中的所有協程都將收到對應類型的零值然后結束阻塞狀態；
• 當隊列 sendq 非空時，此時 buf 必為滿，sendq 中的所有協程都會產生 panic ，在 buf 中數據仍然會保留直到被其他協程讀取。

　　空通道是指將一個channel賦值為nil，或者定義后不調用make進行初始化。按照Go語言的語言規范，讀寫空通道是永遠阻塞的。其實在函數runtime.chansend和runtime.chanrecv開頭就有判斷這類情況，如果發現參數c是空的，則直接將當前的goroutine放到等待隊列，狀態設置為waiting。

　　讀一個關閉的通道，永遠不會阻塞，會返回一個通道數據類型的零值。這個實現也很簡單，將零值復制到調用函數的參數ep中。寫一個關閉的通道，則會panic。關閉一個空通道，也會導致panic。

3. channel的高級用法

3.1 條件變量(condition variable)

　　類型於 POSIX 接口中線程通知其他線程某個事件發生的條件變量，channel 的特性也可以用來當成協程之間同步的條件變量。因為 channel 只是用來通知，所以 channel 中具體的數據類型和值並不重要，這種場景一般用 struct {} 作為 channel 的類型。

一對一通知

　　類似 pthread_cond_signal() 的功能，用來在一個協程中通知另個某一個協程事件發生：

package main

import (
    "fmt"
    "time"
)

func main() {
    ch := make(chan struct{})
    nums := make([]int, 100)

    go func() {
        time.Sleep(time.Second)
        for i := 0; i < len(nums); i++ {
            nums[i] = i
        }
        // send a finish signal
        ch <- struct{}{}
    }()

    // wait for finish signal
    <-ch
    fmt.Println(nums)
}

廣播通知

　　類似 pthread_cond_broadcast() 的功能。利用從已關閉的 channel 讀取數據時總是非阻塞的特性，可以實現在一個協程中向其他多個協程廣播某個事件發生的通知：

package main

import (
    "fmt"
    "time"
)

func main() {
    N := 10
    exit := make(chan struct{})
    done := make(chan struct{}, N)

    // start N worker goroutines
    for i := 0; i < N; i++ {
        go func(n int) {
            for {
                select {
                // wait for exit signal
                case <-exit:
                    fmt.Printf("worker goroutine #%d exit\n", n)
                    done <- struct{}{}
                    return
                case <-time.After(time.Second):
                    fmt.Printf("worker goroutine #%d is working...\n", n)
                }
            }
        }(i)
    }

    time.Sleep(3 * time.Second)
    // broadcast exit signal
    close(exit)
    // wait for all worker goroutines exit
    for i := 0; i < N; i++ {
        <-done
    }
    fmt.Println("main goroutine exit")
}

3.2 信號量

　　channel 的讀/寫相當於信號量的 P / V 操作，下面的示例程序中 channel 相當於信號量：

package main

import (
    "log"
    "math/rand"
    "time"
)

type Seat int
type Bar chan Seat

func (bar Bar) ServeConsumer(customerId int) {
    log.Print("-> consumer#", customerId, " enters the bar")
    seat := <-bar // need a seat to drink
    log.Print("consumer#", customerId, " drinks at seat#", seat)
    time.Sleep(time.Second * time.Duration(2+rand.Intn(6)))
    log.Print("<- consumer#", customerId, " frees seat#", seat)
    bar <- seat // free the seat and leave the bar
}

func main() {
    rand.Seed(time.Now().UnixNano())

    bar24x7 := make(Bar, 10) // the bar has 10 seats
    // Place seats in an bar.
    for seatId := 0; seatId < cap(bar24x7); seatId++ {
        bar24x7 <- Seat(seatId) // none of the sends will block
    }

    // a new consumer try to enter the bar for each second
    for customerId := 0; ; customerId++ {
        time.Sleep(time.Second)
        go bar24x7.ServeConsumer(customerId)
    }
}

3.3 互斥量

　　互斥量相當於二元信號里，所以 cap 為 1 的 channel 可以當成互斥量使用：

package main

import "fmt"

func main() {
    mutex := make(chan struct{}, 1) // the capacity must be one

    counter := 0
    increase := func() {
        mutex <- struct{}{} // lock
        counter++
        <-mutex // unlock
    }

    increase1000 := func(done chan<- struct{}) {
        for i := 0; i < 1000; i++ {
            increase()
        }
        done <- struct{}{}
    }

    done := make(chan struct{})
    go increase1000(done)
    go increase1000(done)
    <-done; <-done
    fmt.Println(counter) // 2000
}

4. 關閉 channel

　　關閉不再需要使用的 channel 並不是必須的。跟其他資源比如打開的文件、socket 連接不一樣，這類資源使用完后不關閉后會造成句柄泄露，channel 使用完后不關閉也沒有關系，channel 沒有被任何協程用到后最終會被 GC 回收。關閉 channel 一般是用來通知其他協程某個任務已經完成了。golang 也沒有直接提供判斷 channel 是否已經關閉的接口，雖然可以用其他不太優雅的方式自己實現一個：

func isClosed(ch chan int) bool {
    select {
    case <-ch:
        return true
    default:
    }
    return false
}

　　不過實現一個這樣的接口也沒什么必要。因為就算通過 isClosed() 得到當前 channel 當前還未關閉，如果試圖往 channel 里寫數據，仍然可能會發生 panic ，因為在調用 isClosed() 后，其他協程可能已經把 channel 關閉了。
關閉 channel 時應該注意以下准則：

• 不要在讀取端關閉 channel ，因為寫入端無法知道 channel 是否已經關閉，往已關閉的 channel 寫數據會 panic ；
• 有多個寫入端時，不要再寫入端關閉 channle ，因為其他寫入端無法知道 channel 是否已經關閉，關閉已經關閉的 channel 會發生 panic ；
• 如果只有一個寫入端，可以在這個寫入端放心關閉 channel 。

　　關閉 channel 粗暴一點的做法是隨意關閉，如果產生了 panic 就用 recover 避免進程掛掉。稍好一點的方案是使用標准庫的 sync 包來做關閉 channel 時的協程同步，不過使用起來也稍微復雜些。下面介紹一種優雅些的做法。

4.1 一寫多讀

　　這種場景下這個唯一的寫入端可以關閉 channel 用來通知讀取端所有數據都已經寫入完成了。讀取端只需要用 for range 把 channel 中數據遍歷完就可以了，當 channel 關閉時，for range 仍然會將 channel 緩沖中的數據全部遍歷完然后再退出循環：

package main

import (
    "fmt"
    "sync"
)

func main() {
    wg := &sync.WaitGroup{}
    ch := make(chan int, 100)

    send := func() {
        for i := 0; i < 100; i++ {
            ch <- i
        }
        // signal sending finish
        close(ch)
    }

    recv := func(id int) {
        defer wg.Done()
        for i := range ch {
            fmt.Printf("receiver #%d get %d\n", id, i)
        }
        fmt.Printf("receiver #%d exit\n", id)
    }

    wg.Add(3)
    go recv(0)
    go recv(1)
    go recv(2)
    send()

    wg.Wait()
}

4.2 多寫一讀

　　這種場景下雖然可以用 sync.Once 來解決多個寫入端重復關閉 channel 的問題，但更優雅的辦法設置一個額外的 channel ，由讀取端通過關閉來通知寫入端任務完成不要再繼續再寫入數據了：

package main

import (
    "fmt"
    "sync"
)

func main() {
    wg := &sync.WaitGroup{}
    ch := make(chan int, 100)
    done := make(chan struct{})

    send := func(id int) {
        defer wg.Done()
        for i := 0; ; i++ {
            select {
            case <-done:
                // get exit signal
                fmt.Printf("sender #%d exit\n", id)
                return
            case ch <- id*1000 + i:
            }
        }
    }

    recv := func() {
        count := 0
        for i := range ch {
            fmt.Printf("receiver get %d\n", i)
            count++
            if count >= 1000 {
                // signal recving finish
                close(done)
                return
            }
        }
    }

    wg.Add(3)
    go send(0)
    go send(1)
    go send(2)
    recv()

    wg.Wait()
}

4.2 多寫多讀

　　這種場景稍微復雜，和上面的例子一樣，也需要設置一個額外 channel 用來通知多個寫入端和讀取端。另外需要起一個額外的協程來通過關閉這個 channel 來廣播通知：

package main

import (
    "fmt"
    "sync"
    "time"
)

func main() {
    wg := &sync.WaitGroup{}
    ch := make(chan int, 100)
    done := make(chan struct{})

    send := func(id int) {
        defer wg.Done()
        for i := 0; ; i++ {
            select {
            case <-done:
                // get exit signal
                fmt.Printf("sender #%d exit\n", id)
                return
            case ch <- id*1000 + i:
            }
        }
    }

    recv := func(id int) {
        defer wg.Done()
        for {
            select {
            case <-done:
                // get exit signal
                fmt.Printf("receiver #%d exit\n", id)
                return
            case i := <-ch:
                fmt.Printf("receiver #%d get %d\n", id, i)
                time.Sleep(time.Millisecond)
            }
        }
    }

    wg.Add(6)
    go send(0)
    go send(1)
    go send(2)
    go recv(0)
    go recv(1)
    go recv(2)

    time.Sleep(time.Second)
    // signal finish
    close(done)
    // wait all sender and receiver exit
    wg.Wait()
}

　　channle 作為 golang 最重要的特性，用起來還是比較方便的。傳統的 C 里要實現類似的功能的話，一般需要用到 socket 或者 FIFO 來實現，另外還要考慮數據包的完整性與並發沖突的問題，channel 則屏蔽了這些底層細節，使用者只需要考慮讀寫就可以了。 channel 是引用類型，了解一下 channel 底層的機制對更好的使用 channel 還是很用必要的。雖然操作原語簡單，但涉及到阻塞的問題，使用不當可能會造成死鎖或者無限制的協程創建最終導致進程掛掉。