摘要：比如主協程啟動個子協程，主協程等待所有子協程退出后再繼續后續流程，這種場景下也可輕易實現。這個例子中，父協程僅僅是等待子協程結束，其實父協程也可以向管道中寫入數據通知子協程結束，這時子協程需要定期地探測管道中是否有消息出現。

一.設計原理

Go 語言中最常見的、也是經常被人提及的設計模式就是：

"不要通過共享內存來通信，我們應該使用通信來共享內存"

通過共享內存來通信是直接讀取內存的數據,而通過通信來共享內存，是通過發送消息的方式來進行同步。

而通過發送消息來同步的這種方式常見的就是 Go 采用的通信順序進程 CSP(Communication Sequential Process) 模型以及 Erlang 采用的 Actor 模型，這兩種方式都是通過通信來共享內存。

如下圖所示

大部分的語言采用的都是第一種方式直接去操作內存，然后通過互斥鎖，CAS 等操作來保證并發安全。Go 引入了 Channel 和 Goroutine 實現 CSP 模型來解耦這個操作。

• 優點:

  • 在 Goroutine 當中我們就不用手動去做資源的鎖定與釋放，同時將生產者和消費者進行了解耦，Channel 其實和消息隊列很相似。

• 缺點：

  • 由于 Channel 底層也是通過這些低級的同步原語實現的，所以性能上會差一些，如果有極高的性能要求時也可以用 sync 包中提供的低級同步原語

先入先出

目前的 Channel 收發操作均遵循了先進先出的設計，具體規則如下：

• 先從 Channel 讀取數據的 Goroutine 會先接收到數據；
• 先向 Channel 發送數據的 Goroutine 會得到先發送數據的權利；

無鎖管道

鎖(Lock) 是一種常見的并發控制技術，我們一般會將鎖分成樂觀鎖 和 悲觀鎖，即樂觀并發控制和悲觀并發控制，無鎖（lock-free）隊列更準確的描述是使用樂觀并發控制的隊列。樂觀并發控制也叫樂觀鎖，很多人都會誤以為樂觀鎖是與悲觀鎖差不多，然而它并不是真正的鎖，只是一種并發控制的思想.

樂觀并發控制本質上是基于驗證的協議，我們使用原子指令 CAS（compare-and-swap 或者 compare-and-set）在多線程中同步數據，無鎖隊列的實現也依賴這一原子指令。

從某種程度上說，Channel 是一個用于同步和通信的有鎖隊列，使用互斥鎖解決程序中可能存在的線程競爭問題

Go 語言社區也在 2014 年提出了無鎖 Channel 的實現方案，該方案將 Channel 分成了以下三種類型：

1. 同步 Channel — 無緩沖區，發送方會直接將數據交給（Handoff）接收方
   

2. 異步channel: 基于環形緩存的傳統生產者消費者模型；
   

3. chan struct{} 類型的異步 Channel — struct{} 類型不占用內存空間，不需要實現緩沖區和直接發送（Handoff）的語義；

二.數據結構

Go 語言的 Channel 在運行時使用 runtime.hchan 結構體表示。我們在 Go 語言中創建新的 Channel 時，實際上創建的都是如下所示的結構：

type hchan struct {	qcount   uint           // 隊列中元素總數量	dataqsiz uint           // 循環隊列的長度	buf      unsafe.Pointer // 指向長度為 dataqsiz 的底層數組，只有在有緩沖時這個才有意義	elemsize uint16         // 能夠發送和接受的元素大小	closed   uint32         // 是否關閉	elemtype *_type         // 元素的類型	sendx    uint           // 當前已發送的元素在隊列當中的索引位置	recvx    uint           // 當前已接收的元素在隊列當中的索引位置	recvq    waitq          // 接收 Goroutine 鏈表	sendq    waitq          // 發送 Goroutine 鏈表	lock mutex              // 互斥鎖}// waitq 是一個雙向鏈表，里面保存了 goroutinetype waitq struct {	first *sudog	last  *sudog}

如下圖所示，channel 底層其實是一個循環隊列

三.創建管道

Go 語言中所有 Channel 的創建都會使用 make 關鍵字。創建的表達式使用 make(chan T, cap) 來創建 channel.

如果不向 make 傳遞表示緩沖區大小的參數，那么就會設置一個默認值 0，也就是當前的 Channel 不存在緩沖區。

四. 發送數據

當想要向 Channel 發送數據時，就需要使用 ch <- i 語句.

在發送數據的邏輯執行之前會先為當前 Channel 加鎖，防止多個線程并發修改數據。

如果 Channel 已經關閉，那么向該 Channel 發送數據時會報 “send on closed channel” 錯誤并中止程序。

4.1 直接發送

如果 Channel 沒有被關閉并且已經有處于讀等待的 Goroutine，會取出最先陷入等待的 Goroutine 并直接向它發送數據：

直接發送的過程稱為兩個部分：

1. 調用 runtime.sendDirect將發送的數據直接拷貝到 x = <-c 表達式中變量 x 所在的內存地址上；
2. 調用 runtime.goready 將等待接收數據的 Goroutine 標記成可運行狀態 Grunnable 并把該 Goroutine 放到發送方所在的處理器的 runnext 上等待執行，該處理器在下一次調度時會立刻喚醒數據的接收方；

需要注意的是，發送數據的過程只是將接收方的 Goroutine 放到了處理器的 runnext 中，程序沒有立刻執行該 Goroutine。

4.2 緩沖區

如果創建的 Channel 包含緩沖區并且 Channel 中的數據沒有裝滿，會使用 runtime.chanbuf 計算出下一個可以存儲數據的位置，然后通過 runtime.typedmemmove 將發送的數據拷貝到緩沖區中并增加 sendx 索引和 qcount 計數器。

4.3 阻塞發送

當 Channel 沒有接收者能夠處理數據時，向 Channel 發送數據會被下游阻塞，當然使用 select 關鍵字可以向 Channel 非阻塞地發送消息。

4.4 小結

可以簡單梳理和總結一下使用 ch <- i 表達式向 Channel 發送數據時遇到的幾種情況：

1. 如果當前 Channel 的 recvq 上存在已經被阻塞的 Goroutine，那么會直接將數據發送給當前 Goroutine 并將其設置成下一個運行的 Goroutine；
2. 如果 Channel 存在緩沖區并且其中還有空閑的容量，我們會直接將數據存儲到緩沖區 sendx 所在的位置上；
3. 如果不滿足上面的兩種情況，當前 Goroutine 也會陷入阻塞等待其他的協程從 Channel 接收數據；

五. 接收數據

可以使用兩種不同的方式去接收 Channel 中的數據：

i <- chi, ok <- ch

5.1 直接接收

會根據緩沖區的大小分別處理不同的情況

1. 如果 Channel 不存在緩沖區，直接從發送者那里把數據拷貝給接收變量
2. 如果是有緩沖 channel
   • 將隊列中的數據拷貝到接收方的內存地址；
   • 將發送隊列頭的數據拷貝到緩沖區中，釋放一個阻塞的發送方；

5.2 緩沖區

當 Channel 的緩沖區中已經包含數據時，從 Channel 中接收數據會直接從緩沖區中 的索引位置中取出數據進行處理：

5.3 阻塞接收

當 Channel 的發送隊列中不存在等待的 Goroutine 并且緩沖區中也不存在任何數據時，從管道中接收數據的操作會變成阻塞的，然而不是所有的接收操作都是阻塞的，與 select 語句結合使用時就可能會使用到非阻塞的接收操作：

六. 關閉channel

使用 close(ch) 來關閉 channel 最后會調用 runtime 中的 closechan 方法.

1. 關閉一個 nil 的 channel 和已關閉了的 channel 都會導致 panic
2. 關閉 channel 后會釋放所有因為 channel 而阻塞的 Goroutine

七. 使用場景

channel一般用于協程之間的通信，channel也可以用于并發控制。比如主協程啟動N個子協程，主協程等待所有子協程退出后再繼續后續流程，這種場景下channel也可輕易實現。

7.1 使用channel控制子協程

package mainimport (    "time"    "fmt")func Process(ch chan int) {    //Do some work...    time.Sleep(time.Second)    ch <- 1 //管道中寫入一個元素表示當前協程已結束}func main() {    channels := make([]chan int, 10) //創建一個10個元素的切片，元素類型為channel    for i:= 0; i < 10; i++ {        channels[i] = make(chan int) //切片中放入一個channel        go Process(channels[i])      //啟動協程，傳一個管道用于通信    }    for i, ch := range channels {  //遍歷切片，等待子協程結束        <-ch        fmt.Println("Routine ", i, " quit!")    }}

輸出：

Routine  0  quit!Routine  1  quit!Routine  2  quit!Routine  3  quit!Routine  4  quit!Routine  5  quit!Routine  6  quit!Routine  7  quit!Routine  8  quit!Routine  9  quit!

上面程序通過創建N個channel來管理N個協程，每個協程都有一個channel用于跟父協程通信，父協程創建完所有協程后等待所有協程結束。

這個例子中，父協程僅僅是等待子協程結束，其實父協程也可以向管道中寫入數據通知子協程結束，這時子協程需要定期地探測管道中是否有消息出現。

7.2 通過關閉 channel 實現一對多的通知

關閉 channel 時會釋放所有阻塞的 Goroutine，所以我們就可以利用這個特性來做一對多的通知，除了一對多之外我們還用了 done 做了多對一的通知，當然多對一這種情況還是建議直接使用 WaitGroup 即可

package mainimport (	"fmt"	"time")func run(stop <-chan struct{}, done chan<- struct{}) {	// 每一秒打印一次	for {		select {		case <-stop:			fmt.Println("stop...")			// 接收到停止后，向 done 管道中發送數據,然后退出函數			done <- struct{}{}			return		// 超時1秒將輸出hello		case <-time.After(time.Second):			fmt.Println("hello...")		}	}}func main() {	// 一對多，使用無緩沖通道，當關閉chan后，其他程序中接收到關閉信號后會統一執行操作	stop := make(chan struct{})	// 多對一，當關閉后，關閉一個chan, 寫入一個數據到管道中	done := make(chan struct{}, 10)	for i := 0; i < 10; i++ {		go run(stop, done)	}	// 模擬超時時間	time.Sleep(5 * time.Second)	close(stop)	for i := 0; i < 10; i++ {		<-done	}}

輸出：

hello...hello...hello......hello..stop...stop...stop...stop...stop...stop...stop...stop...stop...stop...

7.3 使用 channel 做異步編程

利用無緩沖channel,接收早于發送的特點，只有當數據寫入后，接收才能完成實現數據一致性

package mainimport (	"fmt")// 這里只能讀func read(c <-chan int) {	fmt.Println("read:", <-c)}// 這里只能寫func write(c chan<- int) {	c <- 0}func main() {	c := make(chan int)	go write(c)	read(c)}

7.4 超時控制

超時控制還是建議使用 context

func run(stop <-chan struct{}, done chan<- struct{}) {	// 每一秒打印一次 hello	for {		select {		case <-stop:			fmt.Println("stop...")			done <- struct{}{}			return		case <-time.After(time.Second):			fmt.Println("hello")		}	}}

7.5 協程池

根據控制Channel的緩存大小來控制并發執行的Goroutine的最大數目

var limit = make(chan int, 3)func main() {    for _, w := range work {        go func() {            limit <- 1            w()            <-limit        }()    }    select{}}

最后一句select{}是一個空的管道選擇語句，該語句會導致main線程阻塞，從而避免程序過早退出。還有for{}、<-make(chan int)等諸多方法可以達到類似的效果。因為main線程被阻塞了，如果需要程序正常退出的話可以通過調用os.Exit(0)實現。

八. 參考

1. https://draveness.me/golang/docs/part3-runtime/ch06-concurrency/golang-channel/
2. https://www.topgoer.cn/docs/gozhuanjia/chapter055.1-channel
3. https://lailin.xyz/post/go-training-week3-channel.html
4. https://chai2010.cn/advanced-go-programming-book/ch1-basic/ch1-05-mem.html