快捷導(dǎo)航

Go并發(fā)編程中使用channel的方法

更新時間：2021年11月23日 09:29:02 作者：深度思維者

本文給大家介紹Go并發(fā)編程中使用channel的方法，通過實(shí)例代碼給大家介紹的非常詳細(xì)，對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值，需要的朋友參考下吧

一.設(shè)計(jì)原理

Go 語言中最常見的、也是經(jīng)常被人提及的設(shè)計(jì)模式就是：

"不要通過共享內(nèi)存來通信，我們應(yīng)該使用通信來共享內(nèi)存"

通過共享內(nèi)存來通信是直接讀取內(nèi)存的數(shù)據(jù),而通過通信來共享內(nèi)存，是通過發(fā)送消息的方式來進(jìn)行同步。

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

如下圖所示

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

優(yōu)點(diǎn):

在 Goroutine 當(dāng)中我們就不用手動去做資源的鎖定與釋放，同時將生產(chǎn)者和消費(fèi)者進(jìn)行了解耦，Channel 其實(shí)和消息隊(duì)列很相似。

缺點(diǎn)：

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

先入先出

目前的 Channel 收發(fā)操作均遵循了先進(jìn)先出的設(shè)計(jì)，具體規(guī)則如下：

先從 Channel 讀取數(shù)據(jù)的 Goroutine 會先接收到數(shù)據(jù)；
先向 Channel 發(fā)送數(shù)據(jù)的 Goroutine 會得到先發(fā)送數(shù)據(jù)的權(quán)利；

無鎖管道

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

樂觀并發(fā)控制本質(zhì)上是基于驗(yàn)證的協(xié)議，我們使用原子指令 CAS（compare-and-swap 或者 compare-and-set）在多線程中同步數(shù)據(jù)，無鎖隊(duì)列的實(shí)現(xiàn)也依賴這一原子指令。

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

Go 語言社區(qū)也在 2014 年提出了無鎖 Channel 的實(shí)現(xiàn)方案，該方案將 Channel 分成了以下三種類型：

同步 Channel — 無緩沖區(qū)，發(fā)送方會直接將數(shù)據(jù)交給（Handoff）接收方

異步channel: 基于環(huán)形緩存的傳統(tǒng)生產(chǎn)者消費(fèi)者模型；

chan struct{} 類型的異步 Channel — struct{} 類型不占用內(nèi)存空間，不需要實(shí)現(xiàn)緩沖區(qū)和直接發(fā)送（Handoff）的語義；

二.數(shù)據(jù)結(jié)構(gòu)

Go 語言的 Channel 在運(yùn)行時使用 runtime.hchan 結(jié)構(gòu)體表示。我們在 Go 語言中創(chuàng)建新的 Channel 時，實(shí)際上創(chuàng)建的都是如下所示的結(jié)構(gòu)：

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

如下圖所示，channel 底層其實(shí)是一個循環(huán)隊(duì)列

三.創(chuàng)建管道

Go 語言中所有 Channel 的創(chuàng)建都會使用 make 關(guān)鍵字。創(chuàng)建的表達(dá)式使用 make(chan T, cap) 來創(chuàng)建 channel.

如果不向 make 傳遞表示緩沖區(qū)大小的參數(shù)，那么就會設(shè)置一個默認(rèn)值 0，也就是當(dāng)前的 Channel 不存在緩沖區(qū)。

四. 發(fā)送數(shù)據(jù)

當(dāng)想要向 Channel 發(fā)送數(shù)據(jù)時，就需要使用 ch <- i 語句.

在發(fā)送數(shù)據(jù)的邏輯執(zhí)行之前會先為當(dāng)前 Channel 加鎖，防止多個線程并發(fā)修改數(shù)據(jù)。

如果 Channel 已經(jīng)關(guān)閉，那么向該 Channel 發(fā)送數(shù)據(jù)時會報(bào) “send on closed channel” 錯誤并中止程序。

4.1 直接發(fā)送

如果 Channel 沒有被關(guān)閉并且已經(jīng)有處于讀等待的 Goroutine，會取出最先陷入等待的 Goroutine 并直接向它發(fā)送數(shù)據(jù)：

直接發(fā)送的過程稱為兩個部分：

調(diào)用 runtime.sendDirect將發(fā)送的數(shù)據(jù)直接拷貝到 x = <-c 表達(dá)式中變量 x 所在的內(nèi)存地址上；
調(diào)用 runtime.goready 將等待接收數(shù)據(jù)的 Goroutine 標(biāo)記成可運(yùn)行狀態(tài) Grunnable 并把該 Goroutine 放到發(fā)送方所在的處理器的 runnext 上等待執(zhí)行，該處理器在下一次調(diào)度時會立刻喚醒數(shù)據(jù)的接收方；

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

4.2 緩沖區(qū)

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

4.3 阻塞發(fā)送

當(dāng) Channel 沒有接收者能夠處理數(shù)據(jù)時，向 Channel 發(fā)送數(shù)據(jù)會被下游阻塞，當(dāng)然使用 select 關(guān)鍵字可以向 Channel 非阻塞地發(fā)送消息。

4.4 小結(jié)

可以簡單梳理和總結(jié)一下使用 ch <- i 表達(dá)式向 Channel 發(fā)送數(shù)據(jù)時遇到的幾種情況：

如果當(dāng)前 Channel 的 recvq 上存在已經(jīng)被阻塞的 Goroutine，那么會直接將數(shù)據(jù)發(fā)送給當(dāng)前 Goroutine 并將其設(shè)置成下一個運(yùn)行的 Goroutine；
如果 Channel 存在緩沖區(qū)并且其中還有空閑的容量，我們會直接將數(shù)據(jù)存儲到緩沖區(qū) sendx 所在的位置上；
如果不滿足上面的兩種情況，當(dāng)前 Goroutine 也會陷入阻塞等待其他的協(xié)程從 Channel 接收數(shù)據(jù)；

五. 接收數(shù)據(jù)

可以使用兩種不同的方式去接收 Channel 中的數(shù)據(jù)：

i <- ch

i, ok <- ch

5.1 直接接收

會根據(jù)緩沖區(qū)的大小分別處理不同的情況

如果 Channel 不存在緩沖區(qū)，直接從發(fā)送者那里把數(shù)據(jù)拷貝給接收變量如果是有緩沖 channel將隊(duì)列中的數(shù)據(jù)拷貝到接收方的內(nèi)存地址；將發(fā)送隊(duì)列頭的數(shù)據(jù)拷貝到緩沖區(qū)中，釋放一個阻塞的發(fā)送方；

5.2 緩沖區(qū)

當(dāng) Channel 的緩沖區(qū)中已經(jīng)包含數(shù)據(jù)時，從 Channel 中接收數(shù)據(jù)會直接從緩沖區(qū)中的索引位置中取出數(shù)據(jù)進(jìn)行處理：

5.3 阻塞接收

當(dāng) Channel 的發(fā)送隊(duì)列中不存在等待的 Goroutine 并且緩沖區(qū)中也不存在任何數(shù)據(jù)時，從管道中接收數(shù)據(jù)的操作會變成阻塞的，然而不是所有的接收操作都是阻塞的，與 select 語句結(jié)合使用時就可能會使用到非阻塞的接收操作：

六. 關(guān)閉channel

使用 close(ch) 來關(guān)閉 channel 最后會調(diào)用 runtime 中的 closechan 方法.

關(guān)閉一個 nil 的 channel 和已關(guān)閉了的 channel 都會導(dǎo)致 panic關(guān)閉 channel 后會釋放所有因?yàn)?channel 而阻塞的 Goroutine

七. 使用場景

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

7.1 使用channel控制子協(xié)程

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

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

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

7.2 通過關(guān)閉 channel 實(shí)現(xiàn)一對多的通知

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

package main
 
import (
	"fmt"
	"time"
)
 
func run(stop <-chan struct{}, done chan<- struct{}) {
 
	// 每一秒打印一次
	for {
		select {
		case <-stop:
			fmt.Println("stop...")
			// 接收到停止后，向 done 管道中發(fā)送數(shù)據(jù),然后退出函數(shù)
			done <- struct{}{}
			return
		// 超時1秒將輸出hello
		case <-time.After(time.Second):
			fmt.Println("hello...")
		}
	}
}
 
func main() {
	// 一對多，使用無緩沖通道，當(dāng)關(guān)閉chan后，其他程序中接收到關(guān)閉信號后會統(tǒng)一執(zhí)行操作
	stop := make(chan struct{})
 
	// 多對一，當(dāng)關(guān)閉后，關(guān)閉一個chan, 寫入一個數(shù)據(jù)到管道中
	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,接收早于發(fā)送的特點(diǎn)，只有當(dāng)數(shù)據(jù)寫入后，接收才能完成實(shí)現(xiàn)數(shù)據(jù)一致性

package main
 
import (
	"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 協(xié)程池

根據(jù)控制Channel的緩存大小來控制并發(fā)執(zhí)行的Goroutine的最大數(shù)目

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

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