快捷導(dǎo)航

詳解Go?sync?同步原語

更新時(shí)間：2023年12月27日 12:04:21 作者：Schuyler_yuan

Go?中不僅有?channel?這種?CSP?同步機(jī)制,還有?sync.Mutex、sync.WaitGroup?等比較原始的同步原語,使用它們,可以更靈活的控制數(shù)據(jù)同步和多協(xié)程并發(fā),這篇文章主要介紹了Go?sync?同步原語,需要的朋友可以參考下

Go 中不僅有 channel 這種 CSP 同步機(jī)制，還有 sync.Mutex、sync.WaitGroup 等比較原始的同步原語。使用它們，可以更靈活的控制數(shù)據(jù)同步和多協(xié)程并發(fā)。

sync.Mutex
sync.RWMutex
sync.WaitGroup
sync.Once
sync.Cond
sync.Map

在一個(gè) goroutine 中，如果分配的內(nèi)存沒有被其他 goroutine 訪問，只在該 goroutine 中被使用，不存在資源競爭的問題。但如果同一塊內(nèi)存被多個(gè) goroutine 同時(shí)訪問，就會(huì)不知道誰先訪問，也無法預(yù)料最后結(jié)果。這就產(chǎn)生了資源競爭，這塊內(nèi)存就是共享資源。channel 是并發(fā)安全的，內(nèi)部自加了鎖，但是很多變量或者資源沒有加鎖，就需要 sync 同步原語了。

eg. 啟動(dòng)100個(gè)協(xié)程，讓 nSum 加10，期待的結(jié)果是1000。

package main
import (
        "fmt"
        "time"
)
var nSum = 0
func add(i int) {
        nSum += i
}
func main() {
        for i := 0; i < 100; i++ {
                go add(10)
        }
        time.Sleep(time.Second)
        fmt.Println("sum=", nSum)
}

運(yùn)行完之后，輸出的結(jié)果可能是1000，也可能是990，或是980。

$ while true; do go run gosrc.go; done;

類似 go build、go run、go test，這種 Go 工具鏈命令，添加 -race 標(biāo)識(shí)，幫助檢查 Go 語言代碼是否存在資源競爭。

$ go run -race gosrc.go

導(dǎo)致這種現(xiàn)象的原因是，資源 nSum 并不是并發(fā)安全的，因?yàn)橥瑫r(shí)會(huì)有多個(gè)協(xié)程執(zhí)行 nSum += i，產(chǎn)生不可預(yù)料的結(jié)果。所以需要確保同時(shí)只有一個(gè)協(xié)程執(zhí)行 nSum += i 操作，互斥鎖可以實(shí)現(xiàn)。

sync.Mutex

互斥鎖，是指在同一時(shí)刻只有一個(gè)協(xié)程執(zhí)行某段代碼，其他協(xié)程都要等待該協(xié)程執(zhí)行完畢后才能繼續(xù)執(zhí)行。

下面的實(shí)例中，聲明一個(gè)互斥鎖，然后修改 add 函數(shù)，對 nSum += i 執(zhí)行加鎖保護(hù)，這樣這段代碼在并發(fā)的時(shí)候就安全了，可以得到正確的結(jié)果。

上面這段加鎖保護(hù)的代碼，稱為臨界區(qū)。在同步程序設(shè)計(jì)中，臨界區(qū)指的是一個(gè)訪問共享資源的程序片段，而這些共享資源又無法同時(shí)被多個(gè)協(xié)程訪問的特性。當(dāng)一個(gè)協(xié)程獲得了鎖后，其他的協(xié)程只有等待鎖釋放，才能再去獲得鎖。鎖的 Lock 和 Unlock 方法總是成對的出現(xiàn)。

package main
import (
        "fmt"
        "sync"
        "time"
)
var (
        nSum int
        mutex sync.Mutex
)
func add(i int) {
        mutex.Lock()
        defer mutex.Unlock()
        nSum += i
}
func main() {
        for i := 0; i < 100; i++ {
                go add(10)
        }
        time.Sleep(2*time.Second)
        fmt.Println("nSum=", nSum)
}

運(yùn)行結(jié)果如下，

$ count=0;while (($count < 10)); do go run gomutex.go;((count=$count+1)); done

sync.RWMutex

互斥鎖是完全互斥的，但是有很多實(shí)際的場景下是讀多寫少的，當(dāng)我們并發(fā)的讀取一個(gè)資源不涉及資源修改的時(shí)候是沒有必要加鎖的，這種場景下使用讀寫鎖是更好的一種選擇。讀寫鎖在 Go 語言中使用 sync.RWMutex 類型。

讀寫鎖分為兩種：讀鎖和寫鎖。當(dāng)一個(gè) goroutine 獲取讀鎖之后，其他 goroutine 如果是獲取讀鎖會(huì)繼續(xù)獲得鎖，如果是獲取寫鎖就會(huì)等待；當(dāng)一個(gè) goroutine 獲取寫鎖之后，其他 goroutine 無論是獲取讀鎖還是寫鎖都會(huì)等待。

這里有一個(gè)性能問題，每次讀寫共享資源都要加鎖，性能低下，怎么解決？現(xiàn)在分析這個(gè)特殊的場景，會(huì)有以下三種情況，寫的時(shí)候不能同時(shí)讀（讀未提交），讀的時(shí)候不能同時(shí)寫（讀已提交），讀的時(shí)候可以同時(shí)讀（可重復(fù)讀）。

可能讀到臟數(shù)據(jù)，臟讀
會(huì)產(chǎn)生不可預(yù)料的結(jié)果，幻讀
不管多少協(xié)程讀，都是并發(fā)安全的，可重復(fù)讀。

可以通過讀寫鎖提升性能，對比互斥鎖，讀寫鎖改動(dòng)有兩個(gè)地方，

把鎖的聲明換成讀寫鎖 RWMutex
把讀取數(shù)據(jù)的代碼（函數(shù) readSum）換成讀鎖

這樣性能有很大提升，多個(gè)協(xié)程可以同時(shí)讀取數(shù)據(jù)，不用相互等待。

sync.WaitGroup

用于最終完成的場景，關(guān)鍵點(diǎn)在于一定是等待所有協(xié)程都執(zhí)行完畢。

在前面的程序里邊，為了防止主函數(shù)返回，使用了 time.Sleep 語句強(qiáng)制程序睡眠，因?yàn)橐坏?main goroutine 返回，函數(shù)就退出了。

但這里是有問題的。如果這100個(gè)協(xié)程在兩秒內(nèi)執(zhí)行完畢，main 函數(shù)本該提前返回，但是還是要等夠兩秒才能返回，存在性能問題。如果執(zhí)行超過2秒，函數(shù)返回，有些協(xié)程不會(huì)執(zhí)行，產(chǎn)生不可預(yù)知的結(jié)果。

有沒有辦法監(jiān)聽所有 goroutine 的執(zhí)行？一旦全部執(zhí)行完畢，程序馬上退出，既可以保證所有協(xié)程執(zhí)行完畢，又可以及時(shí)退出節(jié)省時(shí)間，提升性能。

通道 channel 可以實(shí)現(xiàn)，但比較復(fù)雜。所以，Go 提供了 WaitGroup。對上面的例子代碼進(jìn)行改造，分三步執(zhí)行，

聲明一個(gè) WaitGroup，通過 Add 方法設(shè)置一個(gè)計(jì)數(shù)器的值，需要跟蹤多少協(xié)程就設(shè)置多少。
每個(gè)協(xié)程在執(zhí)行完畢的時(shí)候，一定要調(diào) Done 方法，讓計(jì)數(shù)器減1，告訴 WaitGroup 該協(xié)程已經(jīng)執(zhí)行完畢。
最后調(diào)用 Wait 方法，一直等待，直到計(jì)數(shù)器的值變?yōu)?，也就是所有跟蹤的協(xié)程執(zhí)行完畢了。

通過 WaitGroup 可以很好地跟蹤協(xié)程，在協(xié)程執(zhí)行完畢后，整個(gè) main 函數(shù)才能執(zhí)行完畢。

package main
import (
        "fmt"
        "sync"
)
var (
        nSum int
        mutex sync.RWMutex
)
func add(i int) {
        mutex.Lock()
        defer mutex.Unlock()
        nSum += i
}
func main() {
        var wg sync.WaitGroup
        wg.Add(100)
        for i := 0; i < 100; i++ {
                go func() {
                        defer wg.Done()
                        add(10)
                }()
        }
        wg.Wait()
        fmt.Println("nSum=", nSum)
}

運(yùn)行結(jié)果，會(huì)發(fā)現(xiàn)輸出執(zhí)行速度方面會(huì)清爽很多。

sync.WaitGroup適合協(xié)調(diào)多個(gè)goroutine共同做一件事情的場景。比如下載較大的文件時(shí)，為了加快下載速度，我們會(huì)使用多線程（協(xié)程）下載。假設(shè)使用10個(gè)協(xié)程，每個(gè)協(xié)程下載文件的1/10大小，只有10個(gè)協(xié)程都下載好了整個(gè)文件才算是下載好了。再比如流水線上，下個(gè)階段需要上個(gè)階段把所有數(shù)據(jù)準(zhǔn)備好，10個(gè)協(xié)程準(zhǔn)備數(shù)據(jù)，等所有協(xié)程處理完后，統(tǒng)一進(jìn)入下個(gè)階段繼續(xù)執(zhí)行.....

sync.Once

讓代碼只執(zhí)行一次，哪怕是在高并發(fā)的情況下，比如創(chuàng)建一個(gè)單例。

先看個(gè)例子

package main
import (
        "fmt"
        "sync"
)
func main() {
        var once sync.Once
        onceBody := func() {
                fmt.Println("Only once")
        }
        done := make(chan bool)        // 用于等待協(xié)程執(zhí)行完畢
        for i := 0; i < 10; i++ {        // 啟動(dòng) 10 個(gè)協(xié)程
                go func(n int) {
                        fmt.Println(n)
                        once.Do(onceBody)
                        done<-true
                }(i)
        }
        for i := 0; i < 10; i++ {
                <-done
        }
}

運(yùn)行結(jié)果如下，

使用 WaitGroup 來保證子協(xié)程執(zhí)行完畢，也可以這樣寫，

package main
import (
        "fmt"
        "sync"
)
func main() {
        var once sync.Once
        onceBody := func() {
                fmt.Println("Only once")
        }
        var wg sync.WaitGroup
        wg.Add(10)
        for i := 0; i < 10; i++ {
                go func(n int) {
                        fmt.Println(n)
                        once.Do(onceBody)
                        wg.Done()
                }(i)
        }
        wg.Wait()
}

sync.Cond

可以用做發(fā)令槍，關(guān)鍵點(diǎn)在于 goroutine 開始的時(shí)候是等待的。Cond 一聲令下，所有 goroutine 都開始執(zhí)行。sync.Cond 從字面意思看是條件變量，除此之外，還具有阻塞和喚醒協(xié)程的功能，所以可以在滿足一定條件的情況下喚醒協(xié)程。

sync.Cond有三個(gè)方法，

Wait，阻塞當(dāng)前協(xié)程，直到其他協(xié)程調(diào)用signal或broadcast來喚醒，使用時(shí)需要加鎖
Signal，喚醒一個(gè)等待時(shí)間最長的協(xié)程
Broadcast就是廣播，喚醒所有等待的協(xié)程

注意，在調(diào)用 Signal 或者 Broadcast 之前，一定要確保目標(biāo)協(xié)程要處于等待 Wait 阻塞狀態(tài)，不然會(huì)出現(xiàn)死鎖問題。和 java 里邊的 wait、notify、notifyall 類似。

package main
import (
        "fmt"
        "sync"
        "time"
)
func main() {
        cond := sync.NewCond(&sync.Mutex{})
        var wg sync.WaitGroup
        wg.Add(11)
        for i := 0; i < 10; i++ {
                go func(n int) {
                        defer wg.Done()
                        fmt.Println("ready", n)
                        cond.L.Lock()
                        cond.Wait()
                        fmt.Println("go", n)
                        cond.L.Unlock()
                }(i)
        }
        time.Sleep(time.Second)
        go func() {
                defer wg.Done()
                fmt.Println("beng beng...")    // 發(fā)令槍響
                cond.Broadcast()
        }()
        wg.Wait()
}

運(yùn)行結(jié)果如下，

sync.Map

Go 中的 map 類型是并發(fā)不安全的，在實(shí)際開發(fā)中，這種類型不能用在并發(fā)寫的場景，并發(fā)讀還是可以的。不過 slice 是并發(fā)安全的，有時(shí)候可以使用 slice 來代替 map，但需要迭代元素進(jìn)行轉(zhuǎn)換。這時(shí) sync.Map 也是一個(gè)不錯(cuò)的選擇。