一、前言
Go語言在設計上對同步（Synchronization，資料同步和執行緒同步）提供大量的支援，比如 goroutine和channel同步原語，庫層面有
1。 sync：提供基本的同步原語（比如Mutex、RWMutex、Locker）和 工具類（Once、WaitGroup、Cond、Pool、Map）
2。 sync/atomic：提供原子操作（基於硬體指令compare-and-swap）
Golang package sync 剖析（三）：sync。Cond
上一期中，我們介紹了 Go 語言對條件變數的實現 sync。Cond。本期文章我們介紹 package sync 下的另一個工具類：sync。Map。我們先看一個故事：
面試官：看你用過 Go，大概用過多久，感覺到哪個段位？
小明：用了一年半左右，算是精通
面試官：用過 sync。Map 嗎？
小明：沒用過，大致瞭解過
面試官：Go 內建的 map 和 sync。Map 有什麼區別？
小明（假裝會）：內建的map 不支援併發訪問， sync。Map 支援併發訪問
面試官：給內建的 map加上讀寫鎖不一樣能實現嗎？兩個有啥區別？
小明：。。。
小明卒，享年28歲
提到 sync。Map，我們首先想到的是 go 內建的 map［KeyType］ValueType（簡稱 map）。內建 map 基於 hashtable 實現，具有以下幾個特點：
Get/Set 的時間複雜度都是 O（1）
支援範型，使用起來非常方便
不支援併發訪問
如果要支援併發訪問，通常透過加鎖實現，示例程式碼如下：
type Any interface{}type ConcurrentMap struct { sync。RWMutex data map［Any］Any}func NewConcurrentMap（capacity int） *ConcurrentMap { return &ConcurrentMap{ data： make（map［Any］Any， capacity）， }}func （m *ConcurrentMap） Get（key Any） （Any， bool） { m。RLock（） defer m。RUnlock（） val， ok ：= m。data［key］ return val， ok}func （m *ConcurrentMap） Set（key， val Any） { m。Lock（） defer m。Unlock（） m。data［key］ = val}
如果把加鎖的map做成通用類，由於Go不支援範型，要引入interface，可讀性變差。所以通常情況下，我們在 package 級別宣告兩個變數，直接拿來用：
// 假設需要一個 string-> string 的mapvar （ rwlock sync。RWMutex data map［string］string）
那麼問題來了：既然自己給 map 加鎖很簡單，為什麼還要造輪子，搞出來一個 sync。Map？
二、sync。Map 有啥優勢？
我們扒一扒 godoc，原文是英文，下面是我的翻譯：
Map 類似於 Go內建的 map［interface{}］interface{}，但是支援多協程的併發訪問，而不需要額外的鎖或同步機制。它的三個成員函式 Load/Store/Delete 的時間複雜度（均攤時間）都是 O（1）。
Map 型別有特定的應用場景。由於內建 map 在型別安全上表現更好，也更方便對map裡的資料進行定製化處理，加一把鎖就能滿足大多數需求了。
Map 型別針對下面兩個應用場景做了最佳化：
1）特定的 entry
2）多個 goroutine 的併發讀、寫、覆蓋三種操作的 key 沒有交集；
在上面兩個應用場景下，相對於內建map+鎖方案，Map 裡搶佔鎖的行為會少很多。
Map 型別的預設值是空，可以被立即使用。Map例項在被第一次使用以後，不能被複製。
https：//golang。org/pkg/sync/#Map
前兩段話意思是 sync。Map 功能上是一個支援併發訪問的 map，但是有特定的應用場景，大多數情況下你用不到。第三段意思是：sync。Map 針對map粒度有併發讀寫但key粒度極少併發讀寫的場景做了專門最佳化。
具體點說，在上面的 ConcurrentMap 類裡，每次讀寫都會鎖住整個map物件，而 sync。Map 的讀寫在 cache hit 時 lock-free，cache miss 時才需要鎖住整個map物件。下面是一個例子來說明 cache hit 和 cache miss：
package mainimport （ “fmt” “sync”）func main（） { var cache sync。Map cache。Load（“key”） // cache miss cache。Store（“key”， “value”） // cache miss cache。Load（“key”） // cache hit cache。Store（“key”， “value_v1”） // cache hit // 刪除key，先軟刪除 // 觸發特定條件後會硬刪除 cache。Delete（“key”） cache。Load（“key”） // cache hit，但讀不到值 cache。Store（“key”， “value_v2”） // cache hit，但仍需上鎖}
需要注意的是：雖然 godoc 給的最佳化場景是寫一次，讀多次，但並不是絕對的。在對一個 key 寫多次，讀多次時，sync。Map 也能保證結果的正確性。
準確來說，一個key的寫入是 lock-free的，對它的高頻read/update也是lock-free的；一旦觸發 delete，就沒法利用lock-free的效能優勢了。
順便提一句 lock-free。多個執行緒訪問同一塊記憶體時，資料同步是必須的，處理的方式無非是管道、鎖、原子操作。提到lock-free，通常想到 Compare-And-Swap，CAS依賴於原子操作 。
鎖在程式語言層面上實現，不過依賴作業系統的訊號量，可能導致執行緒被休眠和喚醒，而且它包含加鎖和解鎖兩個動作；原子操作是基於硬體指令，不會導致執行緒休眠，一步到位。
當我們說 lock-free 時，是指實現一個數據結構時，用原子操作代替鎖保證同步機制。
回顧下剛才的問題：既然自己給 map 加鎖很方便，為什麼還要造輪子，搞出來一個 sync。Map？
大家心裡應該有一些概念了，但重要的事情值得變著法子重複，sync。Map 的應用場景是：
低頻 create
高頻 read/update
很少/不 delete
注：create/read/update/delete 是從邏輯上對資料操作的定義。
三、sync。Map 怎麼用
這個環節介紹下 sync。Map 的使用方法，首先看它的成員函式：
// Load 等價於 m［key］func （m *Map） Load（key interface{}） （value interface{}， ok bool）// Store 等價於 m［key］=valuefunc （m *Map） Store（key， value interface{}）// LoadOrStore 讀不到key時，會寫入
Map 的用法非常符合直覺，唯二需要說的是 Load 和 Range。
由於 Load 返回的是 interface{}，我們需要進行顯式型別轉換。如果 key 不存在，Load 會返回 nil，型別轉換會 panic。比如，下面這段程式碼中 val2。（string） 會導致panic：
package mainimport （ “fmt” “sync”）func main（） { var m sync。Map m。Store（“key”， “value”） val1， ok1 ：= m。Load（“key”） fmt。Printf（“val1=%v， ok1=%v\n”， val1， ok1） fmt。Printf（“type safe val1 = %v\n”， val1。（string）） val2， ok2 ：= m。Load（“key_not_exist”） fmt。Printf（“val2=%v， ok2=%v\n”， val2， ok2） fmt。Printf（“type safe val2 = %v\n”， val2。（string））}
關於 Range，考慮這個場景：在 Range 執行過程中，如果另一個執行緒執行了 Store/Delete，執行 Range 的執行緒能立即感知到變化嗎？
要回答這個問題，我們複用下上面的 create/read/update/delete 概念，而不是粗粒度的 Load/Store/Delete 等成員函式，結論是：
create：Range 不能感知到新增的 key
read： read 操作不會修改 Map，不討論
update：Range 能感知到已有 key 對應的 value 的變化
delete：Range 能感知到對應 key 的 value 的標記刪除，遍歷時會跳過
由於Range 的執行過程是 1） 載入索引表 2） 遍歷索引表，所以 key 集合在第一步就確定下來了，開始遍歷後無法感知到新增的key；對於已有的key，遍歷過程中透過原子操作取出對應的value，所以能夠感知<遍歷開始，當前時刻>這段時間內發生的update和delete。
四、sync。Map 怎麼實現
前方高能
小明買了張復活卡，網上搜到了一篇標題叫 “Golang package sync 剖析（四）：sync。Map” 的文章，回顧了下面試官的問題，把前三節看完了。看到 lock-free 資料結構，十分好奇，但是急於通關，第四小節沒看就去面試了。
面試官：看你用過 Go，大概用過多久，感覺到哪個段位？
小明：用了一年半左右，比較熟悉，算不上精通
面試官：用過 sync。Map 嗎？
小明：用過，它是基於 lock-free hashtable 實現的，比<內建map， 讀寫鎖>快。對讀操作做了很多最佳化，&%。。#$#@。。%#%%$@*
面試官：嗯，還不錯。實現一個簡單的 sync。Map，把 Load/Store 兩個函式的核心邏輯寫出來就行，可以不支援 Delete
小明：。。。
小明卒，享年28歲
sync。Map 和 sync。WaitGroup 類似，是典型的用起來簡單，實現起來超級複雜的資料結構。事實上，sync。Map 的實現要更復雜一些。
我們先看 Map 的內部結構：
type Map struct { // read欄位內部是一個 readOnly 物件 // Load和Store都首先從這裡載入資料 // 如果沒有，才去訪問 Map。dirty read atomic。Value // mu 用於控制 dirty 變數的同步 // 呼叫Store時，對於新key或已刪除的key，都要寫到dirty // 呼叫Load時，如果 miss 達到 len（dirty），把dirty淺複製到 Map。read。m mu Mutex dirty map［interface{}］*entry // misses 用來記錄Load裡讀 Map。read欄位時的cache miss數 misses int}// readOnly 是一個只讀結構// m 是一個只讀索引表，從 dirty 淺複製過來// m 和 Map。dirty 裡 的 *entry 指向同樣的記憶體位置type readOnly struct { m map［interface{}］*entry amended bool // 是否需要同步}// entry 本身不存資料// 而是把資料放到指標 p 裡type entry struct { p unsafe。Pointer // *interface{}}
Map 之所以被稱為 lock-free，是因為它的兩級快取。
一級快取
一級快取是 read 欄位，每次進行 Load/Store/Delete 時，在命中一級快取時，執行的操作是：
載入索引表 readOnly。m，這一步是原子操作
透過 key 讀取對應的 *entry
讀取/更新 entry。p，也是原子操作
注：呼叫 Delete 時，如果能命中一級快取，entry。p 被置為 nil。
二級快取
二級快取是 dirty 欄位，依賴鎖保證同步。只有在一級快取cache miss 時，key才會穿透到二級快取。
對於 Load，cache miss 且兩級快取不同步時執行的操作是：
加鎖
取數
記錄cache miss 如果 misses == len（dirty） 則重新整理一級快取，清空二級快取
返回結果，解鎖
對於 Store，cache miss時執行的操作是：
加鎖
如果二級快取cache hit，則寫資料
如果二級快取 cache miss，
如果兩級快取“已同步”（amended==false），則需要將一級快取刷到二級快取上，並將一級快取置為“不同步”（amended=true），然後把新 entry 寫入二級快取
如果兩級快取不同步（amended==true），就把新 entry 寫入二級快取
解鎖
對於 Delete，cache miss時執行的操作是：
加鎖
double check 一級快取 3。1。 如果兩級快取已經同步（amended==false），則直接將key 從 dirty 刪除 3。2。 如果兩級快取不同步（amended==false），則執行軟刪除（將entry。p 置為nil）
解鎖
一級快取的讀寫是原子操作，二級快取的讀寫透過鎖保證同步。在使用過程中，應當儘量避免一級快取被穿透。說完了 sync。Map 的實現機制，我們看看原始碼。
成員函式 Load
Load 的操作相對簡單：
從一級快取 m。read 取數
一級快取cache miss，從二級快取 m。dirty 取數
如果 cache miss 積累到閾值，觸發一級快取的更新
func （m *Map） Load（key interface{}） （value interface{}， ok bool） { // 從一級快取 m。read 取數 read， _ ：= m。read。Load（）。（readOnly） e， ok ：= read。m［key］ if ！ok && read。amended { // 一級快取被擊穿，且兩級快取不同步 m。mu。Lock（） // 加鎖後 double check 一級快取 // 保證<上次讀取read， 加鎖>期間新增的key能被感知到 read， _ = m。read。Load（）。（readOnly） e， ok = read。m［key］ if ！ok && read。amended { // 一級快取未命中&&兩級快取不同步時 // 從二級快取m。dirty取數 e， ok = m。dirty［key］ // missLocked 記錄 cache miss // 如果 misses == len（m。dirty） // 則重新整理一級快取 m。read， 並清空二級快取 m。missLocked（） } m。mu。Unlock（） } // 兩級快取都沒有這個數 if ！ok { return nil， false } return e。load（）}
從程式碼我們可以看到，如果兩級快取不同步且都被擊穿，Load 的效能會比較差。
成員函式 Store
func （m *Map） Store（key， value interface{}） { // 檢查一級快取 m。read ，並更新 read， _ ：= m。read。Load（）。（readOnly） if e， ok ：= read。m［key］； ok && e。tryStore（&value） { return } // 一級快取未命中，或命中已刪除的key時 // fallback 到二級快取 m。dirty m。mu。Lock（） // 加鎖後 double check 一級快取 // 保證<上次讀取read， 加鎖>期間新增的key能被感知到 read， _ = m。read。Load（）。（readOnly） if e， ok ：= read。m［key］； ok { if e。unexpungeLocked（） { // 如果 key 被軟刪除，更新到二級快取 m。dirty m。dirty［key］ = e } e。storeLocked（&value） } else if e， ok ：= m。dirty［key］； ok { // 一級快取 m。read 被穿透 // 二級快取 m。dirty 被命中 e。storeLocked（&value） } else { // 一級快取 m。read 被穿透 // 二級快取 m。dirty 也被穿透 if ！read。amended { // 兩級快取已同步，說明二級快取 m。dirty 是 nil // 將一級快取 m。read 的key 全部刷到二級快取 m。dirty 裡 // 並將快取同步狀態置為“不同步”（amended=true） m。dirtyLocked（） m。read。Store（readOnly{m： read。m， amended： true}） } // 將新的 entry 寫入二級快取 m。dirty［key］ = newEntry（value） } m。mu。Unlock（）}
Store 一個 entry
成員函式 Delete
func （m *Map） Delete（key interface{}） { // 檢查一級快取 m。read ，並更新 read， _ ：= m。read。Load（）。（readOnly） e， ok ：= read。m［key］ if ！ok && read。amended { // 一級快取未命中，且兩級快取不同步 // fallback 到二級快取 m。dirty m。mu。Lock（） // 加鎖後 double check 一級快取 // 保證<上次讀取read， 加鎖>期間新增的key能被感知到 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（） }}
Delete 一個 key 時，如果命中一級快取，則更新一級快取，將 entry。p 置為 nil，是原子操作；如果命中二級快取，那麼直接從索引 m。dirty 刪除。
我們稍微細品一品，如果只命中二級快取，邏輯非常簡單。但是如果命中一級快取，會有兩個問題：
Load 和 Range 時，如何知道一個key被刪除了？
如果被刪除的 key 個數佔比非常高，會非常浪費記憶體，如何進行清理？
先說問題1，key 命中一級快取，進行 Delete 時，我們可以從一級快取拿到該 key 對應的 *entry，方法是 m。read。Load（）。（readOnly）。m［key］，我們拿到 *entry 以後，透過原子操作更新 entry。p，程式碼如下：
// 為了便於理解// readOnly 和 entry 也放在這兒type readOnly struct { m map［interface{}］*entry amended bool // 是否需要同步}type entry struct { p unsafe。Pointer // *interface{}}// expunged 是一個佔位符指標，用來標記entry已刪除var expunged = unsafe。Pointer（new（interface{}））// 執行軟刪除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 } }}
執行軟刪除以後，由於二級快取 m。dirty［key］ 指向的記憶體地址和 m。read。Load（）。（readOnly）。m［key］ 一樣，所以也被“偷偷”更新了。
說說問題2，一個 key 被刪除以後，資料會刪除，但是索引表裡還記錄著佔位符，佔位符的比例太高的話也會影響整體效能。上面講 Store 函式時，我們提到：
如果兩級快取都cache miss，且兩級快取“已同步”，則需要將一級快取刷到二級快取上，並將一級快取置為“不同步”（amended=true），然後把新 entry 寫入二級快取。
為什麼需要將一級快取刷到二級快取呢？我們先看看刷的時候做了什麼：
func （m *Map） dirtyLocked（） { // 兩級快取已同步，所以 m。dirty 肯定是nil 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 { // 讀取entry的刪除狀態 // 未刪除的entry 才複製到 m。dirty if ！e。tryExpungeLocked（） { m。dirty［k］ = e } }}
檢查刪除狀態的函式 tryExpungeLocked 寫得非常風騷，這篇文章的細節已經很多了，這裡不再列出來，有興趣的同學可以看原始碼。
五、小結一下
sync。Map 的底層是典型的 lock-free hash table，一級快取透過原子操作提供效能保障，二級快取藉助鎖保障邏輯的正確性和完備性。兩級快取的同步是 sync。Map 實現裡最複雜的部分，本文儘可能從全域性上說清楚，細節上肯定會有不少疏漏，希望大家能閱讀原始碼，對 sync。Map 有比較透徹的理解。
小明又買了一張復活卡，…
六、References
sync。Map
https：//golang。org/pkg/sync/#Map
The World’s Simplest Lock-Free Hash Table
https：//preshing。com/20130605/the-worlds-simplest-lock-free-hash-table/
