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

是什么

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

sync.map 是線程安全的，讀取，插入，刪除也都保持著常數(shù)級(jí)的時(shí)間復(fù)雜度。
sync.map 的零值是有效的，并且零值是一個(gè)空的 map。在第一次使用之后，不允許被拷貝。

有什么用

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

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

如何使用

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

package main

import (
	"fmt"
	"sync"
)

func main() {
	var m sync.Map
	// 1. 寫入
	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 步，寫入兩個(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)存在，因此寫入不成功，并且讀出原值。

程序輸出：

18
stefno 20
qcrao 18
nil> false
20

sync.map 適用于讀多寫少的場(chǎng)景。對(duì)于寫多的場(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 類型，可以并發(fā)地讀。但如果需要更新 read，則需要加鎖保護(hù)。對(duì)于 read 中存儲(chǔ)的 entry 字段，可能會(huì)被并發(fā)地 CAS 更新。但是如果要更新一個(gè)之前已被刪除的 entry，則需要先將其狀態(tài)從 expunged 改為 nil，再拷貝到 dirty 中，然后再更新。

dirty 是一個(gè)非線程安全的原始 map。包含新寫入的 key，并且包含 read 中的所有未被刪除的 key。這樣，可以快速地將 dirty 提升為 read 對(duì)外提供服務(wù)。如果 dirty 為 nil，那么下一次寫入時(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)有三種：

當(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)表示：

Store

先來(lái)看 expunged：

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

它是一個(gè)指向任意類型的指針，用來(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，直接寫入更新 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 寫入 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})
		}
		// 寫入新 key，在 dirty 中存儲(chǔ) value
		m.dirty[key] = newEntry(value)
	}
	m.mu.Unlock()
}

整體流程：

• 如果在 read 里能夠找到待存儲(chǔ)的 key，并且對(duì)應(yīng)的 entry 的 p 值不為 expunged，也就是沒(méi)被刪除時(shí)，直接更新對(duì)應(yīng)的 entry 即可。
• 第一步?jīng)]有成功：要么 read 中沒(méi)有這個(gè) key，要么 key 被標(biāo)記為刪除。則先加鎖，再進(jìn)行后續(xù)的操作。
• 再次在 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。
• 如果 read 中沒(méi)有此 key，那就查看 dirty 中是否有此 key，如果有，則直接更新對(duì)應(yīng)的 value，這時(shí) read 中還是沒(méi)有此 key。
• 最后一步，如果 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 寫入 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 不是線程安全的，所以需要加上互斥鎖
		// 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，整體流程如下：

• 首先是 fast path，直接在 read 中找，如果找到了直接調(diào)用 entry 的 load 方法，取出其中的值。
• 如果 read 中沒(méi)有這個(gè) key，且 amended 為 fase，說(shuō)明 dirty 為空，那直接返回 空和 false。
• 如果 read 中沒(méi)有這個(gè) key，且 amended 為 true，說(shuō)明 dirty 中可能存在我們要找的 key。當(dāng)然要先上鎖，再嘗試去 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 類似，都是先從 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ì)寫入 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)銷分?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 是線程安全的，讀取，插入，刪除也都保持著常數(shù)級(jí)的時(shí)間復(fù)雜度。
2. 通過(guò)讀寫分離，降低鎖時(shí)間來(lái)提高效率，適用于讀多寫少的場(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. 新寫入的 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ù)。

參考資料

【德志大佬-設(shè)計(jì)并發(fā)安全的 map】https://halfrost.com/go_map_chapter_one/

【德志大佬-設(shè)計(jì)并發(fā)安全的 map】https://halfrost.com/go_map_chapter_two/

【關(guān)于 sync.map 為什么沒(méi)有 len 方法的 issue】https://github.com/golang/go/issues/20680

【芮神增加了 len 方法】http://xiaorui.cc/archives/4972

【圖解 map 操作】https://wudaijun.com/2018/02/go-sync-map-implement/

【從一道面試題開(kāi)始】https://segmentfault.com/a/1190000018657984

【源碼分析】https://zhuanlan.zhihu.com/p/44585993

【行文通暢，流程圖清晰】https://juejin.im/post/5d36a7cbf265da1bb47da444

到此這篇關(guān)于深度解密 Go 語(yǔ)言中的 sync.map的文章就介紹到這了,更多相關(guān)Go 語(yǔ)言sync.map內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！