快捷導(dǎo)航

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

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

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

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

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

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

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

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

如下圖所示

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

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

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

缺點(diǎn)：

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

先入先出

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

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

無(wú)鎖管道

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

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

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

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

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

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

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

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

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

type hchan struct {
	qcount   uint           // 隊(duì)列中元素總數(shù)量
	dataqsiz uint           // 循環(huán)隊(duì)列的長(zhǎng)度
	buf      unsafe.Pointer // 指向長(zhǎng)度為 dataqsiz 的底層數(shù)組，只有在有緩沖時(shí)這個(gè)才有意義
	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 是一個(gè)雙向鏈表，里面保存了 goroutine
type waitq struct {
	first *sudog
	last  *sudog
}

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

三.創(chuàng)建管道

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

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

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

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

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

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

4.1 直接發(fā)送

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

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

調(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í)會(huì)立刻喚醒數(shù)據(jù)的接收方；

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

4.2 緩沖區(qū)

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

4.3 阻塞發(fā)送

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

4.4 小結(jié)

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

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

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

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

i <- ch

i, ok <- ch

5.1 直接接收

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

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

5.2 緩沖區(qū)

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

5.3 阻塞接收

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

六. 關(guān)閉channel

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

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

七. 使用場(chǎng)景

channel一般用于協(xié)程之間的通信，channel也可以用于并發(fā)控制。比如主協(xié)程啟動(dòng)N個(gè)子協(xié)程，主協(xié)程等待所有子協(xié)程退出后再繼續(xù)后續(xù)流程，這種場(chǎng)景下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 //管道中寫入一個(gè)元素表示當(dāng)前協(xié)程已結(jié)束
}
 
func main() {
    channels := make([]chan int, 10) //創(chuàng)建一個(gè)10個(gè)元素的切片，元素類型為channel
 
    for i:= 0; i < 10; i++ {
        channels[i] = make(chan int) //切片中放入一個(gè)channel
        go Process(channels[i])      //啟動(dòng)協(xié)程，傳一個(gè)管道用于通信
    }
 
    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!

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

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

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

關(guān)閉 channel 時(shí)會(huì)釋放所有阻塞的 Goroutine，所以我們就可以利用這個(gè)特性來(lái)做一對(duì)多的通知，除了一對(duì)多之外我們還用了 done 做了多對(duì)一的通知，當(dāng)然多對(duì)一這種情況還是建議直接使用 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
		// 超時(shí)1秒將輸出hello
		case <-time.After(time.Second):
			fmt.Println("hello...")
		}
	}
}
 
func main() {
	// 一對(duì)多，使用無(wú)緩沖通道，當(dāng)關(guān)閉chan后，其他程序中接收到關(guān)閉信號(hào)后會(huì)統(tǒng)一執(zhí)行操作
	stop := make(chan struct{})
 
	// 多對(duì)一，當(dāng)關(guān)閉后，關(guān)閉一個(gè)chan, 寫入一個(gè)數(shù)據(jù)到管道中
	done := make(chan struct{}, 10)
 
	for i := 0; i < 10; i++ {
		go run(stop, done)
	}
 
	// 模擬超時(shí)時(shí)間
	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 做異步編程

利用無(wú)緩沖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 超時(shí)控制

超時(shí)控制還是建議使用 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的緩存大小來(lái)控制并發(fā)執(zhí)行的Goroutine的最大數(shù)目

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

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