5.7 并發安全散列表 · golan基礎

# 5.7 并發安全散列表語言內建的散列表`map`結構并不是并發安全的，在并發的對該結構進行讀寫時，甚至可能產生不可恢復的運行時恐慌，導致程序將不受控制的崩潰，這在對于可用性有一定要求的 Web 場景下，這幾乎是一個致命的缺陷。為了解決并發問題，`sync`包中提供了一種特殊的并發安全散列表`sync.Map`結構。該結構的使用文檔說明了該結構的實現針對了讀多寫少的這一特殊的場景進行優化，并得到了優于普通的`map`加`sync.Mutex`結構的性能，例如： ``` // MutexMap 是一個簡單的 map + sync.Mutex 的并發安全散列表實現 type MutexMap struct { data map[interface{}]interface{} mu sync.Mutex } func (m *MutexMap) Load(k interface{}) (v interface{}, ok bool) { m.mu.Lock() defer m.mu.Unlock() v, ok = m.data[k] return } func (m *MutexMap) Store(k, v interface{}) { m.mu.Lock() defer m.mu.Unlock() m.data[k] = v } ``` ## 5.7.1 sync.Map 的性能在討論`sync.Map`的設計之前我們不妨先通過基準測試來對其主要使用場景進行一個初步的認識。正如前面所說，`sync.Map`的主要使用場景是「讀多寫少」，但即便是對于「讀多寫少」的場景，我們也需要對不同的情況進行討論。作為拋磚引玉，這里我們只針對這一種場景進行基準測試，讀者也可以自行據此舉一反三： **對于一個給定的 sync.Map，并發場景下存在 1 個 goroutine 對某個 key 進行并發寫，并同時存在 n-1 個 goroutine 對同一個 key產生并發讀。** 根據這個描述，我們不難寫出如下的基準測試： ``` func BenchmarkLoadStoreCollision(b *testing.B) { ms := [...]mapInterface{ &MutexMap{data: map[interface{}]interface{}{}}, &RWMutexMap{data: map[interface{}]interface{}{}}, &sync.Map{}, } // 測試對于同一個 key 的 n-1 并發讀和 1 并發寫的性能 for _, m := range ms { b.Run(fmt.Sprintf("%T", m), func(b *testing.B) { var i int64 b.RunParallel(func(pb *testing.PB) { // 記錄并發執行的 goroutine id gid := int(atomic.AddInt64(&i, 1) - 1) if gid == 0 { // gid 為 0 的 goroutine 負責并發寫 for i := 0; pb.Next(); i++ { m.Store(0, i) } } else { // gid 不為 0 的 goroutine 負責并發讀 for pb.Next() { m.Load(0) } } }) }) } } ``` 其中`mapInterface`用于讓基準測試可以針對具有相同操作不同實現的散列表進行循環處理；而`RWMutexMap`僅僅只是將`mu`字段從`sync.Mutex`改為了`sync.RWMutex`： ``` type mapInterface interface { Load(k interface{}) (v interface{}, ok bool) Store(k, v interface{}) } // RWMutexMap 是一個簡單的 map + sync.RWMutex 的并發安全散列表實現 type RWMutexMap struct { data map[interface{}]interface{} mu sync.RWMutex } func (m *RWMutexMap) Load(k interface{}) (v interface{}, ok bool) { /* ...實現相同... */ } func (m *RWMutexMap) Store(k, v interface{}) { /* ...實現相同... */ } ``` 在 Go 的基準測試工具中，`b.RunParallel`將執行 GOMAXPROCS 個并發的 Goroutine，為了測試隨并發 Goroutine 數量增多情況下散列表的性能變化情況，可以通過`-cpu`參數來動態調整 GOMAXPROCS 的數量，于是： ``` $ go test -v -run=none -bench=. -count=10 -cpu=2,4,8,16,32,64,128,256,512 | tee bench.txt ``` 運行完畢后，再使用 benchstat 工具對測試結果從統計意義上消除系統誤差： ``` $ benchstat bench.txt name time/op LoadStoreCollision/*map_test.MutexMap-2 42.7ns ± 1% LoadStoreCollision/*map_test.MutexMap-4 60.0ns ± 2% LoadStoreCollision/*map_test.MutexMap-8 76.5ns ± 5% LoadStoreCollision/*map_test.MutexMap-16 90.7ns ± 3% LoadStoreCollision/*map_test.MutexMap-32 94.2ns ± 6% LoadStoreCollision/*map_test.MutexMap-64 109ns ± 1% LoadStoreCollision/*map_test.MutexMap-128 134ns ±14% LoadStoreCollision/*map_test.MutexMap-256 175ns ±16% LoadStoreCollision/*map_test.MutexMap-512 223ns ± 6% LoadStoreCollision/*map_test.RWMutexMap-2 123ns ± 8% LoadStoreCollision/*map_test.RWMutexMap-4 140ns ± 8% LoadStoreCollision/*map_test.RWMutexMap-8 141ns ± 4% LoadStoreCollision/*map_test.RWMutexMap-16 133ns ± 6% LoadStoreCollision/*map_test.RWMutexMap-32 126ns ± 7% LoadStoreCollision/*map_test.RWMutexMap-64 131ns ±15% LoadStoreCollision/*map_test.RWMutexMap-128 152ns ± 5% LoadStoreCollision/*map_test.RWMutexMap-256 152ns ± 6% LoadStoreCollision/*map_test.RWMutexMap-512 164ns ± 3% LoadStoreCollision/*sync.Map-2 33.7ns ± 3% LoadStoreCollision/*sync.Map-4 14.9ns ± 4% LoadStoreCollision/*sync.Map-8 9.67ns ± 7% LoadStoreCollision/*sync.Map-16 7.24ns ± 6% LoadStoreCollision/*sync.Map-32 6.44ns ± 4% LoadStoreCollision/*sync.Map-64 6.45ns ± 2% LoadStoreCollision/*sync.Map-128 6.67ns ± 8% LoadStoreCollision/*sync.Map-256 6.69ns ± 3% LoadStoreCollision/*sync.Map-512 6.19ns ± 1% ``` 我們將這個結果進行可視化，如圖 1 所示，展現了`sync.Map`與其他兩種實現下，隨并發數量 n 變化而產生的性能對比： ![](https://golang.design/under-the-hood/assets/map-syncmap.png)**圖1：`MutexMap`、`RWMutexMap`與`sync.Map`之間寫少讀多場景下的性能對比** 從圖 1 中我們可以非常直觀的看出，隨著并發數量的增加，`sync.Map`在多讀場景下始終保持非常優秀的性能，而`Mutex`/`RWMutex`則隨著并發數量的增加導致性能穩步的下降，耗時越來越多。因此，`sync.Map`針對讀多寫少的場景下確實存在非常強的優化。那么這究竟是如何做到的呢？接下來我們就來就據此逐步展開討論`sync.Map`的優化設計。 ## 5.7.2 結構既然是并發安全，因此 sync.Map 一定會包含 Mutex。那么宣稱的多次讀場景下的優化一定是使用了某種特殊的機制來保證安全的情況下可以不再使用 Mutex。 ``` // Map 是一種并發安全的 map[interface{}]interface{}，在多個 goroutine 中沒有額外的鎖條件 // 讀取、存儲和刪除操作的時間復雜度平均為常量 // // Map 類型非常特殊，大部分代碼應該使用原始的 Go map。它具有單獨的鎖或協調以獲得類型安全且更易維護。 // // Map 類型針對兩種常見的用例進行優化： // 1. 給定 key 只會產生寫一次但是卻會多次讀，類似乎只增的緩存 // 2. 多個 goroutine 讀、寫以及覆蓋不同的 key // 這兩種情況下，與單獨使用 Mutex 或 RWMutex 的 map 相比，會顯著降低競爭情況 // // 零值 Map 為空且可以直接使用，Map 使用后不能復制 type Map struct { mu Mutex // read 包含 map 內容的一部分，這些內容對于并發訪問是安全的（有或不使用 mu）。 // // read 字段 load 總是安全的，但是必須使用 mu 進行 store。 // // 存儲在 read 中的 entry 可以在沒有 mu 的情況下并發更新， // 但是更新已經刪除的 entry 需要將 entry 復制到 dirty map 中，并使用 mu 進行刪除。 read atomic.Value // 只讀 // dirty 含了需要 mu 的 map 內容的一部分。為了確保將 dirty map 快速地轉為 read map， // 它還包括了 read map 中所有未刪除的 entry。 // // 刪除的 entry 不會存儲在 dirty map 中。在 clean map 中，被刪除的 entry 必須被刪除并添加到 dirty 中， // 然后才能將新的值存儲為它 // // 如果 dirty map 為 nil，則下一次的寫行為會通過 clean map 的淺拷貝進行初始化 dirty map[interface{}]*entry // misses 計算了從 read map 上一次更新開始的 load 數，需要 lock 以確定 key 是否存在。 // // 一旦發生足夠的 misses 足以囊括復制 dirty map 的成本，dirty map 將被提升為 read map（處于未修改狀態） // 并且 map 的下一次 store 將生成新的 dirty 副本。 misses int } ``` 在這個結構中，可以看到`read`和`dirty`分別對應兩個`map`，但`read`的結構比較特殊，是一個`atomic.Value`類型。從`misses`的描述中可以大致看出 sync.Map 的思路是發生足夠多的讀時，就將 dirty map 復制一份到 read map 上。從而實現在 read map 上的讀操作不再需要昂貴的 Mutex 操作。 ## 5.7.3 寫操作 Store 我們先來看看`Store()`。 ``` // Store 存儲 key 對應的 value func (m *Map) Store(key, value interface{}) { // 獲得 read map read, _ := m.read.Load().(readOnly) // 修改一個已經存在的值 // 讀取 read map 中的值 // 如果讀到了，則嘗試更新 read map 的值，如果更新成功，則直接返回，否則還要繼續處理（當且僅當要更新的值被標記為刪除） // 如果沒讀到，則還要繼續處理（read map 中不存在） if e, ok := read.m[key]; ok && e.tryStore(&value) { return } (...) } ``` 可以看到，首先發生的是更新已經存在值的情況：更新操作直接更新 read map 中的值，如果成功則不需要進行任何操作，如果沒有成功才繼續處理。我們來看一下`tryStore`。 ``` // tryStore 在 entry 還沒有被刪除的情況下存儲其值 // // 如果 entry 被刪除了，則 tryStore 返回 false 且不修改 entry func (e *entry) tryStore(i *interface{}) bool { for { // 讀取 entry p := atomic.LoadPointer(&e.p) // 如果 entry 已經刪除，則無法存儲，返回 if p == expunged { return false } // 交換 p 和 i 的值，如果成功則立即返回 if atomic.CompareAndSwapPointer(&e.p, p, unsafe.Pointer(i)) { return true } } } ``` 從`tryStore`可以看出，在更新操作中只要沒有發生 key 的刪除情況，即值已經在 dirty map 中標記為刪除，更新操作一定只更新到 read map 中，不涉及與 dirty map 之間的數據同步。我們繼續看`Store()`剩下的部分。下面的情況就相對復雜一些了，在鎖住結構后，要做的第一件事情就是更新剛才讀過的 read map。剛才我們僅僅只是修改一個已經存在的值，現在我們面臨三種情況： **情況1** ``` // Store 存儲 key 對應的 value func (m *Map) Store(key, value interface{}) { (...) m.mu.Lock() // 經過剛才的一系列操作，read map 可能已經更新了 // 因此需要再讀一次 read, _ = m.read.Load().(readOnly) if e, ok := read.m[key]; ok { // 修改一個已經存在的值 if e.unexpungeLocked() { // 說明 entry 先前是被標記為刪除了的，現在我們又要存儲它，只能向 dirty map 進行更新了 m.dirty[key] = e } // 無論先前刪除與否，都要更新 read map e.storeLocked(&value) } else if e, ok := m.dirty[key]; ok { (...) } else { (...) } m.mu.Unlock() } ``` 這種情況下，本質上還分兩種情況： 1. 可能因為是一個已經刪除的值（之前的`tryStore`失敗） 2. 可能先前僅保存在 dirty map 然后同步到了 read map（這是可能的，我們后面讀 Load 時再來分析 dirty map 是如何同步到 read map 的）對于第一種而言，我們需要重新將這個已經刪除的值標記為沒有刪除，然后將這個值同步回 dirty map（刪除操作只刪除 dirty map，之后再說）對于第二種狀態，我們直接更新 read map，不需要打擾 dirty map。 **情況2** ``` // Store 存儲 key 對應的 value func (m *Map) Store(key, value interface{}) { (...) if e, ok := read.m[key]; ok { (...) } else if e, ok := m.dirty[key]; ok { e.storeLocked(&value) // 更新 dirty map 的值即可 } else { (...) } m.mu.Unlock() } ``` 我們發現 read map 中沒有想要更新的值，那么看一下 dirty map 有沒有，結果發現是有的，那么我們直接修改 dirty map，不去打擾 read map。 **情況3** ``` // Store 存儲 key 對應的 value func (m *Map) Store(key, value interface{}) { (...) if e, ok := read.m[key]; ok { (...) } else if e, ok := m.dirty[key]; ok { (...) } else { // 如果 dirty map 里沒有 read map 沒有的值（兩者相同） if !read.amended { // 首次添加一個新的值到 dirty map 中 // 確保已被分配并標記為 read map 是不完備的(dirty map 有 read map 沒有的) m.dirtyLocked() // 更新 amended，標記 read map 中缺少了值（標記為兩者不同） m.read.Store(readOnly{m: read.m, amended: true}) } // 不管 read map 和 dirty map 相同與否，正式保存新的值 m.dirty[key] = newEntry(value) } m.mu.Unlock() } // 只是簡單的創建一個 entry // entry 是一個對應于 map 中特殊 key 的 slot type entry struct { // p 指向 interface{} 類型的值，用于保存 entry // // 如果 p == nil，則 entry 已被刪除，且 m.dirty == nil // // 如果 p == expunged, 則 entry 已經被刪除，m.dirty != nil ，則 entry 不在 m.dirty 中 // // 否則，entry 仍然有效，且被記錄在 m.read.m[key] ，但如果 m.dirty != nil，則在 m.dirty[key] 中 // // 一個 entry 可以被原子替換為 nil 來刪除：當 m.dirty 下一次創建時，它會自動將 nil 替換為 expunged 且 // 讓 m.dirty[key] 成為未設置的狀態。 // // 與一個 entry 關聯的值可以被原子替換式的更新，提供的 p != expunged。如果 p == expunged， // 則與 entry 關聯的值只能在 m.dirty[key] = e 設置后被更新，因此會使用 dirty map 來查找 entry。 p unsafe.Pointer // *interface{} } func newEntry(i interface{}) *entry { return &entry{p: unsafe.Pointer(&i)} } ``` read map 和 dirty map 都沒有，只能是存儲一個新值了。當然，在更新之前我們還要再檢查一下 read map 和 dirty map 的狀態。如果 read map 和 dirty map 中存儲的內容是相同的，那么我們這次存儲新的數據只會存儲在 dirty map 中，因此會造成 read map 和 dirty map 的不一致。 read map 和 dirty map 相同的情況，首先調用`dirtyLocked()`。 ``` func (m *Map) dirtyLocked() { // 如果 dirty map 為空，則一切都很好，返回 if m.dirty != nil { return } // 獲得 read map read, _ := m.read.Load().(readOnly) // 創建一個與 read map 大小一樣的 dirty map m.dirty = make(map[interface{}]*entry, len(read.m)) // 依次將 read map 的值復制到 dirty map 中。 for k, e := range read.m { if !e.tryExpungeLocked() { m.dirty[k] = e } } } func (e *entry) tryExpungeLocked() (isExpunged bool) { // 獲取 entry 的值 p := atomic.LoadPointer(&e.p) // 如果 entry 值是 nil for p == nil { // 檢查是否被標記為已經刪除 if atomic.CompareAndSwapPointer(&e.p, nil, expunged) { // 成功交換，說明被標記為刪除 return true } // 刪除操作失敗，說明 expunged 是 nil，則重新讀取一下 p = atomic.LoadPointer(&e.p) } // 直到讀到的 p不為 nil 時，則判斷是否是標記為刪除的對象 return p == expunged } ``` 這個步驟中將 read map 中沒有被標記為刪除的值全部同步到了 dirty map 中。然后將 dirty map 標記為與 read map 不同，因為接下來我們馬上要把向 dirty map 存值了。好了，至此我們完成了整個存儲過程。小結一下： 1. 存儲過程遵循互不影響的原則，如果在 read map 中讀到，則只更新 read map，如果在 dirty map 中讀到，則只更新 dirty map。 2. 優先從 read map 中讀，更新失敗才讀 dirty map。 3. 存儲新值的時候，如果 dirty map 中沒有 read map 中的值，那么直接將整個 read map 同步到 dirty map。這時原來的 dirty map 被徹底覆蓋（一些值依賴 GC 進行清理）。 ## 5.7.4 讀操作 Load Load 的操作就是從 dirty map 或者 read map 中查找所存儲的值。 ``` // Load 返回了存儲在 map 中對應于 key 的值 value，如果不存在則返回 nil // ok 表示了值能否在 map 中找到 func (m *Map) Load(key interface{}) (value interface{}, ok bool) { // 拿到只讀 read map read, _ := m.read.Load().(readOnly) // 從只讀 map 中讀 key 對應的 value e, ok := read.m[key] // 如果在 read map 中找不到，且 dirty map 包含 read map 中不存在的 key，則進一步查找 if !ok && read.amended { m.mu.Lock() // 鎖住后，再讀一次 read map read, _ = m.read.Load().(readOnly) e, ok = read.m[key] // 如果這時 read map 確實讀不到，且 dirty map 與 read map 不一致 if !ok && read.amended { // 則從 dirty map 中讀 e, ok = m.dirty[key] // 無論 entry 是否找到，記錄一次 miss：該 key 會采取 slow path 進行讀取，直到 // dirty map 被提升為 read map。 m.missLocked() } m.mu.Unlock() } // 如果 read map 或者 dirty map 中找不到 key，則確實沒找到，返回 nil 和 false if !ok { return nil, false } // 如果找到了，則返回讀到的值 return e.load() } func (e *entry) load() (value interface{}, ok bool) { // 讀 entry 的值 p := atomic.LoadPointer(&e.p) // 如果值為 nil 或者已經刪除 if p == nil || p == expunged { // 則讀不到 return nil, false } // 否則讀值 return *(*interface{})(p), true } ``` 可以看到： 1. 如果 read map 中已經找到了該值，則不需要去訪問 dirty map（慢）。 2. 但如果沒找到，且 dirty map 與 read map 沒有差異，則也不需要去訪問 dirty map。 3. 如果 dirty map 和 read map 有差異，則我們需要鎖住整個 Map，然后再讀取一次 read map 來防止并發導致的上一次讀取失誤 4. 如果鎖住后，確實 read map 讀取不到且 dirty map 和 read map 一致，則不需要去讀 dirty map 了，直接解鎖返回。 5. 如果鎖住后，read map 讀不到，且 dirty map 與 read map 不一致，則該 key 可能在 dirty map 中，我們需要從 dirty map 中讀取，并記錄一次 miss（在 read map 中 miss）。當記錄 miss 時，涉及`missLocked`操作： ``` // 此方法調用時，整個 map 是鎖住的 func (m *Map) missLocked() { // 增加一次 miss m.misses++ // 如果 miss 的次數小于 dirty map 的 key 數 // 則直接返回 if m.misses < len(m.dirty) { return } // 否則將 dirty map 同步到 read map 去 m.read.Store(readOnly{m: m.dirty}) // 清空 dirty map m.dirty = nil // miss 計數歸零 m.misses = 0 } ``` 可以看出，miss 如果大于了 dirty 所存儲的 key 數時，會將 dirty map 同步到 read map，并將自身清空，miss 計數歸零。 ## 5.7.5 刪除操作 Delete 再來看刪除操作。 ` // Delete 刪除 key 對應的 value func (m *Map) Delete(key interface{}) { // 獲得 read map read, _ := m.read.Load().(readOnly) // 從 read map 中讀取需要刪除的 key e, ok := read.m[key] // 如果 read map 中沒找到，且 read map 與 dirty map 不一致 // 說明要刪除的值在 dirty map 中 if !ok && read.amended { // 在 dirty map 中需要加鎖 m.mu.Lock() // 再次讀 read map read, _ = m.read.Load().(readOnly) // 從 read map 中取值 e, ok = read.m[key] // 沒取到，read map 和 dirty map 不一致 if !ok && read.amended { // 刪除 dierty map 的值 delete(m.dirty, key) } m.mu.Unlock() }`` // 如果 read map 中找到了 if ok { // 則執行刪除 e.delete() } } func (e *entry) delete() (hadValue bool) { for { // 讀取 entry 的值 p := atomic.LoadPointer(&e.p) // 如果 p 等于 nil，或者 p 已經標記刪除 if p == nil || p == expunged { // 則不需要刪除 return false } // 否則，將 p 的值與 nil 進行原子換 if atomic.CompareAndSwapPointer(&e.p, p, nil) { // 刪除成功（本質只是解除引用，實際上是留給 GC 清理） return true } } } ``` 從實現上來看，刪除操作相對簡單，當需要刪除一個值時，移除 read map 中的值，本質上僅僅只是解除對變量的引用。實際的回收是由 GC 進行處理。如果 read map 中并未找到要刪除的值，才會去嘗試刪除 dirty map 中的值。 ## 5.7.6 迭代操作 Range 有了上面的存取基礎，這時候來看 Range 一切都顯得很自然： ``` // Range 為每個 key 順序的調用 f。如果 f 返回 false，則 range 會停止迭代。 // // Range 的時間復雜度可能會是 O(N) 即便是 f 返回 false。 func (m *Map) Range(f func(key, value interface{}) bool) { // 讀取 read map read, _ := m.read.Load().(readOnly) // 如果 read map 和 dirty map 不一致，則需要進一步操作 if read.amended { m.mu.Lock() // 再讀一次，如果還是不一致，則將 dirty map 提升為 read map 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() } // 在 read 變量中讀（可能是 read map ，也可能是 dirty map 同步過來的 map） for k, e := range read.m { // 讀 readOnly，load 會檢查該值是否被標記為刪除 v, ok := e.load() // 如果已經刪除，則跳過 if !ok { continue } // 如果 f 返回 false，則停止迭代 if !f(k, v) { break } } } ``` 既然要 Range 整個 map，則需要考慮 dirty map 與 read map 不一致的問題，如果不一致，則直接將 dirty map 同步到 read map 中。 ## 5.7.7 讀寫操作 LoadOrStore ``` // LoadOrStore 在 key 已經存在時，返回存在的值，否則存儲當前給定的值 // loaded 為 true 表示 actual 讀取成功，否則為 false 表示 value 存儲成功 func (m *Map) LoadOrStore(key, value interface{}) (actual interface{}, loaded bool) { // 讀 read map read, _ := m.read.Load().(readOnly) // 如果 read map 中已經讀到 if e, ok := read.m[key]; ok { // 嘗試存儲（可能 key 是一個已刪除的 key） actual, loaded, ok := e.tryLoadOrStore(value) // 如果存儲成功，則直接返回 if ok { return actual, loaded } } // 否則，涉及 dirty map，加鎖 m.mu.Lock() // 再讀一次 read map read, _ = m.read.Load().(readOnly) if e, ok := read.m[key]; ok { // 如果 read map 中已經讀到，則看該值是否被刪除 if e.unexpungeLocked() { // 沒有被刪除，則通過 dirty map 存 m.dirty[key] = e } actual, loaded, _ = e.tryLoadOrStore(value) } else if e, ok := m.dirty[key]; ok { // 如果 read map 沒找到, dirty map 找到了 // 嘗試 laod or store，并記錄 miss actual, loaded, _ = e.tryLoadOrStore(value) m.missLocked() } else { // 否則就是存一個新的值 // 如果 read map 和 dirty map 相同，則開始標記不同 if !read.amended { m.dirtyLocked() m.read.Store(readOnly{m: read.m, amended: true}) } // 存到 dirty map 中去 m.dirty[key] = newEntry(value) actual, loaded = value, false } m.mu.Unlock() // 返回存取狀態 return actual, loaded } ``` 我們已經看過 Load 或 Store 的單獨過程了，`LoadOrStore`方法無非是兩則的結合，比較簡單，這里就不再細說了。 ## 5.7.8 小結我們來回顧一下 sync.Map 中 read map 和 dirty map 的同步過程： 1. 當 Store 一個新值會發生：read map –> dirty map 2. dirty map –> read map：當 read map 進行 Load 失敗 len(dirty map) 次之后發生因此，無論是存儲還是讀取，read map 中的值一定能在 dirty map 中找到。無論兩者如何同步，sync.Map 通過 entry 指針操作，保證數據永遠只有一份，一旦 read map 中的值修改，dirty map 中保存的指針就能直接讀到修改后的值。當存儲新值時，一定發生在 dirty map 中。當讀取舊值時，如果 read map 讀到則直接返回，如果沒有讀到，則嘗試加鎖去 dirty map 中取。這也就是官方宣稱的 sync.Map 適用于一次寫入多次讀取的情景。