Go語言的并發(fā)模型是CSP（通信順序進(jìn)程），提倡通過通信來進(jìn)行內(nèi)存共享，而不是通過共享內(nèi)存來實(shí)現(xiàn)通信。

控制并發(fā)有三種經(jīng)典的方式，使用 channel 通知實(shí)現(xiàn)并發(fā)控制、使用 sync 包中的 WaitGroup 實(shí)現(xiàn)并發(fā)控制、使用 Context 上下文實(shí)現(xiàn)并發(fā)控制。

一、使用 channel 通知實(shí)現(xiàn)并發(fā)控制

1、無緩沖通道

無緩沖通道，又叫做阻塞通道。發(fā)送方 (goroutine) 和接收方 (gouroutine) 必須是同步的，同時(shí)準(zhǔn)備好，如果沒有同時(shí)準(zhǔn)備好的話，一方就會(huì)一直阻塞住，直到另一方準(zhǔn)備好為止。

使用無緩沖通道進(jìn)行通信，將發(fā)送和接收的 goroutine 同步化，因此，無緩沖通道也被稱為同步通道。

ch := make(chan int) // 創(chuàng)建無緩沖通道

使用無緩沖通道實(shí)現(xiàn)并發(fā)控制：

package main

import "fmt"

func recv(c chan int) {
	fmt.Println("開始接收")
	ret := <-c
	fmt.Println("接收成功", ret)
}

func main() {
	ch := make(chan int)
	go recv(ch) // 啟用 goroutine 從通道接收值
	ch <- 10
	fmt.Println("發(fā)送成功")
}

當(dāng)子協(xié)程從無緩沖 channel 里接收值時(shí)，沒有發(fā)送方，子協(xié)程阻塞等待，直到主協(xié)程往無緩沖 channel 里發(fā)送值，子協(xié)程開始執(zhí)行，然后主協(xié)程開始執(zhí)行。

2、有緩沖通道

只要通道的容量大于0，那么該通道就是有緩沖通道，通道的容量表示通道中能最多存放元素的數(shù)量。發(fā)送方在緩沖區(qū)滿的時(shí)候阻塞，接收方不阻塞；接收方在緩沖區(qū)為空的時(shí)候阻塞，發(fā)送方不阻塞。

ch := make(chan int, 10) // 創(chuàng)建一個(gè)緩沖區(qū)為10的有緩沖通道
fmt.Println(len(ch)) // 通過len函數(shù)獲取當(dāng)前通道內(nèi)元素?cái)?shù)量
fmt.Println(cap(ch)) // 通過cap函數(shù)獲取通道的容量

使用緩沖區(qū)為 1 的通道實(shí)現(xiàn)并發(fā)控制：

package main

import (
	"fmt"
	"time"
)

func recv(c chan int) {
	fmt.Println("開始接收")
	ret := <-c
	fmt.Println("接收成功", ret)
}

func main() {
	ch := make(chan int, 1)
	ch <- 10
	go recv(ch) // 啟用 goroutine 從通道接收值
	time.Sleep(time.Second)
	fmt.Println("發(fā)送成功")
}

當(dāng)主協(xié)程往緩沖區(qū)為1的 channel 里發(fā)送值時(shí)，不阻塞，子協(xié)程啟動(dòng)，從無緩沖 channel 里接收值，主協(xié)程睡眠1秒，等待子協(xié)程執(zhí)行完，主協(xié)程在執(zhí)行。

二、使用 sync 包中的 WaitGroup 實(shí)現(xiàn)并發(fā)控制

1、sync.WaitGroup

在 sync 包中提供了 WaitGroup ，它會(huì)等待它收集的所有 goroutine 任務(wù)全部完成。

在主協(xié)程中調(diào)用 Add() 添加需要執(zhí)行 goroutine 的數(shù)量，在每一個(gè) goroutine 執(zhí)行完成后調(diào)用 Done() ，表示這個(gè) goroutine 已經(jīng)完成，主協(xié)程調(diào)用 Wait() 阻塞等待所有 goroutine 執(zhí)行完成，當(dāng)所有的 goroutine 都執(zhí)行完成后，主協(xié)程返回。

實(shí)現(xiàn)原理：sync.WaitGroup 內(nèi)部維護(hù)著一個(gè)計(jì)數(shù)器，計(jì)數(shù)器的值可以增加和減少，當(dāng)我們啟動(dòng) N 個(gè)并發(fā)任務(wù)時(shí)，將計(jì)數(shù)器增加 N，每個(gè)任務(wù)通過調(diào)用Done方法將計(jì)數(shù)器減1，通過調(diào)用Wait()來等待并發(fā)任務(wù)執(zhí)行完，當(dāng)計(jì)數(shù)器的值為 0 時(shí)，表示所有并發(fā)任務(wù)都已經(jīng)完成。

sync.WaitGroup有以下三種方法：

• Add(N int) : 計(jì)數(shù)器 + N
• Done() : 計(jì)數(shù)器 - 1
• Wait() : 阻塞，直到計(jì)數(shù)器變?yōu)?

package main

import (
	"fmt"
	"sync"
)

var wg sync.WaitGroup

func hello() {
	defer wg.Done()
	fmt.Println("Hello Goroutine!")
}
func main() {
	wg.Add(3)
	go hello() // 啟動(dòng)3個(gè)goroutine去執(zhí)行hello函數(shù)
	go hello()
	go hello()
	fmt.Println("main goroutine done!")
	wg.Wait()
}

擴(kuò)展：

在Golang官網(wǎng)中，有這么一句話：

A WaitGroup must not be copied after first use.

意思是，在 WaitGroup 第一次使用后，不能被拷貝。

為什么呢？？？

通過下面的例子我們淺淺分析一下。

package main

import (
	"fmt"
	"sync"
)

func main() {
	wg := sync.WaitGroup{}
	for i := 0; i < 5; i++ {
		wg.Add(1)
		go func(wg sync.WaitGroup, i int) {
			fmt.Println(i)
			wg.Done()
		}(wg, i)
	}
	wg.Wait()
}

提示所有的 goroutine 都已經(jīng)睡眠了，出現(xiàn)了死鎖。這是因?yàn)?wg 值拷貝傳遞到了子 goroutine 中，導(dǎo)致只有 Add 操作，Done 操作是在 wg 的副本執(zhí)行的， wg 的作用域?yàn)樽訁f(xié)程，而不是全局，因此主協(xié)程就死鎖了。

改正方法：

• 指針，將匿名函數(shù)中 wg 的傳入類型改為 *sync.WaitGroup 。

func main() {
	wg := sync.WaitGroup{}
	for i := 0; i < 5; i++ {
		wg.Add(1)
		go func(wg *sync.WaitGroup, i int) {
			fmt.Println(i)
			wg.Done()
		}(&wg, i)
	}
	wg.Wait()
}

• 閉包，將匿名函數(shù)中的 wg 的傳入?yún)?shù)去掉，在匿名函數(shù)中可以直接使用外面的 wg 變量。

func main() {
	wg := sync.WaitGroup{}
	for i := 0; i < 5; i++ {
		wg.Add(1)
		go func(i int) {
			fmt.Println(i)
			wg.Done()
		}(i)
	}
	wg.Wait()
}

2、sync.Once

很多場景下，我們需要確保某些操作在高并發(fā)時(shí)只執(zhí)行一次，例如只加載一次配置文件。Go語言中的sync包提供了一個(gè)針對(duì)只執(zhí)行一次場景的解決方案 sync.Once。

sync.Once只有一個(gè)Do方法，Do(f func()) 。

實(shí)現(xiàn)原理：sync.Once 內(nèi)部包含一個(gè)互斥鎖和一個(gè)布爾值，互斥鎖保證布爾值和數(shù)據(jù)的安全，而布爾值用來記錄初始化是否完成，這樣設(shè)計(jì)就能保證初始化操作的時(shí)候是并發(fā)安全的，并且初始化操作也不會(huì)執(zhí)行多次。

樣例如下：延遲一個(gè)開銷很大的初始化操作到真正用到它的時(shí)候再執(zhí)行。

package main

import (
	"image"
	"sync"
)

var icons map[string]image.Image

var loadIconsOnce sync.Once

func loadIcons() {
	// 加載圖片
	icons = map[string]image.Image{}
}

// Icon 是并發(fā)安全的
func Icon(name string) image.Image {
	loadIconsOnce.Do(loadIcons)
	return icons[name]
}

三、使用 Context 上下文實(shí)現(xiàn)并發(fā)控制

1、簡介

在一些簡單場景下使用 channel 和 WaitGroup 已經(jīng)足夠了，但是當(dāng)面臨一些復(fù)雜多變的網(wǎng)絡(luò)并發(fā)場景下 channel 和 WaitGroup 顯得有些力不從心了。在并發(fā)程序中，由于超時(shí)、取消操作或者一些異常情況，往往需要進(jìn)行搶占操作或者中斷后續(xù)操作。

舉個(gè)例子：在 Go http包的Server中，每一個(gè)請(qǐng)求在都有一個(gè)對(duì)應(yīng)的 goroutine 去處理。請(qǐng)求處理函數(shù)通常會(huì)啟動(dòng)額外的 goroutine 用來訪問后端服務(wù)，比如數(shù)據(jù)庫和RPC服務(wù)，用來處理一個(gè)請(qǐng)求的 goroutine 通常需要訪問一些與請(qǐng)求特定的數(shù)據(jù)，比如終端用戶的身份認(rèn)證信息、驗(yàn)證相關(guān)的token、請(qǐng)求的截止時(shí)間。 當(dāng)一個(gè)請(qǐng)求被取消或超時(shí)時(shí)，所有用來處理該請(qǐng)求的 goroutine 都應(yīng)該迅速中斷退出，然后系統(tǒng)才能釋放這些 goroutine 占用的資源。

所以我們需要一種可以跟蹤 goroutine 的方案，才可以達(dá)到控制他們的目的，這就是Go語言為我們提供的 Context，稱之為上下文非常貼切，它就是 goroutine 的上下文。它包括一個(gè)程序的運(yùn)行環(huán)境、現(xiàn)場和快照等。每個(gè)程序要運(yùn)行時(shí)，都需要知道當(dāng)前程序的運(yùn)行狀態(tài)，通常Go 將這些封裝在一個(gè) Context 里，再將它傳給要執(zhí)行的 goroutine 。context 包主要是用來處理多個(gè) goroutine 之間共享數(shù)據(jù)，及多個(gè) goroutine 的管理。

context常用的使用場景：

• 一個(gè)請(qǐng)求對(duì)應(yīng)多個(gè)goroutine之間的數(shù)據(jù)交互
• 超時(shí)控制
• 上下文控制

2、context 包

context 包的核心是 struct Context，接口聲明如下：

type Context interface {
	// 返回Context的超時(shí)時(shí)間（超時(shí)返回場景）
	Deadline() (deadline time.Time, ok bool)
	// 在Context超時(shí)或取消時(shí)（即結(jié)束了）返回一個(gè)關(guān)閉的channel，取消信號(hào)
    // 如果當(dāng)前Context超時(shí)或取消時(shí)，Done方法會(huì)返回一個(gè)channel，然后其他地方就可以通過判斷Done方法是否有返回（channel），如果有則說明Context已結(jié)束
    // 故其可以作為廣播通知其他相關(guān)方本Context已結(jié)束，請(qǐng)做相關(guān)處理。
	Done() <-chan struct{}
	// 返回Context取消的原因
	Err() error
	// 返回Context相關(guān)數(shù)據(jù)
	Value(key any) any
}

Context 對(duì)象是線程安全的（底層數(shù)據(jù)結(jié)構(gòu)加了互斥鎖），你可以把一個(gè) Context 對(duì)象傳遞給任意個(gè)數(shù)的 gorotuine，對(duì)它執(zhí)行取消操作時(shí)，所有 goroutine 都會(huì)接收到取消信號(hào)。

一個(gè) Context 不能擁有 Cancel 方法，同時(shí)我們也只能 Done channel 接收數(shù)據(jù)。原因是：接收取消信號(hào)的函數(shù)和發(fā)送信號(hào)的函數(shù)通常不是一個(gè)。一個(gè)典型的場景是：父操作為子操作操作啟動(dòng) goroutine，子操作也就不能取消父操作。

3、繼承 context

context 包提供了一些函數(shù)，協(xié)助用戶從現(xiàn)有的 Context 對(duì)象創(chuàng)建新的 Context 對(duì)象。這些 Context 對(duì)象形成一棵樹：當(dāng)一個(gè) Context 對(duì)象被取消時(shí)，繼承自它的所有 Context 都會(huì)被取消。

Background 是所有 Context 對(duì)象樹的根，它不能被取消。context 包提供了三種context，分別是普通context、超時(shí)context、帶值的context：

func Background() Context {
	return backgroundCtx{}
}

// 普通context，通常這樣調(diào)用： 
ctx, cancel := context.WithCancel(context.Background())
func WithCancel(parent Context) (ctx Context, cancel CancelFunc)

// 帶超時(shí)的context，超時(shí)之后會(huì)自動(dòng)close對(duì)象的Done，與調(diào)用CancelFunc的效果一樣
// WithDeadline 明確地設(shè)置一個(gè)d指定的系統(tǒng)時(shí)鐘時(shí)間，如果超過就觸發(fā)超時(shí)
// WithTimeout 設(shè)置一個(gè)相對(duì)的超時(shí)時(shí)間，也就是deadline設(shè)為timeout加上當(dāng)前的系統(tǒng)時(shí)間
// 因?yàn)閮烧呤聦?shí)上都依賴于系統(tǒng)時(shí)鐘，所以可能存在微小的誤差，所以官方不推薦把超時(shí)間隔設(shè)置得太小
// 通常這樣調(diào)用：
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
func WithDeadline(parent Context, d time.Time) (Context, CancelFunc)
func WithTimeout(parent Context, timeout time.Duration) (Context, CancelFunc)

// 帶有值的context，沒有CancelFunc，所以它只用于值的多goroutine傳遞和共享
// 通常這樣調(diào)用：
ctx := context.WithValue(context.Background(), "key", myValue)
func WithValue(parent Context, key, val interface{}) Context

WithCancel 和 WithTimeout 函數(shù)會(huì)返回繼承的 Context 對(duì)象， 這些對(duì)象可以比它們的父 Context 更早地取消。當(dāng)請(qǐng)求處理函數(shù)返回時(shí)，與該請(qǐng)求關(guān)聯(lián)的 Context 會(huì)被取消。當(dāng)使用多個(gè)副本發(fā)送請(qǐng)求時(shí)，可以使用 WithCancel 取消多余的請(qǐng)求。 WithTimeout 在設(shè)置對(duì)后端服務(wù)器請(qǐng)求超時(shí)時(shí)間時(shí)非常有用。WithValue 函數(shù)能夠?qū)⒄?qǐng)求作用域的數(shù)據(jù)與 Context 對(duì)象建立關(guān)系。

4、context 例子

下面的例子，主要描述的是通過一個(gè) channel 實(shí)現(xiàn)一個(gè)為循環(huán)次數(shù)為5的循環(huán)。

package main

import (
	"context"
	"fmt"
	"time"
)

func childFunc(cont context.Context, num *int) {
	ctx, _ := context.WithCancel(cont)
	for {
		select {
		case <-ctx.Done():
			fmt.Println("child Done : ", ctx.Err())
			return
		}
	}
}

func main() {
	gen := func(ctx context.Context) <-chan int {
		dst := make(chan int)
		n := 1
		go func() {
			for {
				select {
				case <-ctx.Done():
					fmt.Println("parent Done : ", ctx.Err())
					return // returning not to leak the goroutine
				case dst <- n:
					n++
					go childFunc(ctx, &n)
				}
			}
		}()
		return dst
	}

	ctx, cancel := context.WithCancel(context.Background())
	for n := range gen(ctx) {
		fmt.Println(n)
		if n >= 5 {
			break
		}
	}
	cancel()
	time.Sleep(5 * time.Second)
}

在每一個(gè)循環(huán)中產(chǎn)生一個(gè)goroutine，每一個(gè)goroutine中都傳入context，在每個(gè)goroutine中通過傳入 ctx 創(chuàng)建一個(gè)子Context，并且通過 select 一直監(jiān)控該Context的運(yùn)行情況，當(dāng)父 Context 退出的時(shí)候，代碼中并沒有明顯調(diào)用子 Context 的 Cancel 函數(shù)，但是分析結(jié)果，子 Context 還是被正確合理的關(guān)閉了，這是因?yàn)?，所有基于這個(gè) Context 或者衍生的子 Context 都會(huì)收到通知，這時(shí)就可以進(jìn)行清理操作了，最終釋放 goroutine，這就優(yōu)雅的解決了 goroutine 啟動(dòng)后不可控的問題。

5、context 使用原則

• 不要把 context 放在結(jié)構(gòu)體中，要以參數(shù)的方式傳遞。
• 以 context 作為參數(shù)的函數(shù)方法，應(yīng)該把 context 作為第一個(gè)參數(shù)，放在第一位。
• 給一個(gè)函數(shù)方法傳遞 context 的時(shí)候，不要傳遞nil，如果不知道傳遞什么，就使用context.TODO。
• context 的 Value 相關(guān)方法應(yīng)該傳遞必須的數(shù)據(jù)，不要什么數(shù)據(jù)都使用這個(gè)傳遞。
• context 是線程安全的，底層數(shù)據(jù)結(jié)構(gòu)加了互斥鎖，可以放心的在多個(gè)goroutine中傳遞。

到此這篇關(guān)于Golang 并發(fā)控制模型的實(shí)現(xiàn)的文章就介紹到這了,更多相關(guān)Golang 并發(fā)控制內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！ 

最新評(píng)論