Go Mutex 的基本用法

Mutex 我們一般只會(huì)用到它的兩個(gè)方法：

• Lock：獲取互斥鎖。(只會(huì)有一個(gè)協(xié)程可以獲取到鎖，通常用在臨界區(qū)開(kāi)始的地方。)
• Unlock: 釋放互斥鎖。(釋放獲取到的鎖，通常用在臨界區(qū)結(jié)束的地方。)

Mutex 的模型可以用下圖表示：

說(shuō)明：

• 同一時(shí)刻只能有一個(gè)協(xié)程獲取到 Mutex 的使用權(quán)，其他協(xié)程需要排隊(duì)等待（也就是上圖的 G1->G2->Gn）。
• 擁有鎖的協(xié)程從臨界區(qū)退出的時(shí)候需要使用 Unlock 來(lái)釋放鎖，這個(gè)時(shí)候等待隊(duì)列的下一個(gè)協(xié)程可以獲取到鎖（實(shí)際實(shí)現(xiàn)比這里說(shuō)的復(fù)雜很多，后面會(huì)細(xì)說(shuō)），從而進(jìn)入臨界區(qū)。
• 等待的協(xié)程會(huì)在 Lock 調(diào)用處阻塞，Unlock 的時(shí)候會(huì)使得一個(gè)等待的協(xié)程解除阻塞的狀態(tài)，得以繼續(xù)執(zhí)行。

這幾點(diǎn)也是 Mutex 的基本原理。

Go Mutex 原子操作

Mutex結(jié)構(gòu)體定義：

type Mutex struct {
   state int32 // 狀態(tài)字段
   sema  uint32 // 信號(hào)量
}

其中 state 字段記錄了四種不同的信息：

這四種不同信息在源碼中定義了不同的常量：

const (
   mutexLocked      = 1 << iota // 表示有 goroutine 擁有鎖
   mutexWoken                   // 喚醒（就是第 2 位）
   mutexStarving                // 饑餓（第 3 位）
   mutexWaiterShift = iota      // 表示第 4 位開(kāi)始，表示等待者的數(shù)量
   starvationThresholdNs = 1e6  // 1ms 進(jìn)入饑餓模式的等待時(shí)間閾值
)

而 sema 的含義比較簡(jiǎn)單，就是一個(gè)用作不同 goroutine 同步的信號(hào)量。

go 的 Mutex 實(shí)現(xiàn)中，state 字段是一個(gè) 32 位的整數(shù)，不同的位記錄了四種不同信息，在這種情況下， 只需要通過(guò)原子操作就可以保證一次性實(shí)現(xiàn)對(duì)四種不同狀態(tài)信息的更改，而不需要更多額外的同步機(jī)制。

但是毋庸置疑，這種實(shí)現(xiàn)會(huì)大大降低代碼的可讀性，因?yàn)橥ㄟ^(guò)一個(gè)整數(shù)來(lái)記錄不同的信息， 就意味著，需要通過(guò)各種位運(yùn)算來(lái)實(shí)現(xiàn)對(duì)這個(gè)整數(shù)不同位的修改。

當(dāng)然，這只是 Mutex 實(shí)現(xiàn)中最簡(jiǎn)單的一種位運(yùn)算了。下面以 state 記錄的四種不同信息為維度來(lái)具體講解一下：

mutexLocked：這是 state 的最低位，1 表示鎖被占用，0 表示鎖沒(méi)有被占用。

new := mutexLocked 新?tīng)顟B(tài)為上鎖狀態(tài)

mutexWoken: 這是表示是否有協(xié)程被喚醒了的狀態(tài)

• new = (old - 1<<mutexWaiterShift) | mutexWoken 等待者數(shù)量減去 1 的同時(shí)，設(shè)置喚醒標(biāo)識(shí)
• new &^= mutexWoken 清除喚醒標(biāo)識(shí)

mutexStarving：饑餓模式的標(biāo)識(shí)

new |= mutexStarving 設(shè)置饑餓標(biāo)識(shí)

等待者數(shù)量：state >> mutexWaiterShift 就是等待者的數(shù)量，也就是上面提到的 FIFO 隊(duì)列中 goroutine 的數(shù)量

• new += 1 << mutexWaiterShift 等待者數(shù)量加 1
• delta := int32(mutexLocked - 1<<mutexWaiterShift) 上鎖的同時(shí)，將等待者數(shù)量減 1

在上面做了這一系列的位運(yùn)算之后，我們會(huì)得到一個(gè)新的 state 狀態(tài)，假設(shè)名為 new，那么我們就可以通過(guò) CAS 操作來(lái)將 Mutex 的 state 字段更新：

atomic.CompareAndSwapInt32(&m.state, old, new)

通過(guò)上面這個(gè)原子操作，我們就可以一次性地更新 Mutex 的 state 字段，也就是一次性更新了四種狀態(tài)信息。

這種通過(guò)一個(gè)整數(shù)記錄不同狀態(tài)的寫(xiě)法在 sync 包其他的一些地方也有用到，比如 WaitGroup 中的 state 字段。

最后，對(duì)于這種操作，我們需要注意的是，因?yàn)槲覀冊(cè)趫?zhí)行 CAS 前后是沒(méi)有其他什么鎖或者其他的保護(hù)機(jī)制的， 這也就意味著上面的這個(gè) CAS 操作是有可能會(huì)失敗的，那如果失敗了怎么辦呢？

如果失敗了，也就意味著肯定有另外一個(gè) goroutine 率先執(zhí)行了 CAS 操作并且成功了，將 state 修改為了一個(gè)新的值。 這個(gè)時(shí)候，其實(shí)我們前面做的一系列位運(yùn)算得到的結(jié)果實(shí)際上已經(jīng)不對(duì)了，在這種情況下，我們需要獲取最新的 state，然后再次計(jì)算得到一個(gè)新的 state。

所以我們會(huì)在源碼里面看到 CAS 操作是寫(xiě)在 for 循環(huán)里面的。

state的狀態(tài)及枚舉

在看下面代碼之前，一定要記住這幾個(gè)狀態(tài)之間的 與運(yùn)算 或運(yùn)算，否則代碼里的與運(yùn)算或運(yùn)算

state:   |32|31|...|3|2|1|
         __________/ | |
               |      | |
               |      | mutex的占用狀態(tài)（1被占用，0可用）
               |      |
               |      mutex的當(dāng)前goroutine是否被喚醒
               |
               當(dāng)前阻塞在mutex上的goroutine數(shù)

互斥鎖的作用

互斥鎖是保證同步的一種工具，主要體現(xiàn)在以下2個(gè)方面：

避免多個(gè)線程在同一時(shí)刻操作同一個(gè)數(shù)據(jù)塊 (sum)

可以協(xié)調(diào)多個(gè)線程，以避免它們?cè)谕粫r(shí)刻執(zhí)行同一個(gè)代碼塊 (sum++)

什么時(shí)候用

需要保護(hù)一個(gè)數(shù)據(jù)或數(shù)據(jù)塊時(shí)

需要協(xié)調(diào)多個(gè)協(xié)程串行執(zhí)行同一代碼塊，避免并發(fā)問(wèn)題時(shí)

比如 經(jīng)常遇到A給B轉(zhuǎn)賬100元的例子，這個(gè)時(shí)候就可以用互斥鎖來(lái)實(shí)現(xiàn)。

注意的坑

1. 不同 goroutine 可以 Unlock 同一個(gè) Mutex，但是 Unlock 一個(gè)無(wú)鎖狀態(tài)的 Mutex 就會(huì)報(bào)錯(cuò)。

2. 因?yàn)?mutex 沒(méi)有記錄 goroutine_id，所以要避免在不同的協(xié)程中分別進(jìn)行上鎖/解鎖操作，不然很容易造成死鎖。

建議： 先 Lock 再 Unlock、兩者成對(duì)出現(xiàn)。

3. Mutex 不是可重入鎖

Mutex 不會(huì)記錄持有鎖的協(xié)程的信息，所以如果連續(xù)兩次 Lock 操作，就直接死鎖了。

如何實(shí)現(xiàn)可重入鎖？記錄上鎖的 goroutine 的唯一標(biāo)識(shí)，在重入上鎖/解鎖的時(shí)候只需要增減計(jì)數(shù)。

type RecursiveMutex struct {
   sync.Mutex
   owner     int64 // 當(dāng)前持有鎖的 goroutine id // 可以換成其他的唯一標(biāo)識(shí)
   recursion int32 // 這個(gè) goroutine 重入的次數(shù)
}
func (m *RecursiveMutex) Lock() {
   gid := goid.Get()  // 獲取唯一標(biāo)識(shí)
   // 如果當(dāng)前持有鎖的 goroutine 就是這次調(diào)用的 goroutine，說(shuō)明是重入
   if atomic.LoadInt64(&m.owner) == gid {
      m.recursion++
      return
   }
   m.Mutex.Lock()
   // 獲得鎖的 goroutine 第一次調(diào)用，記錄下它的 goroutine id，調(diào)用次數(shù)加1
   atomic.StoreInt64(&m.owner, gid)
   m.recursion = 1
}
func (m *RecursiveMutex) Unlock() {
   gid := goid.Get()
   // 非持有鎖的 goroutine 嘗試釋放鎖，錯(cuò)誤的使用
   if atomic.LoadInt64(&m.owner) != gid {
      panic(fmt.Sprintf("wrong the owner(%d): %d!", m.owner, gid))
   }
   // 調(diào)用次數(shù)減1
   m.recursion--
   if m.recursion != 0 { // 如果這個(gè) goroutine 還沒(méi)有完全釋放，則直接返回
      return
   }
   // 此 goroutine 最后一次調(diào)用，需要釋放鎖
   atomic.StoreInt64(&m.owner, -1)
   m.Mutex.Unlock()
}

4.多高的 QPS 才能讓 Mutex 產(chǎn)生強(qiáng)烈的鎖競(jìng)爭(zhēng)？

模擬一個(gè) 10ms 的接口，接口邏輯中使用全局共享的 Mutex，會(huì)發(fā)現(xiàn)在較低 QPS 的時(shí)候就開(kāi)始產(chǎn)生激烈的鎖競(jìng)爭(zhēng)（打印鎖等待時(shí)間和接口時(shí)間）。

解決方式：首先要盡量避免使用 Mutex。如果要使用 Mutex，盡量多聲明一些 Mutex，采用取模分片的方式去使用其中一個(gè) Mutex 進(jìn)行資源控制。避免一個(gè) Mutex 對(duì)應(yīng)過(guò)多的并發(fā)。

簡(jiǎn)單總結(jié)：壓測(cè)或者流量高的時(shí)候發(fā)現(xiàn)系統(tǒng)不正常，打開(kāi) pprof 發(fā)現(xiàn) goroutine 指標(biāo)在飆升，并且大量 Goroutine 都阻塞在 Mutex 的 Lock 上，這種現(xiàn)象下基本就可以確定是鎖競(jìng)爭(zhēng)。

5. Mutex 千萬(wàn)不能被復(fù)制

因?yàn)閺?fù)制的時(shí)候會(huì)將原鎖的 state 值也進(jìn)行復(fù)制。復(fù)制之后，一個(gè)新 Mutex 可能莫名處于持有鎖、喚醒或者饑餓狀態(tài)，甚至等阻塞等待數(shù)量遠(yuǎn)遠(yuǎn)大于0。而原鎖 Unlock 的時(shí)候，卻不會(huì)影響復(fù)制鎖。

關(guān)于鎖的使用建議

寫(xiě)業(yè)務(wù)時(shí)不能全局使用同一個(gè) Mutex

千萬(wàn)不要將要加鎖和解鎖分到兩個(gè)以上 Goroutine 中進(jìn)行（容易形成死鎖）

Mutex 千萬(wàn)不能被復(fù)制（包括不能通過(guò)函數(shù)參數(shù)傳遞），否則會(huì)復(fù)制傳參前鎖的狀態(tài)：已鎖定 or 未鎖定。很容易產(chǎn)生死鎖，關(guān)鍵是編譯器還發(fā)現(xiàn)不了這個(gè) Deadlock~

盡量避免使用 Mutex，如果非使用不可，盡量多聲明一些 Mutex，采用取模分片的方式去使用其中一個(gè) Mutex（分段鎖）（盡量減小鎖的顆粒度）

參考

標(biāo)準(zhǔn)庫(kù)文檔 —— sync.Mutex

以上就是Golang Mutex錯(cuò)過(guò)會(huì)后悔的重要知識(shí)點(diǎn)分享的詳細(xì)內(nèi)容，更多關(guān)于Golang Mutex的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

最新評(píng)論