引言

我們知道，go 里面提供了 map 這種類(lèi)型讓我們可以存儲(chǔ)鍵值對(duì)數(shù)據(jù)，但是如果我們?cè)诓l(fā)的情況下使用 map 的話，就會(huì)發(fā)現(xiàn)它是不支持并發(fā)地進(jìn)行讀寫(xiě)的（會(huì)報(bào)錯(cuò)）。 在這種情況下，我們可以使用 sync.Mutex 來(lái)保證并發(fā)安全，但是這樣會(huì)導(dǎo)致我們?cè)谧x寫(xiě)的時(shí)候，都需要加鎖，這樣就會(huì)導(dǎo)致性能的下降。

除了使用互斥鎖這種相對(duì)低效的方式，我們還可以使用 sync.Map 來(lái)保證并發(fā)安全，它在某些場(chǎng)景下有比使用 sync.Mutex 更高的性能。 本文就來(lái)探討一下 sync.Map 中的一些大家比較感興趣的問(wèn)題，比如為什么有了 map 還要 sync.Map？它為什么快？sync.Map 的適用場(chǎng)景（注意：不是所有情況下都快。）等。

關(guān)于 sync.Map 的設(shè)計(jì)與實(shí)現(xiàn)原理，會(huì)在下一篇中再做講解。

map 在并發(fā)下的問(wèn)題

如果我們看過(guò) map 的源碼，就會(huì)發(fā)現(xiàn)其中有不少會(huì)引起 fatal 錯(cuò)誤的地方，比如 mapaccess1（從 map 中讀取 key 的函數(shù)）里面，如果發(fā)現(xiàn)正在寫(xiě) map，則會(huì)有 fatal 錯(cuò)誤。 （如果還沒(méi)看過(guò)，可以跟著這篇 《go map 設(shè)計(jì)與實(shí)現(xiàn)》 看一下）

if h.flags&hashWriting != 0 {
    fatal("concurrent map read and map write")
}

map 并發(fā)讀寫(xiě)異常的例子

下面是一個(gè)實(shí)際使用中的例子：

var m = make(map[int]int)
// 往 map 寫(xiě) key 的協(xié)程
go func() {
   // 往 map 寫(xiě)入數(shù)據(jù)
    for i := 0; i < 10000; i++ {
        m[i] = i
    }
}()
// 從 map 讀取 key 的協(xié)程
go func() {
   // 從 map 讀取數(shù)據(jù)
    for i := 10000; i > 0; i-- {
        _ = m[i]
    }
}()
// 等待兩個(gè)協(xié)程執(zhí)行完畢
time.Sleep(time.Second)

這會(huì)導(dǎo)致報(bào)錯(cuò)：

fatal error: concurrent map read and map write

這是因?yàn)槲覀兺瑫r(shí)對(duì) map 進(jìn)行讀寫(xiě)，而 map 不支持并發(fā)讀寫(xiě)，所以會(huì)報(bào)錯(cuò)。如果 map 允許并發(fā)讀寫(xiě)，那么可能在我們使用的時(shí)候會(huì)有很多錯(cuò)亂的情況出現(xiàn)。 （具體如何錯(cuò)亂，我們可以對(duì)比多線程的場(chǎng)景思考一下，本文不展開(kāi)了）。

使用 sync.Mutex 保證并發(fā)安全

對(duì)于 map 并發(fā)讀寫(xiě)報(bào)錯(cuò)的問(wèn)題，其中一種解決方案就是使用 sync.Mutex 來(lái)保證并發(fā)安全， 但是這樣會(huì)導(dǎo)致我們?cè)谧x寫(xiě)的時(shí)候，都需要加鎖，這樣就會(huì)導(dǎo)致性能的下降。

使用 sync.Mutex 來(lái)保證并發(fā)安全，上面的代碼可以改成下面這樣：

var m = make(map[int]int)
// 互斥鎖
var mu sync.Mutex
// 寫(xiě) map 的協(xié)程
go func() {
    for i := 0; i < 10000; i++ {
        mu.Lock() // 寫(xiě) map，加互斥鎖
        m[i] = i
        mu.Unlock()
    }
}()
// 讀 map 的協(xié)程序
go func() {
    for i := 10000; i > 0; i-- {
        mu.Lock() // 讀 map，加互斥鎖
        _ = m[i]
        mu.Unlock()
    }
}()
time.Sleep(time.Second)

這樣就不會(huì)報(bào)錯(cuò)了，但是性能會(huì)有所下降，因?yàn)槲覀冊(cè)谧x寫(xiě)的時(shí)候都需要加鎖。（如果需要更高性能，可以繼續(xù)讀下去，不要急著使用 sync.Mutex）

sync.Mutex 的常見(jiàn)的用法是在結(jié)構(gòu)體中嵌入 sync.Mutex，而不是定義獨(dú)立的兩個(gè)變量。

使用 sync.RWMutex 保證并發(fā)安全

在上一小節(jié)中，我們使用了 sync.Mutex 來(lái)保證并發(fā)安全，但是在讀和寫(xiě)的時(shí)候我們都需要加互斥鎖。 這就意味著，就算多個(gè)協(xié)程進(jìn)行并發(fā)讀，也需要等待鎖。 但是互斥鎖的粒度太大了，但實(shí)際上，并發(fā)讀是沒(méi)有什么太大問(wèn)題的，應(yīng)該被允許才對(duì)，如果我們?cè)试S并發(fā)讀，那么就可以提高性能。

當(dāng)然 go 的開(kāi)發(fā)者也考慮到了這一點(diǎn)，所以在 sync 包中提供了 sync.RWMutex，這個(gè)鎖可以允許進(jìn)行并發(fā)讀，但是寫(xiě)的時(shí)候還是需要等待鎖。 也就是說(shuō)，一個(gè)協(xié)程在持有寫(xiě)鎖的時(shí)候，其他協(xié)程是既不能讀也不能寫(xiě)的，只能等待寫(xiě)鎖釋放才能進(jìn)行讀寫(xiě)。

使用 sync.RWMutex 來(lái)保證并發(fā)安全，我們可以改成下面這樣：

var m = make(map[int]int)
// 讀寫(xiě)鎖（允許并發(fā)讀，寫(xiě)的時(shí)候是互斥的）
var mu sync.RWMutex
// 寫(xiě)入 map 的協(xié)程
go func() {
    for i := 0; i < 10000; i++ {
        // 寫(xiě)入的時(shí)候需要加鎖
        mu.Lock()
        m[i] = i
        mu.Unlock()
    }
}()
// 讀取 map 的協(xié)程
go func() {
    for i := 10000; i > 0; i-- {
        // 讀取的時(shí)候需要加鎖，但是這個(gè)鎖是讀鎖
        // 多個(gè)協(xié)程可以同時(shí)使用 RLock 而不需要等待
        mu.RLock()
        _ = m[i]
        mu.RUnlock()
    }
}()
// 另外一個(gè)讀取 map 的協(xié)程
go func() {
    for i := 20000; i > 10000; i-- {
        // 讀取的時(shí)候需要加鎖，但是這個(gè)鎖是讀鎖
        // 多個(gè)協(xié)程可以同時(shí)使用 RLock 而不需要等待
        mu.RLock()
        _ = m[i]
        mu.RUnlock()
    }
}()
time.Sleep(time.Second)

這樣就不會(huì)報(bào)錯(cuò)了，而且性能也提高了，因?yàn)槲覀冊(cè)谧x的時(shí)候，不需要等待鎖。

說(shuō)明：

• 多個(gè)協(xié)程可以同時(shí)使用 RLock 而不需要等待，這是讀鎖。
• 只有一個(gè)協(xié)程可以使用 Lock，這是寫(xiě)鎖，有寫(xiě)鎖的時(shí)候，其他協(xié)程不能讀也不能寫(xiě)。
• 持有寫(xiě)鎖的協(xié)程，可以使用 Unlock 來(lái)釋放鎖。
• 寫(xiě)鎖釋放之后，其他協(xié)程才能獲取到鎖（讀鎖或者寫(xiě)鎖）。

也就是說(shuō)，使用 sync.RWMutex 的時(shí)候，讀操作是可以并發(fā)執(zhí)行的，但是寫(xiě)操作是互斥的。 這樣一來(lái)，相比 sync.Mutex 來(lái)說(shuō)等待鎖的次數(shù)就少了，自然也就能獲得更好的性能了。

gin 框架里面就使用了 sync.RWMutex 來(lái)保證 Keys 讀寫(xiě)操作的并發(fā)安全。

有了讀寫(xiě)鎖為什么還要有 sync.Map？

通過(guò)上面的內(nèi)容，我們知道了，有下面兩種方式可以保證并發(fā)安全：

• 使用 sync.Mutex，但是這樣的話，讀寫(xiě)都是互斥的，性能不好。
• 使用 sync.RWMutex，可以并發(fā)讀，但是寫(xiě)的時(shí)候是互斥的，性能相對(duì) sync.Mutex 要好一些。

但是就算我們使用了 sync.RWMutex，也還是有一些鎖的開(kāi)銷(xiāo)。那么我們能不能再優(yōu)化一下呢？答案是可以的。那就是使用 sync.Map。

sync.Map 在鎖的基礎(chǔ)上做了進(jìn)一步優(yōu)化，在一些場(chǎng)景下使用原子操作來(lái)保證并發(fā)安全，性能更好。

使用原子操作替代讀鎖

但是就算使用 sync.RWMutex，讀操作依然還有鎖的開(kāi)銷(xiāo)，那么有沒(méi)有更好的方式呢？ 答案是有的，就是使用原子操作來(lái)替代讀鎖。

舉一個(gè)很常見(jiàn)的例子就是多個(gè)協(xié)程同時(shí)讀取一個(gè)變量，然后對(duì)這個(gè)變量進(jìn)行累加操作：

var a int32
var wg sync.WaitGroup
wg.Add(2)
go func() {
    for i := 0; i < 10000; i++ {
        a++
    }
    wg.Done()
}()
go func() {
    for i := 0; i < 10000; i++ {
        a++
    }
    wg.Done()
}()
wg.Wait()
// a 期望結(jié)果應(yīng)該是 20000才對(duì)。
fmt.Println(a) // 實(shí)際：17089，而且每次都不一樣

這個(gè)例子中，我們期望的結(jié)果是 a 的值是 20000，但是實(shí)際上，每次運(yùn)行的結(jié)果都不一樣，而且都不會(huì)等于 20000。 其中很簡(jiǎn)單粗暴的一種解決方法是加鎖，但是這樣的話，性能就不好了，但是我們可以使用原子操作來(lái)解決這個(gè)問(wèn)題：

var a atomic.Int32
var wg sync.WaitGroup
wg.Add(2)
go func() {
    for i := 0; i < 10000; i++ {
        a.Add(1)
    }
    wg.Done()
}()
go func() {
    for i := 0; i < 10000; i++ {
        a.Add(1)
    }
    wg.Done()
}()
wg.Wait()
fmt.Println(a.Load()) // 20000

鎖跟原子操作的性能差多少？

我們來(lái)看一下，使用鎖和原子操作的性能差多少：

func BenchmarkMutexAdd(b *testing.B) {
   var a int32
   var mu sync.Mutex
   for i := 0; i < b.N; i++ {
      mu.Lock()
      a++
      mu.Unlock()
   }
}
func BenchmarkAtomicAdd(b *testing.B) {
   var a atomic.Int32
   for i := 0; i < b.N; i++ {
      a.Add(1)
   }
}

結(jié)果：

BenchmarkMutexAdd-12       100000000          10.07 ns/op
BenchmarkAtomicAdd-12      205196968           5.847 ns/op

我們可以看到，使用原子操作的性能比使用鎖的性能要好一些。

也許我們會(huì)覺(jué)得上面這個(gè)例子是寫(xiě)操作，那么讀操作呢？我們來(lái)看一下：

func BenchmarkMutex(b *testing.B) {
   var mu sync.RWMutex
   for i := 0; i < b.N; i++ {
      mu.RLock()
      mu.RUnlock()
   }
}
func BenchmarkAtomic(b *testing.B) {
   var a atomic.Int32
   for i := 0; i < b.N; i++ {
      _ = a.Load()
   }
}

結(jié)果：

BenchmarkMutex-12      100000000          10.12 ns/op
BenchmarkAtomic-12     1000000000          0.3133 ns/op

我們可以看到，使用原子操作的性能比使用鎖的性能要好很多。而且在 BenchmarkMutex 里面甚至還沒(méi)有做讀取數(shù)據(jù)的操作。

sync.Map 里面的原子操作

sync.Map 里面相比 sync.RWMutex，性能更好的原因就是使用了原子操作。 在我們從 sync.Map 里面讀取數(shù)據(jù)的時(shí)候，會(huì)先使用一個(gè)原子 Load 操作來(lái)讀取 sync.Map 里面的 key（從 read 中讀?。?注意：這里拿到的是 key 的一份快照，我們對(duì)其進(jìn)行讀操作的時(shí)候也可以同時(shí)往 sync.Map 中寫(xiě)入新的 key，這是保證它高性能的一個(gè)很關(guān)鍵的設(shè)計(jì)（類(lèi)似讀寫(xiě)分離）。

sync.Map 里面的 Load 方法里面就包含了上述的流程：

// Load 方法從 sync.Map 里面讀取數(shù)據(jù)。
func (m *Map) Load(key any) (value any, ok bool) {
   // 先從只讀 map 里面讀取數(shù)據(jù)。
   // 這一步是不需要鎖的，只有一個(gè)原子操作。
   read := m.loadReadOnly()
   e, ok := read.m[key]
   if !ok && read.amended { // 如果沒(méi)有找到，并且 dirty 里面有一些 read 中沒(méi)有的 key，那么就需要從 dirty 里面讀取數(shù)據(jù)。
      // 這里才需要鎖
      m.mu.Lock()
      read = m.loadReadOnly()
      e, ok = read.m[key]
      if !ok && read.amended {
         e, ok = m.dirty[key]
         m.missLocked()
      }
      m.mu.Unlock()
   }
   // key 不存在
   if !ok {
      return nil, false
   }
   // 使用原子操作讀取
   return e.Load()
}

上面的代碼我們可能還看不懂，但是沒(méi)關(guān)系，這里我們只需要知道的是，從 sync.Map 讀取數(shù)據(jù)的時(shí)候，會(huì)先做原子操作，如果沒(méi)找到，再進(jìn)行加鎖操作，這樣就減少了使用鎖的頻率了，自然也就可以獲得更好的性能（但要注意的是并不是所有情況下都能獲得更好的性能）。至于具體實(shí)現(xiàn)，在下一篇文章中會(huì)進(jìn)行更加詳細(xì)的分析。

也就是說(shuō)，sync.Map 之所以更快，是因?yàn)橄啾?RWMutex，進(jìn)一步減少了鎖的使用，而這也就是 sync.Map 存在的原因了

sync.Map 的基本用法

現(xiàn)在我們知道了，sync.Map 里面是利用了原子操作來(lái)減少鎖的使用。但是我們好像連 sync.Map 的一些基本操作都還不了解，現(xiàn)在就讓我們?cè)賮?lái)看看 sync.Map 的基本用法。

sync.Map 的使用還是挺簡(jiǎn)單的，map 中有的操作，在 sync.Map 都有，只不過(guò)區(qū)別是，在 sync.Map 中，所有的操作都需要通過(guò)調(diào)用其方法來(lái)進(jìn)行。 sync.Map 里面幾個(gè)常用的方法有（CRUD）：

• Store：我們新增或者修改數(shù)據(jù)的時(shí)候，都可以使用 Store 方法。
• Load：讀取數(shù)據(jù)的方法。
• Range：遍歷數(shù)據(jù)的方法。
• Delete：刪除數(shù)據(jù)的方法。

var m sync.Map
// 寫(xiě)入/修改
m.Store("foo", 1)
// 讀取
fmt.Println(m.Load("foo")) // 1 true
// 遍歷
m.Range(func(key, value interface{}) bool {
    fmt.Println(key, value) // foo 1
    return true
})
// 刪除
m.Delete("foo")
fmt.Println(m.Load("foo")) // nil false

注意：在 sync.Map 中，key 和 value 都是 interface{} 類(lèi)型的，也就是說(shuō)，我們可以使用任意類(lèi)型的 key 和 value。 而不像 map，只能存在一種類(lèi)型的 key 和 value。從這個(gè)角度來(lái)看，它的類(lèi)型類(lèi)似于 map[any]any。

另外一個(gè)需要注意的是，Range 方法的參數(shù)是一個(gè)函數(shù)，這個(gè)函數(shù)如果返回 false，那么遍歷就會(huì)停止。

sync.Map 的使用場(chǎng)景

在 sync.Map 源碼中，已經(jīng)告訴了我們 sync.Map 的使用場(chǎng)景：

The Map type is optimized for two common use cases: (1) when the entry for a given
key is only ever written once but read many times, as in caches that only grow,
or (2) when multiple goroutines read, write, and overwrite entries for disjoint
sets of keys. In these two cases, use of a Map may significantly reduce lock
contention compared to a Go map paired with a separate Mutex or RWMutex.

翻譯過(guò)來(lái)就是，Map 類(lèi)型針對(duì)兩種常見(jiàn)用例進(jìn)行了優(yōu)化：

• 當(dāng)給定 key 的條目只寫(xiě)入一次但讀取多次時(shí)，如在只會(huì)增長(zhǎng)的緩存中。（讀多寫(xiě)少）
• 當(dāng)多個(gè) goroutine 讀取、寫(xiě)入和覆蓋不相交的鍵集的條目。（不同 goroutine 操作不同的 key）

在這兩種情況下，與 Go map 與單獨(dú)的 Mutex 或 RWMutex 配對(duì)相比，使用 sync.Map 可以顯著減少鎖競(jìng)爭(zhēng)（很多時(shí)候只需要原子操作就可以）。

總結(jié)

普通的 map 不支持并發(fā)讀寫(xiě)。

有以下兩種方式可以實(shí)現(xiàn) map 的并發(fā)讀寫(xiě)：

• 使用 sync.Mutex 互斥鎖。讀和寫(xiě)的時(shí)候都使用互斥鎖，性能相比 sync.RWMutex 會(huì)差一些。
• 使用 sync.RWMutex 讀寫(xiě)鎖。讀的鎖是可以共享的，但是寫(xiě)鎖是獨(dú)占的。性能相比 sync.Mutex 會(huì)好一些。
• sync.Map 里面會(huì)先進(jìn)行原子操作來(lái)讀取 key，如果讀取不到的時(shí)候，才會(huì)需要加鎖。所以性能相比 sync.Mutex 和 sync.RWMutex 會(huì)好一些。

sync.Map 里面幾個(gè)常用的方法有（CRUD）：

• Store：我們新增或者修改數(shù)據(jù)的時(shí)候，都可以使用 Store 方法。
• Load：讀取數(shù)據(jù)的方法。
• Range：遍歷數(shù)據(jù)的方法。
• Delete：刪除數(shù)據(jù)的方法。

sync.Map 的使用場(chǎng)景，sync.Map 針對(duì)以下兩種場(chǎng)景做了優(yōu)化：

• key 只會(huì)寫(xiě)入一次，但是會(huì)被讀取多次的場(chǎng)景。
• 多個(gè) goroutine 讀取、寫(xiě)入和覆蓋不相交的鍵集的條目。

以上就是go sync.Map基本原理深入解析的詳細(xì)內(nèi)容，更多關(guān)于go sync.Map基本原理的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

最新評(píng)論