一文帶你深入了解Golang中的Mutex

更新時(shí)間：2023年03月31日 08:22:04 作者：eleven26

這篇文章主要為大家詳細(xì)介紹了Golang中Mutex的相關(guān)知識(shí)，知其然，更要知其所以然。文中的示例代碼講解詳細(xì)，感興趣的小伙伴可以了解一下

在我們的日常開(kāi)發(fā)中，總會(huì)有時(shí)候需要對(duì)一些變量做并發(fā)讀寫(xiě)，比如 web 應(yīng)用在同時(shí)接到多個(gè)請(qǐng)求之后，需要對(duì)一些資源做初始化，而這些資源可能是只需要初始化一次的，而不是每一個(gè) http 請(qǐng)求都初始化，在這種情況下，我們需要限制只能一個(gè)協(xié)程來(lái)做初始化的操作，比如初始化數(shù)據(jù)庫(kù)連接等，這個(gè)時(shí)候，我們就需要有一種機(jī)制，可以限制只有一個(gè)協(xié)程來(lái)執(zhí)行這些初始化的代碼。在 go 語(yǔ)言中，我們可以使用互斥鎖（Mutex）來(lái)實(shí)現(xiàn)這種功能。

互斥鎖的定義

這里引用一下維基百科的定義：

互斥鎖（Mutual exclusion，縮寫(xiě) Mutex）是一種用于多線程編程中，防止兩個(gè)線程同時(shí)對(duì)同一公共資源（比如全局變量）進(jìn)行讀寫(xiě)的機(jī)制。該目的通過(guò)將代碼切片成一個(gè)一個(gè)的臨界區(qū)域（critical section）達(dá)成。臨街區(qū)域指的是一塊對(duì)公共資源進(jìn)行訪問(wèn)的代碼，并非一種機(jī)制或是算法。

互斥，顧名思義，也就是只有一個(gè)線程能持有鎖。當(dāng)然，在 go 中，是只有一個(gè)協(xié)程能持有鎖。

下面是一個(gè)簡(jiǎn)單的例子：

var sum int // 和
var mu sync.Mutex // 互斥鎖

// add 將 sum 加 1
func add() {
    // 獲取鎖，只能有一個(gè)協(xié)程獲取到鎖，
    // 其他協(xié)程需要阻塞等待鎖釋放才能獲取到鎖。
   mu.Lock()
   // 臨界區(qū)域
   sum++
   mu.Unlock()
}

func TestMutex(t *testing.T) {
   // 啟動(dòng) 1000 個(gè)協(xié)程
   var wg sync.WaitGroup
   wg.Add(1000)

   for i := 0; i < 1000; i++ {
      go func() {
         // 每個(gè)協(xié)程里面調(diào)用 add()
         add()
         wg.Done()
      }()
   }

   // 等待所有協(xié)程執(zhí)行完畢
   wg.Wait()
   // 最終 sum 的值應(yīng)該是 1000
   assert.Equal(t, 1000, sum)
}

上面的例子中，我們定義了一個(gè)全局變量 sum，用于存儲(chǔ)和，然后定義了一個(gè)互斥鎖 mu，在 add() 函數(shù)中，我們使用 mu.Lock() 來(lái)加鎖，然后對(duì) sum 進(jìn)行加 1 操作，最后使用 mu.Unlock() 來(lái)解鎖，這樣就保證了在任意時(shí)刻，只有一個(gè)協(xié)程能夠?qū)?sum 進(jìn)行加 1 操作，從而保證了在并發(fā)執(zhí)行 add() 操作的時(shí)候 sum 的值是正確的。

上面這個(gè)例子，在我之前的文章中已經(jīng)作為例子出現(xiàn)過(guò)很多次了，這里不再贅述了。

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 的基本原理。

互斥鎖使用的兩個(gè)例子

了解了 go Mutex 基本原理之后，讓我們?cè)賮?lái)看看 Mutex 的一些使用的例子。

gin Context 中的 Set 方法

一個(gè)很常見(jiàn)的場(chǎng)景就是，并發(fā)對(duì) map 進(jìn)行讀寫(xiě)，熟悉 go 的朋友應(yīng)該知道，go 中的 map 是不支持并發(fā)讀寫(xiě)的，如果我們對(duì) map 進(jìn)行并發(fā)讀寫(xiě)會(huì)導(dǎo)致 panic。

而在 gin 的 Context 結(jié)構(gòu)體中，也有一個(gè) map 類(lèi)型的字段 Keys，用來(lái)在上下文間傳遞鍵值對(duì)數(shù)據(jù)，所以在通過(guò) Set 來(lái)設(shè)置鍵值對(duì)的時(shí)候需要使用 c.mu.Lock() 來(lái)先獲取互斥鎖，然后再對(duì) Keys 做設(shè)置。

// Set is used to store a new key/value pair exclusively for this context.
// It also lazy initializes  c.Keys if it was not used previously.
func (c *Context) Set(key string, value any) {
    // 獲取鎖
   c.mu.Lock()
    // 如果 Keys 還沒(méi)初始化，則進(jìn)行初始化
   if c.Keys == nil {
      c.Keys = make(map[string]any)
   }

    // 設(shè)置鍵值對(duì)
   c.Keys[key] = value
    // 釋放鎖
   c.mu.Unlock()
}

同樣的，對(duì) Keys 做讀操作的時(shí)候也需要使用互斥鎖：

// Get returns the value for the given key, ie: (value, true).
// If the value does not exist it returns (nil, false)
func (c *Context) Get(key string) (value any, exists bool) {
    // 獲取鎖
   c.mu.RLock()
    // 讀取 key
   value, exists = c.Keys[key]
    // 釋放鎖
   c.mu.RUnlock()
   return
}

可能會(huì)有人覺(jué)得奇怪，為什么從 map 中讀也還需要鎖。這是因?yàn)?，如果讀的時(shí)候沒(méi)有鎖保護(hù)，那么就有可能在 Set 設(shè)置的過(guò)程中，同時(shí)也在進(jìn)行讀操作，這樣就會(huì) panic 了。

這個(gè)例子想要說(shuō)明的是，像 map 這種數(shù)據(jù)結(jié)構(gòu)本身就不支持并發(fā)讀寫(xiě)，我們這種情況下只有使用 Mutex 了。

sync.Pool 中的 pinSlow 方法

在 sync.Pool 的實(shí)現(xiàn)中，有一個(gè)全局變量記錄了進(jìn)程內(nèi)所有的 sync.Pool 對(duì)象，那就是 allPools 變量，另外有一個(gè)鎖 allPoolsMu 用來(lái)保護(hù)對(duì) allPools 的讀寫(xiě)操作：

var (
   // 保護(hù) allPools 和 oldPools 的互斥鎖。
   allPoolsMu Mutex

   // allPools is the set of pools that have non-empty primary
   // caches. Protected by either 1) allPoolsMu and pinning or 2)
   // STW.
   allPools []*Pool

   // oldPools is the set of pools that may have non-empty victim
   // caches. Protected by STW.
   oldPools []*Pool
)

pinSlow 方法中會(huì)在 allPoolsMu 的保護(hù)下對(duì) allPools 做讀寫(xiě)操作：

func (p *Pool) pinSlow() (*poolLocal, int) {
   // Retry under the mutex.
   // Can not lock the mutex while pinned.
   runtime_procUnpin()
   allPoolsMu.Lock() // 獲取鎖
   defer allPoolsMu.Unlock() // 函數(shù)返回的時(shí)候釋放鎖
   pid := runtime_procPin()
   // poolCleanup won't be called while we are pinned.
   s := p.localSize
   l := p.local
   if uintptr(pid) < s {
      return indexLocal(l, pid), pid
   }
   if p.local == nil {
      allPools = append(allPools, p) // 全局變量修改
   }
   // If GOMAXPROCS changes between GCs, we re-allocate the array and lose the old one.
   size := runtime.GOMAXPROCS(0)
   local := make([]poolLocal, size)
   atomic.StorePointer(&p.local, unsafe.Pointer(&local[0])) // store-release
   runtime_StoreReluintptr(&p.localSize, uintptr(size))     // store-release
   return &local[pid], pid
}

這個(gè)例子主要是為了說(shuō)明使用 mu 的另外一種非常常見(jiàn)的場(chǎng)景：并發(fā)讀寫(xiě)全局變量。

互斥鎖使用的注意事項(xiàng)

互斥鎖如果使用不當(dāng)，可能會(huì)導(dǎo)致死鎖或者出現(xiàn) panic 的情況，下面是一些常見(jiàn)的錯(cuò)誤：

忘記使用 Unlock 釋放鎖。
Lock 之后還沒(méi) Unlock 之前又使用 Lock 獲取鎖。也就是重復(fù)上鎖，go 中的 Mutex 不可重入。
死鎖：位于臨界區(qū)內(nèi)不同的兩個(gè)協(xié)程都想獲取對(duì)方持有的不同的鎖。
還沒(méi) Lock 之前就 Unlock。這會(huì)導(dǎo)致 panic，因?yàn)檫@是沒(méi)有任何意義的。
復(fù)制 Mutex，比如將 Mutex 作為參數(shù)傳遞。

對(duì)于第 1 點(diǎn)，我們往往可以使用 defer 關(guān)鍵字來(lái)做釋放鎖的操作。第 2 點(diǎn)不太好發(fā)現(xiàn)，只能在開(kāi)發(fā)的時(shí)候多加注意。第 3 點(diǎn)我們?cè)谑褂面i的時(shí)候可以考慮盡量避免在臨界區(qū)內(nèi)再去使用別的鎖。最后，Mutex 是不可以復(fù)制的，這個(gè)可以在編譯之前通過(guò) go vet 來(lái)做檢查。

為什么 Mutex 不能被復(fù)制呢？因?yàn)?Mutex 中包含了鎖的狀態(tài)，如果復(fù)制了，那么這個(gè)狀態(tài)也會(huì)被復(fù)制，如果在復(fù)制前進(jìn)行 Lock，復(fù)制后進(jìn)行 Unlock，那就意味著 Lock 和 Unlock 操作的其實(shí)是兩個(gè)不同的狀態(tài)，這樣顯然是不行的，是釋放不了鎖的。

雖然不可以復(fù)制，但是我們可以通過(guò)傳遞指針類(lèi)型的參數(shù)來(lái)傳遞 Mutex。

互斥鎖鎖定的是什么

在前一篇文章中，我們提到過(guò)，原子操作本質(zhì)上是變量級(jí)的互斥鎖。而互斥鎖本身鎖定的又是什么呢？其實(shí)互斥鎖本質(zhì)上是一個(gè)信號(hào)量，它通過(guò)獲取釋放信號(hào)量，最終使得協(xié)程獲得某一個(gè)代碼塊的執(zhí)行權(quán)力。

也就是說(shuō)，互斥鎖，鎖定的是一塊代碼塊。

我們以 go-zero 里面的 collection/fifo.go 為例子說(shuō)明一下：

// Take takes the first element out of q if not empty.
func (q *Queue) Take() (any, bool) {
   // 獲取互斥鎖（只能有一個(gè)協(xié)程獲取到鎖）
   q.lock.Lock()
   // 函數(shù)返回的時(shí)候釋放互斥鎖（獲取到鎖的協(xié)程釋放鎖之后，其他協(xié)程才能進(jìn)行搶占鎖）
   defer q.lock.Unlock()

   // 下面的代碼只有搶占到（也就是互斥鎖鎖定的代碼塊）
   if q.count == 0 {
      return nil, false
   }

   element := q.elements[q.head]
   q.head = (q.head + 1) % len(q.elements)
   q.count--

   return element, true
}

除了鎖定代碼塊的這一個(gè)作用，有另外一個(gè)比較關(guān)鍵的地方也是我們不能忽視的，那就是 互斥鎖并不保證臨界區(qū)內(nèi)操作的變量不能被其他協(xié)程訪問(wèn)。 互斥鎖只能保證一段代碼只能一個(gè)協(xié)程執(zhí)行，但是對(duì)于臨界區(qū)內(nèi)涉及的共享資源，你在臨界區(qū)外也依然是可以對(duì)其進(jìn)行讀寫(xiě)的。

我們以上面的代碼說(shuō)明一下：在上面的 Take 函數(shù)中，我們對(duì) q.head 和 q.count 都進(jìn)行了操作，雖然這些操作代碼位于臨界區(qū)內(nèi)，但是臨界區(qū)并不保證持有鎖期間其他協(xié)程不會(huì)在臨界區(qū)外去修改 q.head 和 q.count。

下面就是一個(gè)非常典型的錯(cuò)誤的例子：

import (
   "fmt"
   "sync"
   "testing"
)

var mu sync.Mutex
var sum int

// 在鎖的保護(hù)下對(duì) sum 做讀寫(xiě)操作
func test() {
   mu.Lock()
   sum++
   mu.Unlock()
}

func TestMutex(t *testing.T) {
   var wg sync.WaitGroup
   wg.Add(1000)

   for i := 0; i < 500; i++ {
      go func() {
         test()
         wg.Done()
      }()

      // 位于臨界區(qū)外，也依然是可以對(duì) sum 做讀寫(xiě)操作的。
      sum++
   }

   wg.Wait()

   fmt.Println(sum)
}

靠譜的做法是，對(duì)于有共享資源的讀寫(xiě)的操作都使用 Mutex 保護(hù)起來(lái)。

當(dāng)然，如果我們只有一個(gè)變量，那么可能使用原子操作就足夠了。

互斥鎖實(shí)現(xiàn)原理

互斥鎖的實(shí)現(xiàn)有以下幾個(gè)關(guān)鍵的地方：

信號(hào)量：這是操作系統(tǒng)中的同步對(duì)象。
等待隊(duì)列：獲取不到互斥鎖的協(xié)程，會(huì)放入到一個(gè)先入先出隊(duì)列的隊(duì)列尾部。這樣信號(hào)量釋放的時(shí)候，可以依次對(duì)它們喚醒。
原子操作：互斥鎖的實(shí)現(xiàn)中，使用了一個(gè)字段來(lái)記錄了幾種不同的狀態(tài)，使用原子操作可以保證幾種狀態(tài)可以一次性變更完成。

我們先來(lái)看看 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)量。

信號(hào)量

go 的 Mutex 是基于信號(hào)量來(lái)實(shí)現(xiàn)的，那信號(hào)量又是什么呢？

維基百科：信號(hào)量是一個(gè)同步對(duì)象，用于保持在 0 至指定最大值之間的一個(gè)計(jì)數(shù)值。當(dāng)線程完成一次對(duì)該 semaphore 對(duì)象的等待（wait）時(shí)，該計(jì)數(shù)值減一；當(dāng)線程完成一次對(duì) semaphore 對(duì)象的釋放（release）時(shí)，計(jì)數(shù)值加一。

上面這個(gè)解釋有點(diǎn)難懂，通俗地說(shuō)，就是一個(gè)數(shù)字，調(diào)用 wait 的時(shí)候，這個(gè)數(shù)字減去 1，調(diào)用 release 的時(shí)候，這個(gè)數(shù)字加上 1。（還有一個(gè)隱含的邏輯是，如果這個(gè)數(shù)小于 0，那么調(diào)用 wait 的時(shí)候會(huì)阻塞，直到它大于 0。）

對(duì)應(yīng)到 go 的 Mutex 中，有兩個(gè)操作信號(hào)量的函數(shù)：

runtime_Semrelease: 自動(dòng)遞增信號(hào)量并通知等待的 goroutine。
runtime_SemacquireMutex: 是一直等到信號(hào)量大于 0，然后自動(dòng)遞減。

我們注意到了，其實(shí) runtime_SemacquireMutex 是有一個(gè)前提條件的，那就是等到信號(hào)量大于 0。其實(shí)信號(hào)量的兩個(gè)操作 P/V 就是一個(gè)加 1 一個(gè)減 1，所以在實(shí)際使用的時(shí)候，也是需要一個(gè)獲取鎖的操作對(duì)應(yīng)一個(gè)釋放鎖的操作，否則，其他協(xié)程都無(wú)法獲取到鎖，因?yàn)樾盘?hào)量一直不滿足。

等待隊(duì)列

go 中如果已經(jīng)有 goroutine 持有互斥鎖，那么其他的協(xié)程會(huì)放入一個(gè) FIFO 隊(duì)列中，如下圖：

說(shuō)明：

G1 表示持有互斥鎖的 goroutine，G2...Gn 表示一個(gè) goroutine 的等待隊(duì)列，這是一個(gè)先入先出的隊(duì)列。
G1 先持有鎖，得以進(jìn)入臨界區(qū)，其他想搶占鎖的 goroutine 阻塞在 Lock 調(diào)用處。
G1 在使用完鎖后，會(huì)使用 Unlock 來(lái)釋放鎖，本質(zhì)上是釋放了信號(hào)量，然后會(huì)喚醒 FIFO 隊(duì)列頭部的 goroutine。
G2 從 FIFO 隊(duì)列中移除，進(jìn)入臨界區(qū)。G2 使用完鎖之后也會(huì)使用 Unlock 來(lái)釋放鎖。

上面只是一個(gè)大概模型，在實(shí)際實(shí)現(xiàn)中，比這個(gè)復(fù)雜很多倍，下面會(huì)繼續(xù)深入講解。

原子操作

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ù)不同位的修改，比如將上鎖的操作：

new |= mutexLocked

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

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

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

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

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

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

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

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

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

這里并沒(méi)有涵蓋 Mutex 中所有的位運(yùn)算，其他操作在下文講解源碼實(shí)現(xiàn)的時(shí)候會(huì)提到。

在上面做了這一系列的位運(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)里面的。

Mutex 的公平性

在前面，我們提到 goroutien 獲取不到鎖的時(shí)候，會(huì)進(jìn)入一個(gè) FIFO 隊(duì)列的隊(duì)列尾，在實(shí)際實(shí)現(xiàn)中，其實(shí)沒(méi)有那么簡(jiǎn)單，為了獲得更好的性能，在實(shí)現(xiàn)的時(shí)候會(huì)盡量先讓運(yùn)行狀態(tài)的 goroutine 獲得鎖，當(dāng)然如果隊(duì)列中的 goroutine 等待太久（大于 1ms），那么就會(huì)先讓隊(duì)列中的 goroutine 獲得鎖。

下面是文檔中的說(shuō)明：

Mutex 可以處于兩種操作模式：正常模式和饑餓模式。在正常模式下，等待者按照FIFO（先進(jìn)先出）的順序排隊(duì)，但是被喚醒的等待者不擁有互斥鎖，會(huì)與新到達(dá)的 Goroutine 競(jìng)爭(zhēng)所有權(quán)。新到達(dá)的 Goroutine 有優(yōu)勢(shì)——它們已經(jīng)在 CPU 上運(yùn)行，數(shù)量可能很多，因此被喚醒的等待者有很大的機(jī)會(huì)失去鎖。在這種情況下，它將排在等待隊(duì)列的前面。如果等待者未能在1毫秒內(nèi)獲取到互斥鎖，則將互斥鎖切換到饑餓模式。在饑餓模式下，互斥鎖的所有權(quán)直接從解鎖 Goroutine 移交給隊(duì)列前面的等待者。新到達(dá)的 Goroutine 即使看起來(lái)未被鎖定，也不會(huì)嘗試獲取互斥鎖，也不會(huì)嘗試自旋。相反，它們會(huì)將自己排隊(duì)在等待隊(duì)列的末尾。如果等待者獲得互斥鎖的所有權(quán)并發(fā)現(xiàn)（1）它是隊(duì)列中的最后一個(gè)等待者，或者（2）它等待時(shí)間少于1毫秒，則將互斥鎖切換回正常模式。正常模式的性能要優(yōu)于饑餓模式，因?yàn)?Goroutine 可以連續(xù)多次獲取互斥鎖，即使有被阻塞的等待者。饑餓模式很重要，可以防止尾部延遲的病態(tài)情況。

簡(jiǎn)單總結(jié)：

1.Mutex 有兩種模式：正常模式、饑餓模式。

2.正常模式下：

被喚醒的 goroutine 和正在運(yùn)行的 goroutine 競(jìng)爭(zhēng)鎖。這樣可以運(yùn)行中的協(xié)程有機(jī)會(huì)先獲取到鎖，從而避免了協(xié)程切換的開(kāi)銷(xiāo)。性能更好。

3.饑餓模式下：

優(yōu)先讓隊(duì)列中的 goroutine 獲得鎖，并且直接放棄時(shí)間片，讓給隊(duì)列中的 goroutine，運(yùn)行中的 goroutine 想獲取鎖要到隊(duì)尾排隊(duì)。更加公平。

Mutex 源碼剖析

Mutex 本身的源碼其實(shí)很少，但是復(fù)雜程度是非常高的，所以第一次看的時(shí)候可能會(huì)非常懵逼，但是不妨礙我們?nèi)チ私馑拇蟾艑?shí)現(xiàn)原理。

Mutex 中主要有兩個(gè)方法，Lock 和 Unlock，使用起來(lái)非常的簡(jiǎn)單，但是它的實(shí)現(xiàn)可不簡(jiǎn)單。下面我們就來(lái)深入了解一下它的實(shí)現(xiàn)。

Lock

Lock 方法的實(shí)現(xiàn)如下：

// Lock 獲取鎖。
// 如果鎖已在使用中，則調(diào)用 goroutine 將阻塞，直到互斥量可用。
func (m *Mutex) Lock() {
   // Fast path: grab unlocked mutex.
   // 上鎖成功則直接返回
   if atomic.CompareAndSwapInt32(&m.state, 0, mutexLocked) {
      return
   }

   // Slow path (outlined so that the fast path can be inlined)
   // 沒(méi)有上鎖成功，這個(gè)時(shí)候需要做的事情就有點(diǎn)多了。
   m.lockSlow()
}

在 Lock 方法中，第一次獲取鎖的時(shí)候是非常簡(jiǎn)單的，一個(gè)簡(jiǎn)單的原子操作設(shè)置一下 mutexLocked 標(biāo)識(shí)就完成了。但是如果這個(gè)原子操作失敗了，表示有其他 goroutine 先獲取到了鎖，這個(gè)時(shí)候就需要調(diào)用 lockSlow 來(lái)做一些額外的操作了：

// 獲取 mutex 鎖
func (m *Mutex) lockSlow() {
   var waitStartTime int64 // 當(dāng)前協(xié)程開(kāi)始等待的時(shí)間
   starving := false       // 當(dāng)前協(xié)程是否是饑餓模式
   awoke := false          // 喚醒標(biāo)志（是否當(dāng)前協(xié)程就是被喚醒的協(xié)程）
   iter := 0               // 自旋次數(shù)（超過(guò)一定次數(shù)如果還沒(méi)能獲得鎖，就進(jìn)入等待）
   old := m.state          // 舊的狀態(tài)，每次 for 循環(huán)會(huì)重新獲取當(dāng)前的狀態(tài)字段

   for {
      // 自旋：目的是讓正在運(yùn)行中的 goroutine 盡快獲取到鎖。
      // 兩種情況不會(huì)自旋：
      // 1. 饑餓模式：在饑餓模式下，鎖會(huì)直接交給等待隊(duì)列中的 goroutine，所以不會(huì)自旋。
      // 2. 鎖被釋放了：另外如果運(yùn)行到這里的時(shí)候，發(fā)現(xiàn)鎖已經(jīng)被釋放了，也就不需要自旋了。
      if old&(mutexLocked|mutexStarving) == mutexLocked && runtime_canSpin(iter) {
         // 設(shè)置 mutexWoken 標(biāo)識(shí)
         // 如果自旋是有意義的，則會(huì)進(jìn)入到這里，嘗試設(shè)置 mutexWoken 標(biāo)識(shí)。
         // 設(shè)置成功在持有鎖的 goroutine 獲取鎖的時(shí)候不會(huì)喚醒等待隊(duì)列中的 goroutine，下一個(gè)獲取鎖的就是當(dāng)前 goroutine。
         if !awoke && old&mutexWoken == 0 && old>>mutexWaiterShift != 0 &&
            atomic.CompareAndSwapInt32(&m.state, old, old|mutexWoken) {
            // 各個(gè)判斷的含義：
            // !awoke 已經(jīng)被喚醒過(guò)一次了，說(shuō)明當(dāng)前協(xié)程是被從等待隊(duì)列中喚醒的協(xié)程/又或者已經(jīng)成功設(shè)置 mutexWoken 標(biāo)識(shí)了，不需要再喚醒了。
            // old&mutexWoken == 0 如果不等于 0 說(shuō)明有 goroutine 被喚醒了，不會(huì)嘗試設(shè)置 mutexWoken 標(biāo)識(shí)
            // old>>mutexWaiterShift != 0 如果等待隊(duì)列為空，當(dāng)前 goroutine 就是下一個(gè)搶占鎖的 goroutine
            // 前面的判斷都通過(guò)了，才會(huì)進(jìn)行 CAS 操作嘗試設(shè)置 mutexWoken 標(biāo)識(shí)
            awoke = true
         }
         runtime_doSpin() // 自旋
         iter++           // 自旋次數(shù) +1（超過(guò)一定次數(shù)會(huì)停止自旋）
         old = m.state    // 再次獲取鎖的最新?tīng)顟B(tài)，之后會(huì)檢查是否鎖被釋放了
         continue         // 繼續(xù)下一次檢查
      }

      new := old
      // 饑餓模式下，新到達(dá)的 goroutines 必須排隊(duì)。
      // 不是饑餓狀態(tài)，直接競(jìng)爭(zhēng)鎖。
      if old&mutexStarving == 0 {
         new |= mutexLocked
      }
      // 進(jìn)入等待隊(duì)列的兩種情況：
      // 1. 鎖依然被占用。
      // 2. 進(jìn)入了饑餓模式。
      if old&(mutexLocked|mutexStarving) != 0 {
         new += 1 << mutexWaiterShift // 等待者數(shù)量 +1
      }
       // 已經(jīng)等待超過(guò)了 1ms，且鎖被其他協(xié)程占用，則進(jìn)入饑餓模式
      if starving && old&mutexLocked != 0 {
         new |= mutexStarving
      }
      // 喚醒之后，需要重置喚醒標(biāo)志。
      // 不管有沒(méi)有獲取到鎖，都是要清除這個(gè)標(biāo)識(shí)的：
      // 獲取到鎖肯定要清除，如果獲取到鎖，需要讓其他運(yùn)行中的 goroutine 來(lái)?yè)屨兼i；
      // 如果沒(méi)有獲取到鎖，goroutine 會(huì)阻塞，這個(gè)時(shí)候是需要持有鎖的 goroutine 來(lái)喚醒的，如果有 mutexWoken 標(biāo)識(shí)，持有鎖的 goroutine 喚醒不了。
      if awoke {
         if new&mutexWoken == 0 {
            throw("sync: inconsistent mutex state")
         }
         new &^= mutexWoken // 重置喚醒標(biāo)志
      }

      // 成功設(shè)置新?tīng)顟B(tài)
      if atomic.CompareAndSwapInt32(&m.state, old, new) {
         // 原來(lái)鎖的狀態(tài)已釋放，并且不是饑餓狀態(tài)，正常請(qǐng)求到了鎖，返回
         if old&(mutexLocked|mutexStarving) == 0 { // 這意味著當(dāng)前的 goroutine 成功獲取了鎖
            break
         }

         // 如果已經(jīng)被喚醒過(guò)，會(huì)被加入到等待隊(duì)列頭。
         queueLifo := waitStartTime != 0
         if waitStartTime == 0 {
            waitStartTime = runtime_nanotime()
         }
         // 阻塞等待
         // queueLifo 為 true，表示加入到隊(duì)列頭。否則，加入到隊(duì)列尾。
         // (首次加入隊(duì)列加入到隊(duì)尾，不是首次加入則加入隊(duì)頭，這樣等待最久的 goroutine 優(yōu)先能夠獲取到鎖。)
         runtime_SemacquireMutex(&m.sema, queueLifo, 1)
         // 從等待隊(duì)列中喚醒，檢查鎖是否應(yīng)該進(jìn)入饑餓模式。
         starving = starving || runtime_nanotime()-waitStartTime > starvationThresholdNs

         // 獲取當(dāng)前的鎖最新?tīng)顟B(tài)
         old = m.state
         // 如果鎖已經(jīng)處于饑餓狀態(tài)，直接搶到鎖，返回。
         // 饑餓模式下，被喚醒的協(xié)程可以直接獲取到鎖。
         // 新來(lái)的 goroutine 都需要進(jìn)入隊(duì)列等待。
         if old&mutexStarving != 0 {
            // 如果這個(gè) goroutine 被喚醒并且 Mutex 處于饑餓模式，P 的所有權(quán)已經(jīng)移交給我們，
            // 但 Mutex 處于不一致的狀態(tài)：mutexLocked 未設(shè)置，我們?nèi)匀槐灰暈榈却?。修?fù)這個(gè)問(wèn)題。
            if old&(mutexLocked|mutexWoken) != 0 || old>>mutexWaiterShift == 0 {
               throw("sync: inconsistent mutex state")
            }
            // 加鎖，并且減少等待者數(shù)量。
            // 實(shí)際上是兩步操作合成了一步：
            // 1. m.state = m.state + 1 （獲取鎖）
            // 2. m.state = m.state - 1<<mutexWaiterShift（waiter - 1）
            delta := int32(mutexLocked - 1<<mutexWaiterShift)
            // 清除饑餓狀態(tài)的兩種情況：
            // 1. 如果不需要進(jìn)入饑餓模式（當(dāng)前被喚醒的 goroutine 的等待時(shí)間小于 1ms）
            // 2. 原來(lái)的等待者數(shù)量為 1，說(shuō)明是最后一個(gè)被喚醒的 goroutine。
            if !starving || old>>mutexWaiterShift == 1 {
               // 退出饑餓模式
               delta -= mutexStarving
            }
            // 原子操作，設(shè)置新?tīng)顟B(tài)。
            atomic.AddInt32(&m.state, delta)
            break
         }
         // 設(shè)置喚醒標(biāo)記，重新?lián)屨兼i（會(huì)與那些運(yùn)行中的 goroutine 一起競(jìng)爭(zhēng)鎖）
         awoke = true
         iter = 0
      } else {
         // CAS 更新?tīng)顟B(tài)失敗，獲取最新?tīng)顟B(tài)，然后重試
         old = m.state
      }
   }
}

我們可以看到，lockSlow 的處理非常的復(fù)雜，又要考慮讓運(yùn)行中的 goroutine 盡快獲取到鎖，又要考慮不能讓等待隊(duì)列中的 goroutine 等待太久。

代碼中注釋很多，再簡(jiǎn)單總結(jié)一下其中的流程：

1.為了讓循環(huán)中的 goroutine 可以先獲取到鎖，會(huì)先讓 goroutine 自旋等待鎖的釋放，這是因?yàn)檫\(yùn)行中的 goroutine 正在占用 CPU，讓它先獲取到鎖可以避免一些不必要的協(xié)程切換，從而獲得更好的性能。

3.自旋完畢之后，會(huì)嘗試獲取鎖，同時(shí)也要根據(jù)舊的鎖狀態(tài)來(lái)更新鎖的不同狀態(tài)信息，比如是否進(jìn)入饑餓模式等。

3.計(jì)算得到一個(gè)新的 state 后，會(huì)進(jìn)行 CAS 操作嘗試更新 state 狀態(tài)。

4.CAS 失敗會(huì)重試上面的流程。

5.CAS 成功之后會(huì)做如下操作：

判斷當(dāng)前是否已經(jīng)獲取到鎖，如果是，則返回，Lock 成功了。
會(huì)判斷當(dāng)前的 goroutine 是否是已經(jīng)被喚醒過(guò)，如果是，會(huì)將當(dāng)前 goroutine 加入到等待隊(duì)列頭部。
調(diào)用 runtime_SemacquireMutex，進(jìn)入阻塞狀態(tài)，等待下一次喚醒。
喚醒之后，判斷是否需要進(jìn)入饑餓模式。
最后，如果已經(jīng)是饑餓模式，當(dāng)前 goroutine 直接獲取到鎖，退出循環(huán)，否則，再進(jìn)行下一次搶占鎖的循環(huán)中。

具體流程我們可以參考一下下面的流程圖：

圖中有一些矩形方框描述了 unlockSlow 的關(guān)鍵流程。

Unlock

Unlock 方法的實(shí)現(xiàn)如下：

// Unlock 釋放互斥鎖。
// 如果 m 在進(jìn)入 Unlock 時(shí)未被鎖定，則會(huì)出現(xiàn)運(yùn)行時(shí)錯(cuò)誤。
func (m *Mutex) Unlock() {
   // Fast path: drop lock bit.
   // unlock 成功
   // unLock 操作實(shí)際上是將 state 減去 1。
   new := atomic.AddInt32(&m.state, -mutexLocked)
   if new != 0 { // 等待隊(duì)列為空的時(shí)候直接返回了
      // 喚醒一個(gè)等待鎖的 goroutine
      m.unlockSlow(new)
   }
}

Unlock 做了兩件事：

釋放當(dāng)前 goroutine 持有的互斥鎖：也就是將 state 減去 1
喚醒等待隊(duì)列中的下一個(gè) goroutine

如果只有一個(gè) goroutine 在使用鎖，只需要簡(jiǎn)單地釋放鎖就可以了。但是如果有其他的 goroutine 在阻塞等待，那么持有互斥鎖的 goroutine 就有義務(wù)去喚醒下一個(gè) goroutine。

喚醒的流程相對(duì)復(fù)雜一些：

// unlockSlow 喚醒下一個(gè)等待鎖的協(xié)程。
func (m *Mutex) unlockSlow(new int32) {
   // 如果未加鎖，則會(huì)拋出錯(cuò)誤。
   if (new+mutexLocked)&mutexLocked == 0 {
      fatal("sync: unlock of unlocked mutex")
   }

   // 下面的操作是喚醒一個(gè)在等待鎖的協(xié)程。
   // 存在兩種情況：
   // 1. 正常模式：
   //  a. 不需要喚醒：沒(méi)有等待者、鎖已經(jīng)被搶占、有其他運(yùn)行中的協(xié)程在嘗試獲取鎖、已經(jīng)進(jìn)入了饑餓模式
   //   b. 需要喚醒：其他情況
   // 2. 饑餓模式：?jiǎn)拘训却?duì)列頭部的那個(gè)協(xié)程
   if new&mutexStarving == 0 {
      // 不是饑餓模式
      old := new
      // 自旋
      for {
         // 下面幾種情況不需要喚醒：
         // 1. 沒(méi)有等待者了（沒(méi)得喚醒）
         // 2. 鎖已經(jīng)被占用（只能有一個(gè) goroutine 持有鎖）
         // 3. 有其他運(yùn)行中的協(xié)程已經(jīng)被喚醒（運(yùn)行中的 goroutine 通過(guò)自旋先搶占到了鎖）
         // 4. 饑餓模式（饑餓模式下，所有新的 goroutine 都要排隊(duì)，饑餓模式會(huì)直接喚醒等待隊(duì)列頭部的 gorutine）
         if old>>mutexWaiterShift == 0 || old&(mutexLocked|mutexWoken|mutexStarving) != 0 {
            return
         }
         // 獲取到喚醒等待者的權(quán)力，開(kāi)始喚醒一個(gè)等待者。
         // 下面這一行實(shí)際上是兩個(gè)操作：
         // 1. waiter 數(shù)量 - 1
         // 2. 設(shè)置 mutexWoken 標(biāo)志
         new = (old - 1<<mutexWaiterShift) | mutexWoken
         if atomic.CompareAndSwapInt32(&m.state, old, new) {
            // 正常模式下喚醒了一個(gè) goroutine
            //（第二個(gè)參數(shù)為 false，表示當(dāng)前的 goroutine 在釋放信號(hào)量后還會(huì)繼續(xù)執(zhí)行直到用完時(shí)間片）
            runtime_Semrelease(&m.sema, false, 1)
            return
         }
         // 喚醒失敗，進(jìn)行下一次嘗試。
         old = m.state
      }
   } else {
      // 饑餓模式：將互斥鎖的所有權(quán)移交給下一個(gè)等待者，并放棄我們的時(shí)間片，以便下一個(gè)等待者可以立即開(kāi)始運(yùn)行。
      // 注意：如果“mutexLocked”未設(shè)置，等待者在喚醒后會(huì)將其設(shè)置。
      // 但是，如果設(shè)置了“mutexStarving”，則仍然認(rèn)為互斥鎖已被鎖定，因此新到來(lái)的goroutine不會(huì)獲取它。
      //
      // 當(dāng)前的 goroutine 放棄 CPU 時(shí)間片，讓給阻塞在 sema 的 goroutine。
      runtime_Semrelease(&m.sema, true, 1)
   }
}

unlockSlow 邏輯相比 lockSlow 要簡(jiǎn)單許多，我們可以再結(jié)合下面的流程圖來(lái)閱讀上面的源碼：

runtime_Semrelease 第二個(gè)參數(shù)的含義

細(xì)心的朋友可能注意到了，在 unlockSlow 的實(shí)現(xiàn)中，有兩處地方調(diào)用了 runtime_Semrelease 這個(gè)方法，這個(gè)方法的作用是釋放一個(gè)信號(hào)量，這樣可以讓阻塞在信號(hào)量上的 goroutine 得以繼續(xù)執(zhí)行。它的第一個(gè)參數(shù)我們都知道，是信號(hào)量，而第二個(gè)參數(shù) true 和 false 分別傳遞了一次，那么 true 和 false 分別有什么作用呢？

答案是，設(shè)置為 true 的時(shí)候，當(dāng)前的 goroutine 會(huì)直接放棄自己的時(shí)間片，將 P 的使用權(quán)交給 Mutex 等待隊(duì)列中的第一個(gè) goroutine，這樣的目的是，讓 Mutex 等待隊(duì)列中的 goroutine 可以盡快地獲取到鎖。

總結(jié)

互斥鎖在并發(fā)編程中也算是非常常見(jiàn)的一種操作了，使用互斥鎖可以限制只有一個(gè) goroutine 可以進(jìn)入臨界區(qū)，這對(duì)于并發(fā)修改全局變量、初始化等情況非常好用。最后，再總結(jié)一下本文所講述的內(nèi)容：

1.互斥鎖是一種用于多線程編程中，防止兩個(gè)線程同時(shí)對(duì)同一公共資源進(jìn)行讀寫(xiě)的機(jī)制。go 中的互斥鎖實(shí)現(xiàn)是 sync.Mutex。

2.Mutex 的操作只有兩個(gè)：

Lock 獲取鎖，同一時(shí)刻只能有一個(gè) goroutine 可以獲取到鎖，其他 goroutine 會(huì)先通過(guò)自旋搶占鎖，搶不到則阻塞等待。
Unlock 釋放鎖，釋放鎖之前必須有 goroutine 持有鎖。釋放鎖之后，會(huì)喚醒等待隊(duì)列中的下一個(gè) goroutine。

3.Mutex 常見(jiàn)的使用場(chǎng)景有兩個(gè)：

并發(fā)讀寫(xiě) map：如 gin 中 Context 的 Keys 屬性的讀寫(xiě)。
并發(fā)讀寫(xiě)全局變量：如 sync.Pool 中對(duì) allPools 的讀寫(xiě)。

4.使用 Mutex 需要注意以下幾點(diǎn)：

不要忘記使用 Unlock 釋放鎖
Lock 之后，沒(méi)有釋放鎖之前，不能再次使用 Lock
注意不同 goroutine 競(jìng)爭(zhēng)不同鎖的情況，需要考慮一下是否有可能會(huì)死鎖
在 Unlock 之前，必須已經(jīng)調(diào)用了 Lock，否則會(huì) panic
在第一次使用 Mutex 之后，不能復(fù)制，因?yàn)檫@樣一來(lái) Mutex 的狀態(tài)也會(huì)被復(fù)制。這個(gè)可以使用 go vet 來(lái)檢查。

5.互斥鎖可以保護(hù)一塊代碼塊只能有一個(gè) goroutine 執(zhí)行，但是不保證臨界區(qū)內(nèi)操作的變量不被其他 goroutine 做并發(fā)讀寫(xiě)操作。

6.go 的 Mutex 基于以下技術(shù)實(shí)現(xiàn)：

信號(hào)量：這是操作系統(tǒng)層面的同步機(jī)制
隊(duì)列：在 goroutine 獲取不到鎖的時(shí)候，會(huì)將這些 goroutine 放入一個(gè) FIFO 隊(duì)列中，下次喚醒會(huì)喚醒隊(duì)列頭的 goroutine
原子操作：state 字段記錄了四種不同的信息，通過(guò)原子操作就可以保證數(shù)據(jù)的完整性

7.go Mutex 的公平性：

正在運(yùn)行的 goroutine 如果需要鎖的話，盡量讓它先獲取到鎖，可以避免不必要的協(xié)程上下文切換。會(huì)和被喚醒的 goroutine 一起競(jìng)爭(zhēng)鎖。
但是如果等待隊(duì)列中的 goroutine 超過(guò)了 1ms 還沒(méi)有獲取到鎖，那么會(huì)進(jìn)入饑餓模式

8.go Mutex 的兩種模式：

正常模式：運(yùn)行中的 goroutine 有一定機(jī)會(huì)比等待隊(duì)列中的 goroutine 先獲取到鎖，這種模式有更好的性能。
饑餓模式：所有后來(lái)的 goroutine 都直接進(jìn)入等待隊(duì)列，會(huì)依次從等待隊(duì)列頭喚醒 goroutine。可以有效避免尾延遲。

9.饑餓模式下，Unlock 的時(shí)候會(huì)直接將當(dāng)前 goroutine 所在 P 的使用權(quán)交給等待隊(duì)列頭部的 goroutine，放棄原本屬于自己的時(shí)間片。

以上就是一文帶你深入了解Golang中的Mutex的詳細(xì)內(nèi)容，更多關(guān)于Golang Mutex的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: