Go 1.9 sync.Map揭秘

目錄?[?]

有并發問題的map

Go 1.9之前的解決方案

sync.Map

Load

Store

Delete

Range

sync.Map的性能

其它

在Go 1.6之前， 內置的map類型是部分goroutine安全的，并發的讀沒有問題，并發的寫可能有問題。自go 1.6之后， 并發地讀寫map會報錯，這在一些知名的開源庫中都存在這個問題，所以go 1.9之前的解決方案是額外綁定一個鎖，封裝成一個新的struct或者單獨使用鎖都可以。

本文帶你深入到sync.Map的具體實現中，看看為了增加一個功能，代碼是如何變的復雜的,以及作者在實現sync.Map的一些思想。

有并發問題的map

官方的faq已經提到內建的map不是線程(goroutine)安全的。

首先，讓我們看一段并發讀寫的代碼,下列程序中一個goroutine一直讀，一個goroutine一只寫同一個鍵值，即即使讀寫的鍵不相同，而且map也沒有"擴容"等操作，代碼還是會報錯。

12345678910111213141516171819

package mainfunc main() { m := make(map[int]int) go func() { for { _ = m[1] } }() go func() { for { m[2] = 2 } }() select {}}

錯誤信息是:?fatal error: concurrent map read and map write。

如果你查看Go的源代碼:?hashmap_fast.go#L118,會看到讀的時候會檢查hashWriting標志， 如果有這個標志，就會報并發錯誤。

寫的時候會設置這個標志:?hashmap.go#L542

1	h.flags |= hashWriting

hashmap.go#L628設置完之后會取消這個標記。

當然，代碼中還有好幾處并發讀寫的檢查， 比如寫的時候也會檢查是不是有并發的寫，刪除鍵的時候類似寫，遍歷的時候并發讀寫問題等。

有時候，map的并發問題不是那么容易被發現, 你可以利用-race參數來檢查。

Go 1.9之前的解決方案

但是，很多時候，我們會并發地使用map對象，尤其是在一定規模的項目中，map總會保存goroutine共享的數據。在Go官方blog的Go maps in action一文中，提供了一種簡便的解決方案。

1234	var counter = struct{ sync.RWMutex m map[string]int}{m: make(map[string]int)}

它使用嵌入struct為map增加一個讀寫鎖。

讀數據的時候很方便的加鎖：

1234	counter.RLock()n := counter.m["some_key"]counter.RUnlock()fmt.Println("some_key:", n)

寫數據的時候:

123	counter.Lock()counter.m["some_key"]++counter.Unlock()

sync.Map

可以說，上面的解決方案相當簡潔，并且利用讀寫鎖而不是Mutex可以進一步減少讀寫的時候因為鎖帶來的性能。

但是，它在一些場景下也有問題，如果熟悉Java的同學，可以對比一下java的ConcurrentHashMap的實現，在map的數據非常大的情況下，一把鎖會導致大并發的客戶端共爭一把鎖，Java的解決方案是shard, 內部使用多個鎖，每個區間共享一把鎖，這樣減少了數據共享一把鎖帶來的性能影響，orcaman提供了這個思路的一個實現：?concurrent-map，他也詢問了Go相關的開發人員是否在Go中也實現這種方案，由于實現的復雜性，答案是Yes, we considered it.,但是除非有特別的性能提升和應用場景，否則沒有進一步的開發消息。

那么，在Go 1.9中sync.Map是怎么實現的呢？它是如何解決并發提升性能的呢？

sync.Map的實現有幾個優化點，這里先列出來，我們后面慢慢分析。

空間換時間。 通過冗余的兩個數據結構(read、dirty),實現加鎖對性能的影響。

使用只讀數據(read)，避免讀寫沖突。

動態調整，miss次數多了之后，將dirty數據提升為read。

double-checking。

延遲刪除。 刪除一個鍵值只是打標記，只有在提升dirty的時候才清理刪除的數據。

優先從read讀取、更新、刪除，因為對read的讀取不需要鎖。

下面我們介紹sync.Map的重點代碼，以便理解它的實現思想。

首先，我們看一下sync.Map的數據結構：

123456789101112131415161718

type Map struct { // 當涉及到dirty數據的操作的時候，需要使用這個鎖 mu Mutex // 一個只讀的數據結構，因為只讀，所以不會有讀寫沖突。 // 所以從這個數據中讀取總是安全的。 // 實際上，實際也會更新這個數據的entries,如果entry是未刪除的(unexpunged), 并不需要加鎖。如果entry已經被刪除了，需要加鎖，以便更新dirty數據。 read atomic.Value // readOnly // dirty數據包含當前的map包含的entries,它包含最新的entries(包括read中未刪除的數據,雖有冗余，但是提升dirty字段為read的時候非常快，不用一個一個的復制，而是直接將這個數據結構作為read字段的一部分),有些數據還可能沒有移動到read字段中。 // 對于dirty的操作哦需要加鎖，因為對它的操作可能會有讀寫競爭。 // 當dirty為空的時候， 比如初始化或者剛提升完，下一次的寫操作會復制read字段中未刪除的數據到這個數據中。 dirty map[interface{}]*entry // 當從Map中讀取entry的時候，如果read中不包含這個entry,會嘗試從dirty中讀取，這個時候會將misses加一， // 當misses累積到 dirty的長度的時候， 就會將dirty提升為read,避免從dirty中miss太多次。因為操作dirty需要加鎖。 misses int}

它的數據結構很簡單，值包含四個字段：read、mu、dirty、misses。

它使用了冗余的數據結構read、dirty。dirty中會包含read中為刪除的entries，新增加的entries會加入到dirty中。

read的數據結構是：

1234	type readOnly struct { m map[interface{}]*entry amended bool // 如果Map.dirty有些數據不在中的時候，這個值為true}

amended指明Map.dirty中有readOnly.m未包含的數據，所以如果從Map.read找不到數據的話，還要進一步到Map.dirty中查找。

對Map.read的修改是通過原子操作進行的。

雖然read和dirty有冗余數據，但這些數據是通過指針指向同一個數據，所以盡管Map的value會很大，但是冗余的空間占用還是有限的。

readOnly.m和Map.dirty存儲的值類型是*entry,它包含一個指針p, 指向用戶存儲的value值。

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

p有三種值：

• nil: entry已被刪除了，并且m.dirty為nil
• expunged: entry已被刪除了，并且m.dirty不為nil，而且這個entry不存在于m.dirty中
• 其它： entry是一個正常的值

以上是sync.Map的數據結構，下面我們重點看看Load、Store、Delete、Range這四個方法，其它輔助方法可以參考這四個方法來理解。

Load

加載方法，也就是提供一個鍵key,查找對應的值value,如果不存在，通過ok反映：

123456789101112131415161718192021222324252627

func (m *Map) Load(key interface{}) (value interface{}, ok bool) { // 1.首先從m.read中得到只讀readOnly,從它的map中查找，不需要加鎖 read, _ := m.read.Load().(readOnly) e, ok := read.m[key] // 2. 如果沒找到，并且m.dirty中有新數據，需要從m.dirty查找，這個時候需要加鎖 if !ok && read.amended { m.mu.Lock() // 雙檢查，避免加鎖的時候m.dirty提升為m.read,這個時候m.read可能被替換了。 read, _ = m.read.Load().(readOnly) e, ok = read.m[key] // 如果m.read中還是不存在，并且m.dirty中有新數據 if !ok && read.amended { // 從m.dirty查找 e, ok = m.dirty[key] // 不管m.dirty中存不存在，都將misses計數加一 // missLocked()中滿足條件后就會提升m.dirty m.missLocked() } m.mu.Unlock() } if !ok { return nil, false } return e.load()}

這里有兩個值的關注的地方。一個是首先從m.read中加載，不存在的情況下，并且m.dirty中有新數據，加鎖，然后從m.dirty中加載。

二是這里使用了雙檢查的處理，因為在下面的兩個語句中，這兩行語句并不是一個原子操作。

12	if !ok && read.amended { m.mu.Lock()

雖然第一句執行的時候條件滿足，但是在加鎖之前，m.dirty可能被提升為m.read,所以加鎖后還得再檢查m.read，后續的方法中都使用了這個方法。

雙檢查的技術Java程序員非常熟悉了，單例模式的實現之一就是利用雙檢查的技術。

可以看到，如果我們查詢的鍵值正好存在于m.read中，無須加鎖，直接返回，理論上性能優異。即使不存在于m.read中，經過miss幾次之后，m.dirty會被提升為m.read，又會從m.read中查找。所以對于更新／增加較少，加載存在的key很多的case,性能基本和無鎖的map類似。

下面看看m.dirty是如何被提升的。?missLocked方法中可能會將m.dirty提升。

123456789

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

上面的最后三行代碼就是提升m.dirty的，很簡單的將m.dirty作為readOnly的m字段，原子更新m.read。提升后m.dirty、m.misses重置， 并且m.read.amended為false。

Store

這個方法是更新或者新增一個entry。

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253

func (m *Map) Store(key, value interface{}) { // 如果m.read存在這個鍵，并且這個entry沒有被標記刪除，嘗試直接存儲。 // 因為m.dirty也指向這個entry,所以m.dirty也保持最新的entry。 read, _ := m.read.Load().(readOnly) if e, ok := read.m[key]; ok && e.tryStore(&value) { return } // 如果`m.read`不存在或者已經被標記刪除 m.mu.Lock() read, _ = m.read.Load().(readOnly) if e, ok := read.m[key]; ok { if e.unexpungeLocked() { //標記成未被刪除 m.dirty[key] = e //m.dirty中不存在這個鍵，所以加如m.dirty } e.storeLocked(&value) //更新 } else if e, ok := m.dirty[key]; ok { // m.dirty存在這個鍵，更新 e.storeLocked(&value) } else { //新鍵值 if !read.amended { //m.dirty中沒有新的數據，往m.dirty中增加第一個新鍵 m.dirtyLocked() //從m.read中復制未刪除的數據 m.read.Store(readOnly{m: read.m, amended: true}) } m.dirty[key] = newEntry(value) //將這個entry加入到m.dirty中 } m.mu.Unlock()}func (m *Map) dirtyLocked() { if m.dirty != nil { return } read, _ := m.read.Load().(readOnly) m.dirty = make(map[interface{}]*entry, len(read.m)) for k, e := range read.m { if !e.tryExpungeLocked() { m.dirty[k] = e } }}func (e *entry) tryExpungeLocked() (isExpunged bool) { p := atomic.LoadPointer(&e.p) for p == nil { // 將已經刪除標記為nil的數據標記為expunged if atomic.CompareAndSwapPointer(&e.p, nil, expunged) { return true } p = atomic.LoadPointer(&e.p) } return p == expunged}

你可以看到，以上操作都是先從操作m.read開始的，不滿足條件再加鎖，然后操作m.dirty。

Store可能會在某種情況下(初始化或者m.dirty剛被提升后)從m.read中復制數據，如果這個時候m.read中數據量非常大，可能會影響性能。

Delete

刪除一個鍵值。

12345678910111213141516

func (m *Map) Delete(key interface{}) { read, _ := m.read.Load().(readOnly) e, ok := read.m[key] 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) } m.mu.Unlock() } if ok { e.delete() }}

同樣，刪除操作還是從m.read中開始， 如果這個entry不存在于m.read中，并且m.dirty中有新數據，則加鎖嘗試從m.dirty中刪除。

注意，還是要雙檢查的。 從m.dirty中直接刪除即可，就當它沒存在過，但是如果是從m.read中刪除，并不會直接刪除，而是打標記：

12345678910111213

func (e *entry) delete() (hadValue bool) { for { p := atomic.LoadPointer(&e.p) // 已標記為刪除 if p == nil || p == expunged { return false } // 原子操作，e.p標記為nil if atomic.CompareAndSwapPointer(&e.p, p, nil) { return true } }}

Range

因為for ... range map是內建的語言特性，所以沒有辦法使用for range遍歷sync.Map, 但是可以使用它的Range方法，通過回調的方式遍歷。

12345678910111213141516171819202122232425262728

func (m *Map) Range(f func(key, value interface{}) bool) { read, _ := m.read.Load().(readOnly) // 如果m.dirty中有新數據，則提升m.dirty,然后在遍歷 if read.amended { //提升m.dirty 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 range是安全的 for k, e := range read.m { v, ok := e.load() if !ok { continue } if !f(k, v) { break } }}

Range方法調用前可能會做一個m.dirty的提升，不過提升m.dirty不是一個耗時的操作。

sync.Map的性能

Go 1.9源代碼中提供了性能的測試：?map_bench_test.go、map_reference_test.go

我也基于這些代碼修改了一下，得到下面的測試數據，相比較以前的解決方案，性能多少回有些提升，如果你特別關注性能，可以考慮sync.Map。

12345678910111213141516

BenchmarkHitAll/*sync.RWMutexMap-4 20000000 83.8 ns/opBenchmarkHitAll/*sync.Map-4 30000000 59.9 ns/opBenchmarkHitAll_WithoutPrompting/*sync.RWMutexMap-4 20000000 96.9 ns/opBenchmarkHitAll_WithoutPrompting/*sync.Map-4 20000000 64.1 ns/opBenchmarkHitNone/*sync.RWMutexMap-4 20000000 79.1 ns/opBenchmarkHitNone/*sync.Map-4 30000000 43.3 ns/opBenchmarkHit_WithoutPrompting/*sync.RWMutexMap-4 20000000 81.5 ns/opBenchmarkHit_WithoutPrompting/*sync.Map-4 30000000 44.0 ns/opBenchmarkUpdate/*sync.RWMutexMap-4 5000000 328 ns/opBenchmarkUpdate/*sync.Map-4 10000000 146 ns/opBenchmarkUpdate_WithoutPrompting/*sync.RWMutexMap-4 5000000 336 ns/opBenchmarkUpdate_WithoutPrompting/*sync.Map-4 5000000 324 ns/opBenchmarkDelete/*sync.RWMutexMap-4 10000000 155 ns/opBenchmarkDelete/*sync.Map-4 30000000 55.0 ns/opBenchmarkDelete_WithoutPrompting/*sync.RWMutexMap-4 10000000 173 ns/opBenchmarkDelete_WithoutPrompting/*sync.Map-4 10000000 147 ns/op

其它

sync.Map沒有Len方法，并且目前沒有跡象要加上 (issue#20680),所以如果想得到當前Map中有效的entries的數量，需要使用Range方法遍歷一次， 比較X疼。

LoadOrStore方法如果提供的key存在，則返回已存在的值(Load)，否則保存提供的鍵值(Store)。

來源：?http://colobu.com/2017/07/11/dive-into-sync-Map/

轉載于:https://www.cnblogs.com/zhangboyu/p/7456687.html

總結

以上是生活随笔為你收集整理的Go 1.9 sync.Map揭秘的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。