工作中，經(jīng)常會(huì)碰到并發(fā)讀寫(xiě) map 而造成 panic 的情況，為什么在并發(fā)讀寫(xiě)的時(shí)候，會(huì) panic 呢？因?yàn)樵诓l(fā)讀寫(xiě)的情況下，map 里的數(shù)據(jù)會(huì)被寫(xiě)亂，之后就是?Garbage in, garbage out，還不如直接 panic 了。

本文目錄如下：

是什么

Go 語(yǔ)言原生 map 并不是線(xiàn)程安全的，對(duì)它進(jìn)行并發(fā)讀寫(xiě)操作的時(shí)候，需要加鎖。而?sync.map?則是一種并發(fā)安全的 map，在 Go 1.9 引入。

sync.map?是線(xiàn)程安全的，讀取，插入，刪除也都保持著常數(shù)級(jí)的時(shí)間復(fù)雜度。

sync.map?的零值是有效的，并且零值是一個(gè)空的 map。在第一次使用之后，不允許被拷貝。

有什么用

一般情況下解決并發(fā)讀寫(xiě) map 的思路是加一把大鎖，或者把一個(gè) map 分成若干個(gè)小 map，對(duì) key 進(jìn)行哈希，只操作相應(yīng)的小 map。前者鎖的粒度比較大，影響效率；后者實(shí)現(xiàn)起來(lái)比較復(fù)雜，容易出錯(cuò)。

而使用?sync.map?之后，對(duì) map 的讀寫(xiě)，不需要加鎖。并且它通過(guò)空間換時(shí)間的方式，使用 read 和 dirty 兩個(gè) map 來(lái)進(jìn)行讀寫(xiě)分離，降低鎖時(shí)間來(lái)提高效率。

如何使用

使用非常簡(jiǎn)單，和普通 map 相比，僅遍歷的方式略有區(qū)別：

package?main

import?(
?"fmt"
?"sync"
)

func?main()??{
?var?m?sync.Map
?//?1.?寫(xiě)入
?m.Store("qcrao",?18)
?m.Store("stefno",?20)

?//?2.?讀取
?age,?_?:=?m.Load("qcrao")
?fmt.Println(age.(int))

?//?3.?遍歷
?m.Range(func(key,?value?interface{})?bool?{
??name?:=?key.(string)
??age?:=?value.(int)
??fmt.Println(name,?age)
??return?true
?})

?//?4.?刪除
?m.Delete("qcrao")
?age,?ok?:=?m.Load("qcrao")
?fmt.Println(age,?ok)

?//?5.?讀取或?qū)懭?
?m.LoadOrStore("stefno",?100)
?age,?_?=?m.Load("stefno")
?fmt.Println(age)
}

第 1 步，寫(xiě)入兩個(gè) k-v 對(duì)；

第 2 步，使用 Load 方法讀取其中的一個(gè) key；

第 3 步，遍歷所有的 k-v 對(duì)，并打印出來(lái)；

第 4 步，刪除其中的一個(gè) key，再讀這個(gè) key，得到的就是 nil；

第 5 步，使用 LoadOrStore，嘗試讀取或?qū)懭?"Stefno"，因?yàn)檫@個(gè) key 已經(jīng)存在，因此寫(xiě)入不成功，并且讀出原值。

程序輸出：

18
stefno?20
qcrao?18
<nil>?false
20

sync.map?適用于讀多寫(xiě)少的場(chǎng)景。對(duì)于寫(xiě)多的場(chǎng)景，會(huì)導(dǎo)致 read map 緩存失效，需要加鎖，導(dǎo)致沖突變多；而且由于未命中 read map 次數(shù)過(guò)多，導(dǎo)致 dirty map 提升為 read map，這是一個(gè) O(N) 的操作，會(huì)進(jìn)一步降低性能。

源碼分析

數(shù)據(jù)結(jié)構(gòu)

先來(lái)看下 map 的數(shù)據(jù)結(jié)構(gòu)。去掉大段的注釋后：

type?Map?struct?{
?mu?Mutex
?read?atomic.Value?//?readOnly
?dirty?map[interface{}]*entry
?misses?int
}

互斥量?mu?保護(hù) read 和 dirty。

read?是 atomic.Value 類(lèi)型，可以并發(fā)地讀。但如果需要更新?read，則需要加鎖保護(hù)。對(duì)于 read 中存儲(chǔ)的 entry 字段，可能會(huì)被并發(fā)地 CAS 更新。但是如果要更新一個(gè)之前已被刪除的 entry，則需要先將其狀態(tài)從 expunged 改為 nil，再拷貝到 dirty 中，然后再更新。

dirty?是一個(gè)非線(xiàn)程安全的原始 map。包含新寫(xiě)入的 key，并且包含?read?中的所有未被刪除的 key。這樣，可以快速地將?dirty?提升為?read?對(duì)外提供服務(wù)。如果?dirty?為 nil，那么下一次寫(xiě)入時(shí)，會(huì)新建一個(gè)新的?dirty，這個(gè)初始的?dirty?是?read?的一個(gè)拷貝，但除掉了其中已被刪除的 key。

每當(dāng)從 read 中讀取失敗，都會(huì)將?misses?的計(jì)數(shù)值加 1，當(dāng)加到一定閾值以后，需要將 dirty 提升為 read，以期減少 miss 的情形。

read map?和?dirty map?的存儲(chǔ)方式是不一致的。

前者使用 atomic.Value，后者只是單純的使用 map。

原因是 read map 使用 lock free 操作，必須保證 load/store 的原子性；而 dirty map 的 load+store 操作是由 lock（就是 mu）來(lái)保護(hù)的。

真正存儲(chǔ)?key/value?的是 read 和 dirty 字段。read?使用 atomic.Value，這是 lock-free 的基礎(chǔ)，保證 load/store 的原子性。dirty?則直接用了一個(gè)原始的 map，對(duì)于它的 load/store 操作需要加鎖。

read?字段里實(shí)際上是存儲(chǔ)的是：

//?readOnly?is?an?immutable?struct?stored?atomically?in?the?Map.read?field.
type?readOnly?struct?{
?m???????map[interface{}]*entry
?amended?bool?//?true?if?the?dirty?map?contains?some?key?not?in?m.
}

注意到 read 和 dirty 里存儲(chǔ)的東西都包含?entry，來(lái)看一下：

type?entry?struct?{
?p?unsafe.Pointer?//?*interface{}
}

很簡(jiǎn)單，它是一個(gè)指針，指向 value?？磥?lái)，read 和 dirty 各自維護(hù)一套 key，key 指向的都是同一個(gè) value。也就是說(shuō)，只要修改了這個(gè) entry，對(duì) read 和 dirty 都是可見(jiàn)的。這個(gè)指針的狀態(tài)有三種：

p 的三種狀態(tài)

當(dāng)?p == nil?時(shí)，說(shuō)明這個(gè)鍵值對(duì)已被刪除，并且 m.dirty == nil，或 m.dirty[k] 指向該 entry。

當(dāng)?p == expunged?時(shí)，說(shuō)明這條鍵值對(duì)已被刪除，并且 m.dirty != nil，且 m.dirty 中沒(méi)有這個(gè) key。

其他情況，p 指向一個(gè)正常的值，表示實(shí)際?interface{}?的地址，并且被記錄在 m.read.m[key] 中。如果這時(shí) m.dirty 不為 nil，那么它也被記錄在 m.dirty[key] 中。兩者實(shí)際上指向的是同一個(gè)值。

當(dāng)刪除 key 時(shí)，并不實(shí)際刪除。一個(gè) entry 可以通過(guò)原子地（CAS 操作）設(shè)置 p 為 nil 被刪除。如果之后創(chuàng)建 m.dirty，nil 又會(huì)被原子地設(shè)置為 expunged，且不會(huì)拷貝到 dirty 中。

如果 p 不為 expunged，和 entry 相關(guān)聯(lián)的這個(gè) value 可以被原子地更新；如果?p == expunged，那么僅當(dāng)它初次被設(shè)置到 m.dirty 之后，才可以被更新。

整體用一張圖來(lái)表示：

sync.map 整體結(jié)構(gòu)

Store

先來(lái)看 expunged：

var?expunged?=?unsafe.Pointer(new(interface{}))

它是一個(gè)指向任意類(lèi)型的指針，用來(lái)標(biāo)記從 dirty map 中刪除的 entry。

//?Store?sets?the?value?for?a?key.
func?(m?*Map)?Store(key,?value?interface{})?{
?//?如果?read?map?中存在該?key??則嘗試直接更改(由于修改的是?entry?內(nèi)部的?pointer，因此?dirty?map?也可見(jiàn))
?read,?_?:=?m.read.Load().(readOnly)
?if?e,?ok?:=?read.m[key];?ok?&&?e.tryStore(&value)?{
??return
?}

?m.mu.Lock()
?read,?_?=?m.read.Load().(readOnly)
?if?e,?ok?:=?read.m[key];?ok?{
??if?e.unexpungeLocked()?{
???//?如果?read?map?中存在該?key，但?p?==?expunged，則說(shuō)明?m.dirty?!=?nil?并且?m.dirty?中不存在該?key?值?此時(shí):
???//????a.?將?p?的狀態(tài)由?expunged??更改為?nil
???//????b.?dirty?map?插入?key
???m.dirty[key]?=?e
??}
??//?更新?entry.p?=?value?(read?map?和?dirty?map?指向同一個(gè)?entry)
??e.storeLocked(&value)
?}?else?if?e,?ok?:=?m.dirty[key];?ok?{
??//?如果?read?map?中不存在該?key，但?dirty?map?中存在該?key，直接寫(xiě)入更新?entry(read?map?中仍然沒(méi)有這個(gè)?key)
??e.storeLocked(&value)
?}?else?{
??//?如果 read map 和 dirty map 中都不存在該 key，則：
??//???a.?如果?dirty?map?為空，則需要?jiǎng)?chuàng)建?dirty?map，并從?read?map?中拷貝未刪除的元素到新創(chuàng)建的?dirty?map
??//????b.?更新?amended?字段，標(biāo)識(shí)?dirty?map?中存在?read?map?中沒(méi)有的?key
??//????c.?將?kv?寫(xiě)入?dirty?map?中，read?不變
??if?!read.amended?{
??????//?到這里就意味著，當(dāng)前的 key 是第一次被加到 dirty map 中。
???// store 之前先判斷一下 dirty map 是否為空，如果為空，就把 read map 淺拷貝一次。
???m.dirtyLocked()
???m.read.Store(readOnly{m:?read.m,?amended:?true})
??}
??//?寫(xiě)入新?key，在?dirty?中存儲(chǔ)?value
??m.dirty[key]?=?newEntry(value)
?}
?m.mu.Unlock()
}

整體流程：

1. 如果在 read 里能夠找到待存儲(chǔ)的 key，并且對(duì)應(yīng)的 entry 的 p 值不為 expunged，也就是沒(méi)被刪除時(shí)，直接更新對(duì)應(yīng)的 entry 即可。
2. 第一步?jīng)]有成功：要么 read 中沒(méi)有這個(gè) key，要么 key 被標(biāo)記為刪除。則先加鎖，再進(jìn)行后續(xù)的操作。
3. 再次在 read 中查找是否存在這個(gè) key，也就是 double check 一下，這也是 lock-free 編程里的常見(jiàn)套路。如果 read 中存在該 key，但?p == expunged，說(shuō)明 m.dirty != nil 并且 m.dirty 中不存在該 key 值 此時(shí): a. 將 p 的狀態(tài)由 expunged ?更改為 nil；b. dirty map 插入 key。然后，直接更新對(duì)應(yīng)的 value。
4. 如果 read 中沒(méi)有此 key，那就查看 dirty 中是否有此 key，如果有，則直接更新對(duì)應(yīng)的 value，這時(shí) read 中還是沒(méi)有此 key。
5. 最后一步，如果 read 和 dirty 中都不存在該 key，則：a. 如果 dirty 為空，則需要?jiǎng)?chuàng)建 dirty，并從 read 中拷貝未被刪除的元素；b. 更新 amended 字段，標(biāo)識(shí) dirty map 中存在 read map 中沒(méi)有的 key；c. 將 k-v 寫(xiě)入 dirty map 中，read.m 不變。最后，更新此 key 對(duì)應(yīng)的 value。

再來(lái)看一些子函數(shù)：

//?如果 entry 沒(méi)被刪，tryStore 存儲(chǔ)值到 entry 中。如果 p == expunged，即 entry 被刪，那么返回 false。
func?(e?*entry)?tryStore(i?*interface{})?bool?{
?for?{
??p?:=?atomic.LoadPointer(&e.p)
??if?p?==?expunged?{
???return?false
??}
??if?atomic.CompareAndSwapPointer(&e.p,?p,?unsafe.Pointer(i))?{
???return?true
??}
?}
}

tryStore?在 Store 函數(shù)最開(kāi)始的時(shí)候就會(huì)調(diào)用，是比較常見(jiàn)的?for?循環(huán)加 CAS 操作，嘗試更新 entry，讓 p 指向新的值。

unexpungeLocked?函數(shù)確保了 entry 沒(méi)有被標(biāo)記成已被清除：

// unexpungeLocked 函數(shù)確保了 entry 沒(méi)有被標(biāo)記成已被清除。
//?如果?entry?先前被清除過(guò)了，那么在?mutex?解鎖之前，它一定要被加入到?dirty?map?中
func?(e?*entry)?unexpungeLocked()?(wasExpunged?bool)?{
?return?atomic.CompareAndSwapPointer(&e.p,?expunged,?nil)
}

Load

func?(m?*Map)?Load(key?interface{})?(value?interface{},?ok?bool)?{
?read,?_?:=?m.read.Load().(readOnly)
?e,?ok?:=?read.m[key]
?//?如果沒(méi)在?read?中找到，并且?amended?為?true，即?dirty?中存在?read?中沒(méi)有的?key
?if?!ok?&&?read.amended?{
??m.mu.Lock()?//?dirty?map?不是線(xiàn)程安全的，所以需要加上互斥鎖
??// double check。避免在上鎖的過(guò)程中 dirty map 提升為 read map。
??read,?_?=?m.read.Load().(readOnly)
??e,?ok?=?read.m[key]
??//?仍然沒(méi)有在?read?中找到這個(gè)?key，并且?amended?為?true
??if?!ok?&&?read.amended?{
???e,?ok?=?m.dirty[key]?//?從?dirty?中找
???//?不管?dirty?中有沒(méi)有找到，都要"記一筆"，因?yàn)樵?dirty?提升為?read?之前，都會(huì)進(jìn)入這條路徑
???m.missLocked()
??}
??m.mu.Unlock()
?}
?if?!ok?{?//?如果沒(méi)找到，返回空，false
??return?nil,?false
?}
?return?e.load()
}

處理路徑分為 fast path 和 slow path，整體流程如下：

1. 首先是 fast path，直接在 read 中找，如果找到了直接調(diào)用 entry 的 load 方法，取出其中的值。
2. 如果 read 中沒(méi)有這個(gè) key，且 amended 為 fase，說(shuō)明 dirty 為空，那直接返回 空和 false。
3. 如果 read 中沒(méi)有這個(gè) key，且 amended 為 true，說(shuō)明 dirty 中可能存在我們要找的 key。當(dāng)然要先上鎖，再?lài)L試去 dirty 中查找。在這之前，仍然有一個(gè) double check 的操作。若還是沒(méi)有在 read 中找到，那么就從 dirty 中找。不管 dirty 中有沒(méi)有找到，都要"記一筆"，因?yàn)樵?dirty 被提升為 read 之前，都會(huì)進(jìn)入這條路徑

這里主要看下?missLocked?的函數(shù)的實(shí)現(xiàn)：

func?(m?*Map)?missLocked()?{
?m.misses++
?if?m.misses?<?len(m.dirty)?{
??return
?}
?//?dirty?map?晉升
?m.read.Store(readOnly{m:?m.dirty})
?m.dirty?=?nil
?m.misses?=?0
}

直接將 misses 的值加 1，表示一次未命中，如果 misses 值小于 m.dirty 的長(zhǎng)度，就直接返回。否則，將 m.dirty 晉升為 read，并清空 dirty，清空 misses 計(jì)數(shù)值。這樣，之前一段時(shí)間新加入的 key 都會(huì)進(jìn)入到 read 中，從而能夠提升 read 的命中率。

再來(lái)看下 entry 的 load 方法：

func?(e?*entry)?load()?(value?interface{},?ok?bool)?{
?p?:=?atomic.LoadPointer(&e.p)
?if?p?==?nil?||?p?==?expunged?{
??return?nil,?false
?}
?return?*(*interface{})(p),?true
}

對(duì)于 nil 和 expunged 狀態(tài)的 entry，直接返回?ok=false；否則，將 p 轉(zhuǎn)成?interface{}?返回。

Delete

//?Delete?deletes?the?value?for?a?key.
func?(m?*Map)?Delete(key?interface{})?{
?read,?_?:=?m.read.Load().(readOnly)
?e,?ok?:=?read.m[key]
?//?如果?read?中沒(méi)有這個(gè)?key，且?dirty?map?不為空
?if?!ok?&&?read.amended?{
??m.mu.Lock()
??read,?_?=?m.read.Load().(readOnly)
??e,?ok?=?read.m[key]
??if?!ok?&&?read.amended?{
???delete(m.dirty,?key)?//?直接從?dirty?中刪除這個(gè)?key
??}
??m.mu.Unlock()
?}
?if?ok?{
??e.delete()?//?如果在?read?中找到了這個(gè)?key，將?p?置為?nil
?}
}

可以看到，基本套路還是和 Load，Store 類(lèi)似，都是先從 read 里查是否有這個(gè) key，如果有則執(zhí)行?entry.delete?方法，將 p 置為 nil，這樣 read 和 dirty 都能看到這個(gè)變化。

如果沒(méi)在 read 中找到這個(gè) key，并且 dirty 不為空，那么就要操作 dirty 了，操作之前，還是要先上鎖。然后進(jìn)行 double check，如果仍然沒(méi)有在 read 里找到此 key，則從 dirty 中刪掉這個(gè) key。但不是真正地從 dirty 中刪除，而是更新 entry 的狀態(tài)。

來(lái)看下?entry.delete?方法：

func?(e?*entry)?delete()?(hadValue?bool)?{
?for?{
??p?:=?atomic.LoadPointer(&e.p)
??if?p?==?nil?||?p?==?expunged?{
???return?false
??}
??if?atomic.CompareAndSwapPointer(&e.p,?p,?nil)?{
???return?true
??}
?}
}

它真正做的事情是將正常狀態(tài)（指向一個(gè) interface{}）的 p 設(shè)置成 nil。沒(méi)有設(shè)置成 expunged 的原因是，當(dāng) p 為 expunged 時(shí)，表示它已經(jīng)不在 dirty 中了。這是 p 的狀態(tài)機(jī)決定的，在?tryExpungeLocked?函數(shù)中，會(huì)將 nil 原子地設(shè)置成 expunged。

tryExpungeLocked?是在新創(chuàng)建 dirty 時(shí)調(diào)用的，會(huì)將已被刪除的 entry.p 從 nil 改成 expunged，這個(gè) entry 就不會(huì)寫(xiě)入 dirty 了。

func?(e?*entry)?tryExpungeLocked()?(isExpunged?bool)?{
?p?:=?atomic.LoadPointer(&e.p)
?for?p?==?nil?{
??//?如果原來(lái)是 nil，說(shuō)明原 key 已被刪除，則將其轉(zhuǎn)為 expunged。
??if?atomic.CompareAndSwapPointer(&e.p,?nil,?expunged)?{
???return?true
??}
??p?=?atomic.LoadPointer(&e.p)
?}
?return?p?==?expunged
}

注意到如果 key 同時(shí)存在于 read 和 dirty 中時(shí)，刪除只是做了一個(gè)標(biāo)記，將 p 置為 nil；而如果僅在 dirty 中含有這個(gè) key 時(shí)，會(huì)直接刪除這個(gè) key。原因在于，若兩者都存在這個(gè) key，僅做標(biāo)記刪除，可以在下次查找這個(gè) key 時(shí)，命中 read，提升效率。若只有在 dirty 中存在時(shí)，read 起不到“緩存”的作用，直接刪除。

LoadOrStore

這個(gè)函數(shù)結(jié)合了 Load 和 Store 的功能，如果 map 中存在這個(gè) key，那么返回這個(gè) key 對(duì)應(yīng)的 value；否則，將 key-value 存入 map。這在需要先執(zhí)行 Load 查看某個(gè) key 是否存在，之后再更新此 key 對(duì)應(yīng)的 value 時(shí)很有效，因?yàn)?LoadOrStore 可以并發(fā)執(zhí)行。

具體的過(guò)程不再一一分析了，可參考 Load 和 Store 的源碼分析。

Range

Range 的參數(shù)是一個(gè)函數(shù)：

f?func(key,?value?interface{})?bool

由使用者提供實(shí)現(xiàn)，Range 將遍歷調(diào)用時(shí)刻 map 中的所有 k-v 對(duì)，將它們傳給 f 函數(shù)，如果 f 返回 false，將停止遍歷。

func?(m?*Map)?Range(f?func(key,?value?interface{})?bool)?{
?read,?_?:=?m.read.Load().(readOnly)
?if?read.amended?{
??m.mu.Lock()
??read,?_?=?m.read.Load().(readOnly)
??if?read.amended?{
???read?=?readOnly{m:?m.dirty}
???m.read.Store(read)
???m.dirty?=?nil
???m.misses?=?0
??}
??m.mu.Unlock()
?}

?for?k,?e?:=?range?read.m?{
??v,?ok?:=?e.load()
??if?!ok?{
???continue
??}
??if?!f(k,?v)?{
???break
??}
?}
}

當(dāng) amended 為 true 時(shí)，說(shuō)明 dirty 中含有 read 中沒(méi)有的 key，因?yàn)?Range 會(huì)遍歷所有的 key，是一個(gè) O(n) 操作。將 dirty 提升為 read，會(huì)將開(kāi)銷(xiāo)分?jǐn)傞_(kāi)來(lái)，所以這里直接就提升了。

之后，遍歷 read，取出 entry 中的值，調(diào)用 f(k, v)。

其他

關(guān)于為何?sync.map?沒(méi)有 Len 方法，參考資料里給出了 issue，bcmills?認(rèn)為對(duì)于并發(fā)的數(shù)據(jù)結(jié)構(gòu)和非并發(fā)的數(shù)據(jù)結(jié)構(gòu)并不一定要有相同的方法。例如，map 有 Len 方法，sync.map 卻不一定要有。就像 sync.map 有 LoadOrStore 方法，map 就沒(méi)有一樣。

有些實(shí)現(xiàn)增加了一個(gè)計(jì)數(shù)器，并原子地增加或減少它，以此來(lái)表示 sync.map 中元素的個(gè)數(shù)。但?bcmills?提出這會(huì)引入競(jìng)爭(zhēng)：atomic?并不是?contention-free?的，它只是把競(jìng)爭(zhēng)下沉到了 CPU 層級(jí)。這會(huì)給其他不需要 Len 方法的場(chǎng)景帶來(lái)負(fù)擔(dān)。

總結(jié)

1. sync.map?是線(xiàn)程安全的，讀取，插入，刪除也都保持著常數(shù)級(jí)的時(shí)間復(fù)雜度。
2. 通過(guò)讀寫(xiě)分離，降低鎖時(shí)間來(lái)提高效率，適用于讀多寫(xiě)少的場(chǎng)景。
3. Range 操作需要提供一個(gè)函數(shù)，參數(shù)是?k,v，返回值是一個(gè)布爾值：f func(key, value interface{}) bool。
4. 調(diào)用 Load 或 LoadOrStore 函數(shù)時(shí)，如果在 read 中沒(méi)有找到 key，則會(huì)將 misses 值原子地增加 1，當(dāng) misses 增加到和 dirty 的長(zhǎng)度相等時(shí)，會(huì)將 dirty 提升為 read。以期減少“讀 miss”。
5. 新寫(xiě)入的 key 會(huì)保存到 dirty 中，如果這時(shí) dirty 為 nil，就會(huì)先新創(chuàng)建一個(gè) dirty，并將 read 中未被刪除的元素拷貝到 dirty。
6. 當(dāng) dirty 為 nil 的時(shí)候，read 就代表 map 所有的數(shù)據(jù)；當(dāng) dirty 不為 nil 的時(shí)候，dirty 才代表 map 所有的數(shù)據(jù)。

   原文鏈接：https://mp.weixin.qq.com/s/mXOU8TElP8bbqaybRKN8eA