Go并發(fā)編程詳解鎖、WaitGroup、Channel
在傳統(tǒng)的編程語言中,如C++、Java、Python等,其并發(fā)邏輯多建立在操作系統(tǒng)線程之上。線程間的通信通常依賴于操作系統(tǒng)提供的基礎原語,包括共享內存、信號、管道、消息隊列及套接字等,其中共享內存是最為普遍的通信方式。但這種基于共享內存的并發(fā)模型在復雜或大規(guī)模業(yè)務場景下往往顯得復雜且易于出錯。
Go語言在設計時即以解決傳統(tǒng)并發(fā)問題為目標,融入了CSP(Communicating Sequential Processes,通信順序進程)模型的理念。CSP模型致力于簡化并發(fā)編程,目標是讓編寫并發(fā)程序的難度與順序程序相當。
在CSP模型中,通信和同步通過一種特定的流程實現(xiàn):生產者產生數(shù)據(jù),然后通過輸出數(shù)據(jù)到輸入/輸出原語,最終到達消費者。Go語言為實現(xiàn)CSP模型,特別引入了Channel機制。Goroutine可以通過Channel進行數(shù)據(jù)的讀寫操作,Channel作為連接多個Goroutine的通信橋梁,簡化了并發(fā)編程的復雜性。
雖然CSP模型在Go語言中占據(jù)主流地位,但Go同樣支持基于共享內存的并發(fā)模型。在Go的sync
包中,提供了包括互斥鎖、讀寫鎖、條件變量和原子操作等在內的多種同步機制,以滿足不同并發(fā)場景下的需求。
互斥鎖(Mutex)
基本概念
互斥鎖(Mutex)是一種用于在并發(fā)環(huán)境中安全訪問共享資源的機制。當一個協(xié)程獲取到鎖時,它將擁有臨界區(qū)的訪問權,而其他請求該鎖的協(xié)程將會阻塞,直到該鎖被釋放。
應用場景
并發(fā)訪問共享資源的情形非常普遍,例如:
- 秒殺系統(tǒng)
- 多個goroutine并發(fā)修改某個變量
- 同時更新用戶信息
如果沒有互斥鎖的控制,將會導致商品超賣、變量數(shù)值不正確、用戶信息更新錯誤等問題。這時候就需要使用互斥鎖來控制并發(fā)訪問。
基本用法
Mutex實現(xiàn)了Locker接口,提供了兩個方法:Lock
和Unlock
。
- Lock方法用于對臨界區(qū)上鎖,獲得該鎖的協(xié)程擁有臨界資源的訪問權,其他請求臨界區(qū)的協(xié)程會阻塞等待該鎖的釋放。
- Unlock方法用于解鎖,釋放鎖使其他協(xié)程可以訪問臨界區(qū)。
package main
import (
"fmt"
"sync"
"time"
)
func main() {
var mu sync.Mutex
var count int
increment := func() {
mu.Lock()
defer mu.Unlock()
count++
fmt.Println("Count:", count)
}
for i := 0; i < 5; i++ {
go increment()
}
time.Sleep(time.Second)
}
易錯場景
- 不可重入的互斥鎖:Go的Mutex是不可重入鎖。由于Mutex鎖沒有記錄鎖的持有者信息,無法得知誰擁有鎖。如果一個獲取了鎖的協(xié)程再次請求鎖,將會被阻塞,形成死鎖。
func example() {
var mu sync.Mutex
mu.Lock()
defer mu.Unlock()
// Do something...
mu.Lock() // 死鎖
}
- Lock和Unlock不配對:未正確配對的Lock和Unlock調用會導致死鎖。如果對已經鎖定的鎖再次調用Lock,將會阻塞;對未鎖定的Mutex調用Unlock將會panic。
func example() {
var mu sync.Mutex
mu.Lock()
// 未調用mu.Unlock()
mu.Unlock() // 正確
}
- 復制已使用的鎖:復制已使用的鎖會導致意外的行為。
func example() {
var mu sync.Mutex
copyMu := mu
copyMu.Lock() // 錯誤
}
基本實現(xiàn)
Mutex結構體有兩個字段:state和sema。
type Mutex struct {
state int32
sema uint32
}
State字段的含義
Mutex有以下四種狀態(tài)。
- mutexLocked:Mutex上鎖標志
- mutexWoken:Mutex喚醒標志
- mutexStarving:Mutex正常/饑餓模式標志
- waiterCount:等待者數(shù)量
Mutex的正常模式和饑餓模式
- 正常模式:等待者隊列遵循先入先出原則,被喚醒的goroutine不會直接獲得鎖,而是與新請求鎖的goroutine競爭鎖。新請求鎖的goroutine由于正在CPU上執(zhí)行,獲得鎖的幾率更大,從而減少上下文切換的性能損失。然而,這可能導致被喚醒的goroutine長時間無法獲得鎖。
- 饑餓模式:當?shù)却龝r間超過閾值1毫秒時,進入饑餓模式。被喚醒的goroutine被放入等待隊列的隊首,當前goroutine在調用Unlock釋放鎖時,會直接將鎖交給等待隊列的隊首,新請求鎖的goroutine不會參與競爭,而是排到等待隊列的隊尾。當?shù)却犃袥]有goroutine或等待時間小于1毫秒時,Mutex將從饑餓模式切換回正常模式。
代碼示例
以下代碼展示了如何使用Mutex在并發(fā)環(huán)境中安全地訪問共享資源:
package main
import (
"fmt"
"sync"
"time"
)
func main() {
var mu sync.Mutex
var count int
increment := func() {
mu.Lock()
defer mu.Unlock()
count++
fmt.Println("Count:", count)
}
for i := 0; i < 5; i++ {
go increment()
}
time.Sleep(time.Second)
}
在上述代碼中,多個goroutine同時調用increment函數(shù),通過Mutex來確保對共享變量count的訪問是安全的。
讀寫鎖(RWMutex)
基本概念
在并發(fā)編程中,為了保證多個協(xié)程安全地訪問共享資源,我們通常使用Mutex互斥鎖。然而,在讀多寫少的場景下,Mutex會導致性能問題,因為所有操作(包括讀操作)都必須串行進行。為了解決這一問題,可以區(qū)分讀操作和寫操作。RWMutex是一種讀寫鎖,同一時間只能被一個寫操作持有,或者被多個讀操作持有。
基本用法
RWMutex提供了五個方法:Lock、Unlock、RLock、RUnlock和RLocker。
- Lock方法用于在寫操作時獲取寫鎖,會阻塞等待當前未釋放的寫鎖。當處于寫鎖狀態(tài)時,新的讀操作將會阻塞等待。
- Unlock方法用于釋放寫鎖。
- RLock方法用于在讀操作時獲取讀鎖,會阻塞等待當前寫鎖的釋放。如果鎖處于讀鎖狀態(tài),當前協(xié)程也能獲取讀鎖。
- RUnlock方法用于釋放讀鎖。
- RLocker方法用于獲取一個Locker接口的對象,調用其Lock方法時會調用RLock方法,調用Unlock方法時會調用RUnlock方法。
package main
import (
"fmt"
"sync"
"time"
)
func main() {
var rw sync.RWMutex
var count int
write := func() {
rw.Lock()
defer rw.Unlock()
count++
fmt.Println("Write:", count)
}
read := func() {
rw.RLock()
defer rw.RUnlock()
fmt.Println("Read:", count)
}
// Start multiple readers
for i := 0; i < 5; i++ {
go read()
}
// Start a single writer
go write()
time.Sleep(time.Second)
}
實現(xiàn)原理
RWMutex主要通過readerCount字段來維護讀鎖的數(shù)量。寫操作時,會將readerCount減去2的30次方變成一個負數(shù),從而阻塞新的讀鎖請求。當寫鎖被釋放時,將readerCount加上2的30次方,恢復成一個整數(shù)并喚醒等待中的讀鎖操作。
易錯場景
RWMutex的易錯場景和Mutex類似,包括以下幾點。
- 不可重入鎖:Go的RWMutex是不可重入鎖。如果一個獲取了鎖的協(xié)程再次請求同一個鎖,將會被阻塞,形成死鎖。
func example() {
var rw sync.RWMutex
rw.Lock()
defer rw.Unlock()
// Do something...
rw.Lock() // 死鎖
}
- Lock和Unlock不配對:未正確配對的Lock和Unlock調用會導致死鎖。如果對已經鎖定的鎖再次調用Lock,將會阻塞;對未鎖定的RWMutex調用Unlock將會panic。
func example() {
var rw sync.RWMutex
rw.Lock()
// 未調用rw.Unlock()
rw.Unlock() // 正確
}
- 復制已使用的鎖:復制已使用的鎖會導致意外行為。
func example() {
var rw sync.RWMutex
copyRw := rw
copyRw.Lock() // 錯誤
}
- 隱蔽的死鎖情景:寫鎖操作等待舊的讀鎖的釋放,舊的讀鎖等待新的讀鎖的釋放,新的讀鎖等待寫鎖的釋放,形成死鎖。
package main
import (
"fmt"
"sync"
"time"
)
func main() {
var rw sync.RWMutex
var count int
write := func() {
rw.Lock()
defer rw.Unlock()
count++
fmt.Println("Write:", count)
}
read := func() {
rw.RLock()
defer rw.RUnlock()
fmt.Println("Read:", count)
}
// 啟動多個讀操作
for i := 0; i < 5; i++ {
go read()
}
// 啟動寫操作
go write()
time.Sleep(time.Second)
}
在上述代碼中,多個goroutine同時調用read函數(shù),通過RWMutex來確保對共享變量count的讀取是安全的。同時,write函數(shù)用于更新共享變量count,確保在寫操作時獨占訪問權。
死鎖
什么是死鎖
死鎖指的是一組進程由于相互持有和等待資源,導致無法繼續(xù)執(zhí)行的狀態(tài)。在這種情況下,所有相關的進程都會無限期阻塞,無法向前推進。具體來說,死鎖發(fā)生在一個進程持有某些資源并等待其他進程釋放其占有的資源,同時這些其他進程也在等待第一個進程釋放資源,形成相互等待的狀態(tài)。
死鎖的必要條件
死鎖的發(fā)生需要滿足以下四個必要條件。
- 互斥條件:資源同一時間只能被一個進程所擁有。
- 請求和保持條件:一個進程已經擁有某些資源,但在等待其他資源時不釋放已持有的資源。
- 不可剝奪條件:進程持有的資源在未使用完畢前,不能被強行剝奪,只能由進程自己釋放。
- 循環(huán)等待條件:存在一個進程集合中的每個進程都在等待另一個進程所持有的資源,形成一個循環(huán)等待鏈。
如何解決死鎖問題
為了解決死鎖問題,可以采取以下兩種策略。
- 檢測和恢復:系統(tǒng)可以定期檢測死鎖的存在,并采取措施恢復。例如,通過回滾進程的一部分操作或強制剝奪資源。
- 破壞死鎖的必要條件是,可以通過設計系統(tǒng)來破壞死鎖的四個必要條件之一,例如:
- 破壞互斥條件:盡量使用共享資源來減少互斥性。
- 破壞請求和保持條件:在進程開始時一次性請求所有資源,或者在請求新的資源之前釋放已持有的資源。
- 破壞不可剝奪條件:設計成可以強制剝奪資源,如通過優(yōu)先級調度。
- 破壞循環(huán)等待條件:對資源進行排序,并要求進程按序請求資源,避免形成循環(huán)等待。
示例代碼
以下是一個Go語言中的死鎖示例,展示了兩個goroutine由于相互等待對方持有的資源而導致的死鎖:
package main
import (
"fmt"
"sync"
)
func main() {
var mutexA, mutexB sync.Mutex
go func() {
mutexA.Lock()
fmt.Println("Goroutine 1: Locked mutexA")
// Simulate some work
mutexB.Lock()
fmt.Println("Goroutine 1: Locked mutexB")
mutexB.Unlock()
mutexA.Unlock()
}()
go func() {
mutexB.Lock()
fmt.Println("Goroutine 2: Locked mutexB")
// Simulate some work
mutexA.Lock()
fmt.Println("Goroutine 2: Locked mutexA")
mutexA.Unlock()
mutexB.Unlock()
}()
// Wait for goroutines to finish (they won't due to deadlock)
select {}
}
在上述代碼中,兩個goroutine分別持有mutexA
和mutexB
,并且嘗試獲取對方的鎖,導致死鎖發(fā)生。每個goroutine無限期等待對方釋放資源,形成相互等待的循環(huán)。
通過了解死鎖的概念、必要條件及解決策略,我們可以更好地設計并發(fā)程序,避免陷入死鎖狀態(tài)。
WaitGroup
基本概念
WaitGroup 是 Go 語言的 sync 包下提供的一種并發(fā)原語,用來解決并發(fā)編排的問題。它主要用于等待一組 goroutine 完成。假設一個大任務需要等待三個小任務完成才能繼續(xù)執(zhí)行,如果采用輪詢的方法,可能會導致兩個問題:一是小任務已經完成但大任務需要很久才能被輪詢到,二是輪詢會造成 CPU 資源的浪費。因此,WaitGroup 通過阻塞等待并喚醒大任務的 goroutine 來解決這個問題。
基本用法
WaitGroup 提供了三個方法:Add、Done 和 Wait。
- Add(delta int):將計數(shù)器增加 delta 值。
- Done():將計數(shù)器的值減一,相當于 Add(-1)。
- Wait():阻塞等待,直到計數(shù)器的值變?yōu)?0,然后喚醒調用者。
實現(xiàn)原理
WaitGroup 維護了兩個計數(shù)器,一個是 v 計數(shù)器,另一個是 w 計數(shù)器。
- 調用 Add 方法時,v 計數(shù)器的值會增加相應的 delta 值。
- 調用 Done 方法時,v 計數(shù)器的值會減一。
- 調用 Wait 方法時,w 計數(shù)器的值會加一。當 v 計數(shù)器的值為 0 時,會喚醒所有的 waiter。
易錯場景
使用 WaitGroup 需要注意以下易錯場景:
- 計數(shù)器的值為負數(shù)會引發(fā) panic;
- v 計數(shù)器增加的值大于減少的值,會造成一直阻塞。
示例代碼
以下是一個使用 WaitGroup 的示例代碼:
package main
import (
"fmt"
"sync"
"time"
)
func worker(id int, wg *sync.WaitGroup) {
defer wg.Done() // Done() 方法用于減少計數(shù)器
fmt.Printf("Worker %d starting\n", id)
time.Sleep(time.Second)
fmt.Printf("Worker %d done\n", id)
}
func main() {
var wg sync.WaitGroup
for i := 1; i <= 3; i++ {
wg.Add(1) // Add() 方法增加計數(shù)器
go worker(i, &wg)
}
wg.Wait() // Wait() 方法阻塞等待所有計數(shù)器為 0
fmt.Println("All workers done")
}
在上述代碼中,main 函數(shù)創(chuàng)建了一個 WaitGroup 并啟動了三個 goroutine,每個 goroutine 執(zhí)行 worker 函數(shù)。在 worker 函數(shù)中,調用 wg.Done() 方法表示當前工作已經完成。main 函數(shù)中的 wg.Wait() 方法阻塞等待,直到所有的 goroutine 都完成工作并調用了 Done 方法。
小結
WaitGroup 是 Go 語言中非常有用的并發(fā)原語,用于等待一組 goroutine 完成。通過合理使用 Add、Done 和 Wait 方法,可以避免輪詢等待帶來的性能問題,并提高并發(fā)編排的效率。在使用 WaitGroup 時,需要注意計數(shù)器的增減操作,避免引發(fā) panic 或長時間阻塞。
Channel
基本概念
Go 語言提倡通過通信來實現(xiàn)共享內存,而不是通過共享內存來通信。Go 的 CSP(Communicating Sequential Processes)并發(fā)模型正是通過 Goroutine 和 Channel 來實現(xiàn)的。Channel 是 Go 語言中用于 goroutine 之間通信的主要工具。
應用場景
Channel 有以下幾類應用場景。
- 數(shù)據(jù)交互:通過 Channel 可以模擬并發(fā)的 Buffer 或者 Queue,實現(xiàn)生產者-消費者模式。
- 數(shù)據(jù)傳遞:通過 Channel 將數(shù)據(jù)傳遞給其他的 goroutine 進行處理。
- 信號通知:Channel 可以用于傳遞一些信號,如 close、data ready 等。
- 并發(fā)編排:通過 Channel 的阻塞等待機制,可以讓一組 goroutine 按照一定的順序并發(fā)或串行執(zhí)行。
- 實現(xiàn)鎖功能:通過 Channel 的阻塞等待機制,可以實現(xiàn)互斥鎖的功能。
基本用法
Channel 有三種類型:
- 只能接收的 Channel:<-chan T。
- 只能發(fā)送的 Channel:chan<- T。
- 既能發(fā)送又能接收的 Channel:chan T。
Channel 通過 make 函數(shù)進行初始化,未初始化的 Channel 的零值是 nil,對 nil 的 Channel 進行接收或發(fā)送操作會導致阻塞。
Channel 可以分為有緩沖和無緩沖兩種。無緩沖的 Channel 是同步的,有緩沖的 Channel 是異步的。發(fā)送操作只有在 Channel 滿時才會阻塞,接收操作只有在 Channel 為空時才會阻塞。
發(fā)送操作是 chan<-,接收操作是 <-chan。接收數(shù)據(jù)時可以返回兩個值,第一個是元素,第二個是一個布爾值,若為 false 則說明 Channel 已經被關閉并且 Channel 中沒有緩存的數(shù)據(jù)。
Go 的內建函數(shù) close、cap、len 都可以操作 Channel 類型,發(fā)送和接收都可以作為 select 語句的 case,Channel 也可以應用于 for range 語句。
實現(xiàn)原理
發(fā)送
在發(fā)送數(shù)據(jù)給 Channel 時,發(fā)送語句會轉化為 chansend 函數(shù):
- 如果 Channel 是 nil,調用者會被阻塞;
- 如果 Channel 已經關閉,發(fā)送操作會導致 panic;
- 如果 recvq 字段有 receiver,則將數(shù)據(jù)交給它,而不需要放入 buffer 中;
- 如果沒有 receiver,則將數(shù)據(jù)放入 buffer 中;
- 如果 buffer 滿了,則發(fā)送者 goroutine 會加入到 sendq 中阻塞休眠,直到被喚醒。
接收
在接收數(shù)據(jù)時,接收語句會轉化為 chanrecv 函數(shù):
- 如果 Channel 是 nil,調用者會被阻塞;
- 如果 Channel 已經被關閉,并且隊列中無緩存元素,則返回 false 和一個對應元素的零值;
- 如果 sendq 中有 sender 并且 buffer 中有數(shù)據(jù),則優(yōu)先從 buffer 中取出,否則從 sendq 中彈出一個 sender,把它的數(shù)據(jù)復制給 receiver;
- 如果沒有 sender,則從 buffer 中正常取一個元素;如果沒有元素,則 receiver 會加入到 recvq 中阻塞等待,直到接收到數(shù)據(jù)或者 Channel 被關閉。
關閉
- 如果 Channel 是 nil,關閉 nil 的 Channel 會導致 panic。
- 如果關閉已經關閉的 Channel 也會導致 panic。
- 否則將 recvq 和 sendq 全部清除并喚醒。
示例代碼
以下是一個使用 Channel 的示例代碼:
package main
import (
"fmt"
"time"
)
// 生產者:生成數(shù)據(jù)并發(fā)送到 channel
func producer(ch chan<- int, count int) {
for i := 0; i < count; i++ {
ch <- i
fmt.Println("Produced:", i)
time.Sleep(time.Millisecond * 500)
}
close(ch) // 關閉 channel,表示生產結束
}
// 消費者:從 channel 接收數(shù)據(jù)并處理
func consumer(ch <-chan int) {
for data := range ch {
fmt.Println("Consumed:", data)
time.Sleep(time.Millisecond * 1000)
}
}
func main() {
ch := make(chan int, 5) // 創(chuàng)建一個帶緩沖的 channel
go producer(ch, 10) // 啟動生產者
consumer(ch) // 啟動消費者
}
在上述代碼中,main 函數(shù)創(chuàng)建了一個帶緩沖的 Channel,并啟動了一個生產者 goroutine 和一個消費者 goroutine。生產者不斷生成數(shù)據(jù)并發(fā)送到 Channel 中,消費者從 Channel 中接收數(shù)據(jù)并進行處理。生產者完成后關閉 Channel,消費者則在接收到所有數(shù)據(jù)后結束。