上一篇文章 《大白話講講 Go 語言的 sync.Map（一）》 講到 entry 資料結構，原因是 Go 語言標準庫的 map 不是執行緒安全的，通過加一層抽象迴避這個問題。
當一個 key 被刪除的時候，比如李四銷戶了，以前要撕掉小賬本，現在可以在大賬本上寫 expunged，
對，什麼也不寫也是 OK 的。也就是說，
entry.p 可能是真正的資料的地址，也可能是 nil，也可能是 expunged。
為什麼無端端搞這個 expunged 幹嘛？因為 sync.Map 實際上是有兩個小賬本，
一個叫 readOnly map（唯讀賬本），一個叫 dirty map（可讀、也可寫賬本）：
type Map struct {
mu sync.Mutex
read atomic.Value // readOnly
dirty map[interface{}]*entry
misses int
}
type readOnly struct {
m map[interface{}]*entry
amended bool // true if the dirty map contains some key not in m.
}
既然有賬本一個變成兩個，那肯定會有些時候出現兩個 map 資料是不一致的情況。
readOnly 結構的 amended 欄位，是一個標記，為 true 的時候代表 dirty map 包含了一些 key，這些 key 不會存在 readOnly map 中。
這個欄位的作用，在於加速查詢的過程。
假設 readOnly 賬本上有 張三、李四、錢五，dirty 賬本除了這三個人，後面又新增了 王六，查詢邏輯就是這樣的：
如果 2 的 amended 標記是兩個賬本資料一致，那就沒有執行 3 的必要了。
我們可以看看原始碼是怎麼實現的：
func (m *Map) Load(key interface{}) (value interface{}, ok bool) {
read, _ := m.read.Load().(readOnly)
// 1. 先在 readOnly 查詢，王六不在
e, ok := read.m[key]
// 2. 判斷 amended ，發現兩個賬本資料是不一致的
if !ok && read.amended {
// 加鎖的原因是，前面步驟 1 的讀取有可能被另一個協程的 missLocked 更改了
// 導致讀出來的值不符合預期，所以加鎖再讀取一次，老套路了。
m.mu.Lock()
read, _ = m.read.Load().(readOnly)
e, ok = read.m[key]
if !ok && read.amended {
// 3. 再去 dirty 賬本查詢，終於找到王六
e, ok = m.dirty[key]
// missLocked 擁有一個計數器，
// 它的作用在於 readOnly 如果一直查不到，經常退化到 dirty，
// 那就把 dirty 作為 readOnly ，直接取代它。
m.missLocked()
}
m.mu.Unlock()
}
if !ok {
return nil, false
}
return e.load() // 還記得大賬本嗎？這裡是拿到最終的值，針對 entry.p == expunged 做了特殊處理。
}
func (m *Map) missLocked() {
// 查不到就遞增 misses 計數器
m.misses++
// 這個判斷條件不是常數，而是 dirty map 的記錄數。
// 這個判斷條件很奇妙，
// 它使得 dirty 取代 readOnly 的時機，和 dirty 的資料量正相關了。
// 也就是說，dirty map 越大，對兩個 map 不一致的容忍度越大，
// 不會有頻繁的取代操作。
if m.misses < len(m.dirty) {
// 如果不是經常查不到，說明 readOnly 還是可以用的，退出。
return
}
// 如果 readOnly 已經沒有存在價值，那就把 dirty 取代 readOnly。
// 此時，dirty 置空，並把 misses 計數器置 0。
// read 和 dirty 的資料型別都是 map[interface{}]*entry，
// 可以直接替換，無需型別轉換，這個設計簡直完美。
m.read.Store(readOnly{m: m.dirty})
m.dirty = nil
m.misses = 0
}
func (e *entry) load() (value interface{}, ok bool) {
p := atomic.LoadPointer(&e.p)
// entry.p 可能是真正的資料的地址，也可能是 nil，也可能是 expunged
if p == nil || p == expunged {
// nil 或者是 expunged 都是不存在的，返回空
return nil, false
}
// 如果是真正的資料地址，那就返回真正的資料（就是拿到大賬本的某一頁紙上的內容）
return *(*interface{})(p), true
}
到這裡已經講完資料讀取這部分的程式碼了，接著再講資料是怎麼寫入的。
上一篇文章我留了一個思考題，
為什麼小賬本不能做到同時修改？限於篇幅，我不會展開。
我現在解答我們有了大賬本，是如何做到同時修改的！
答案在這裡：
// tryStore 顧名思義，就是不斷嘗試的意思。
// 你可以看到有一個無條件的死迴圈，只有某些條件滿足的時候才會退出
// 計算機術語：自旋（自己一直在旋轉）
func (e *entry) tryStore(i *interface{}) bool {
for {
p := atomic.LoadPointer(&e.p)
// readOnly map 儲存的是 entry 結構，p 就是所謂的大賬本，
// p 指向大賬本上某一頁紙上的內容，
// 當賬本查不到的時候，返回查不到。
if p == expunged {
return false
}
// 當賬本可以查到的時候，使用 CAS 把舊的值，替換為新的值。
// 可以查到並替換成功，返回成功，函數退出
// 查不到或者替換失敗，自旋，重試，直到成功為止
if atomic.CompareAndSwapPointer(&e.p, p, unsafe.Pointer(i)) {
return true
}
}
}
問題來了， CAS（Compare and Swap，比較並交換）是什麼東西？我們看這個加法函數：
func add(delta int32) {
for {
// 把原先的值取出來
oldValue := atomic.LoadInt32(&addr)
// 讀取後，如果沒有其他人對它修改（Compare）
// 那就用 oldValue+delta 新值，替換掉原來的值（Swap）
// 成功程式退出，失敗了就自旋重試（可能被其他人改了導致 Compare 不成功）
if atomic.CompareAndSwapInt32(&addr, oldValue, oldValue+delta) {
return
}
}
}
越來越有趣了，atomic.CompareAndSwapInt32 到底是個啥子喲？
它的具體實現在 src/runtime/internal/atomic/asm_amd64.s 裡（不同 CPU 架構，使用的檔案不同，這裡以最常見的 amd64 為例）：
// bool Cas(int32 *val, int32 old, int32 new)
// Atomically:
// if (*val == old) {
// *val = new;
// return 1;
// } else
// return 0;
TEXT runtime∕internal∕atomic·Cas(SB),NOSPLIT,$0-17
MOVQ ptr+0(FP), BX
MOVL old+8(FP), AX
MOVL new+12(FP), CX
LOCK
CMPXCHGL CX, 0(BX)
SETEQ ret+16(FP)
RET
FP（Frame pointer: arguments and locals）：
函數的輸入引數，格式 symbol+offset(FP)，symbol 沒有實際意義，只為了增強程式碼可讀性，但沒有 symbol 程式無法編譯。
ptr+0(FP) 代表第一個引數，取出複製給 BX 暫存器。
由於 ptr 是一個指標，在 64 位的處理器中，一個指標的佔 8 個位元組，
所以第二個引數 old+8(FP)，偏移量 offset 等於 8，
而第三個引數 new+12(FP)，偏移量再加 4 的原因是 int32 佔據 4 個位元組。
LOCK 指令字首會設定處理器的 LOCK# 訊號，鎖定匯流排，阻止其他處理器接管匯流排存取記憶體，
設定 LOCK# 訊號能保證某個處理器對共用記憶體的獨佔使用。
CMPXCHGL CX, 0(BX) 是比較並交換的指令，將 AX 和 CX 比較，相同將 BX 指向的內容 放入 CX，
CMPXCHGL 暗中使用了 AX 暫存器。
兜了一大圈，終於明白大賬本的資料是怎樣被更新的了。
看看資料是怎麼寫入之前，我們要知道資料是怎麼被刪除的：
// 刪除的邏輯是比較簡單的。
func (m *Map) LoadAndDelete(key interface{}) (value interface{}, loaded bool) {
read, _ := m.read.Load().(readOnly)
e, ok := read.m[key]
// key 不存在的時候並且 readOnly map 和 dirty map 不一致時，
// 把 dirty map 對應的記錄刪了。
if !ok && read.amended {
m.mu.Lock()
read, _ = m.read.Load().(readOnly)
e, ok = read.m[key]
if !ok && read.amended {
// 資料不一致的時候，最終讀出來的值以 dirty map 為主，
// 即使 readOnly map 是 !ok 的，但 dirty map 可能是 ok 的，
// 既然值可能是存在的，那就讀取出來。
e, ok = m.dirty[key]
// 刪除操作
delete(m.dirty, key)
// 遞增資料不一致的計數器。
// 太多不一致會把 dirty map 提升為 readOnly map，前面講過了。
m.missLocked()
}
m.mu.Unlock()
}
// key 存在的時候，把 key 置為 nil，注意這裡不是 expunged，
// 這也是我為什麼要先講 Delete 的原因。
if ok {
return e.delete()
}
return nil, false
}
// Delete deletes the value for a key.
func (m *Map) Delete(key interface{}) {
m.LoadAndDelete(key)
}
// delete 將對應的 key 置為 nil！而不是 expunged！
func (e *entry) delete() (value interface{}, ok bool) {
for {
p := atomic.LoadPointer(&e.p)
if p == nil || p == expunged {
return nil, false
}
if atomic.CompareAndSwapPointer(&e.p, p, nil) {
return *(*interface{})(p), true
}
}
}
OK，我們看資料寫入的邏輯，它是整個原始碼中最難理解的，隱含的邏輯關係非常多：
// unexpungeLocked 將 expunged 的標記變成 nil。
func (e *entry) unexpungeLocked() (wasExpunged bool) {
return atomic.CompareAndSwapPointer(&e.p, expunged, nil)
}
// storeLocked 將 entry.p 指向具體的值
func (e *entry) storeLocked(i *interface{}) {
atomic.StorePointer(&e.p, unsafe.Pointer(i))
}
// tryExpungeLocked 嘗試 entry.p == nil 的 entry 標記為刪除(expunged)
func (e *entry) tryExpungeLocked() (isExpunged bool) {
p := atomic.LoadPointer(&e.p)
// for 迴圈的作用，可以保證 p != nil，
// 保證寫時複製過程中，p == nil 的情況不會被寫到 dirty map 中。
for p == nil {
if atomic.CompareAndSwapPointer(&e.p, nil, expunged) {
return true
}
p = atomic.LoadPointer(&e.p)
}
return p == expunged
}
// dirtyLocked 寫時複製，兩個 map 都找不到新增的 key 的時候呼叫的。
func (m *Map) dirtyLocked() {
// dirty 被置為 nil 的情景還記得嗎？
//
// 當 readOnly map 一直讀不到，需要退化到 dirty map 讀取的時候，
// dirty map 會被提升為 readOnly map，
// 此時，dirty map 就會被置空。
//
// 但是，dirtyLocked 被呼叫之前，
// 都是判斷 read.amended 是否為 false
// if !read.amended {...}
// 個人認為，可以直接判斷 if m.dirty == nil {...}，
// 程式碼可讀性更強！下面三行程式碼也可以不要了。
if m.dirty != nil {
return
}
// 遍歷 readOnly map，把裡面的內容都複製到新建立的 dirty map 中。
read, _ := m.read.Load().(readOnly)
m.dirty = make(map[interface{}]*entry, len(read.m))
for k, e := range read.m {
// tryExpungeLocked 將 entry.p == nil 設定為 expunged，
// 遍歷之後，所有的 nil 都變成 expunged 了。
// 返回 false 說明 p 是有值的，要拷貝到 dirty 裡。
// Delete 操作會把有值的狀態，轉移為 nil，
// 並不會把 expunged 狀態轉移為 nil，
// 由於 for 迴圈的存在，p 也不會等於 nil，
// 也就是說，tryExpungeLocked 的 p == expunged 是可以信任的。
if !e.tryExpungeLocked() {
// 如果沒有被刪除，拷貝到 dirty map 中。
m.dirty[k] = e
}
}
}
func (m *Map) Store(key, value interface{}) {
// 如果 readOnly map 有對應的 key，
// 通過 e.tryStore 直接寫入（就是上面更新大賬本的整個過程），
// 注意，tryStore 會在 entry.p == expunged 的情況下失敗。
read, _ := m.read.Load().(readOnly)
if e, ok := read.m[key]; ok && e.tryStore(&value) {
return
}
// readOnly map 找不到，或者 key 被刪除了，
// 那就寫到 dirty map 裡面。
m.mu.Lock()
read, _ = m.read.Load().(readOnly)
if e, ok := read.m[key]; ok {
// unexpungeLocked 將 expunged 的標記變成 nil。
// 當 entry.p == expunged，並且成功替換為 nil，
// 返回 true。
//
// 這個分支的意義在於，寫時複製 dirtyLocked 的時候，
// 資料從 readOnly map 搬遷到 dirty map 中，
// 如果 p 是被刪除的，dirty 是不會有這個 key 的，
// 所以要把它也寫進 dirty 中，保證資料的一致性。
//
// 為什麼好端端的 expunged，要改成 nil？
// unexpungeLocked 是一個原子操作，成功的話，
// 說明 p == expunged，
// 說明寫時複製已經完成。
//
// 為什麼要寫時複製完成之後，才可以去改 dirty？
// 我理解是這樣的：
// 如果不這樣做，dirty 會被你修改成 Store 傳進來的引數，
// 寫時複製又把它修改成 readOnly map 的值，
// 所以更新 readOnly map 就好了。
//
// 這一塊的細節真的非常多，每一塊地方都要小心處理好。
if e.unexpungeLocked() {
m.dirty[key] = e
}
// 寫入值。
e.storeLocked(&value)
} else if e, ok := m.dirty[key]; ok {
// 如果 dirty map 存在就直接更新進去，這個很好理解，
// 因為 readOnly map 找不到會來 dirty 查。
e.storeLocked(&value)
} else {
// 兩個 map 都找不到的時候，說明這是一個新的 key。
//
// 1. 如果 dirty 之前被提升為 readOnly，那就導一份沒有被刪除的 key 進來。
//
// 這個判斷條件，我理解等價於 if m.dirty == nil {...}
if !read.amended {
// 初始化 m.dirty，並把值寫進去（寫時複製）
m.dirtyLocked()
// amended 設定為不一致。
// amended 表示 dirty 是否包含了 readOnly 沒有的記錄，
// 很明顯，read.m[key] 是 !ok 的，
// 下面把值存到 dirty map 裡面了。
m.read.Store(readOnly{m: read.m, amended: true})
}
// 2. 這裡，把值存到 dirty map 中。
m.dirty[key] = newEntry(value)
}
m.mu.Unlock()
}
精妙絕倫！整個寫入的邏輯就講完了，最後看看遍歷吧，非常簡單：
func (m *Map) Range(f func(key, value interface{}) bool) {
read, _ := m.read.Load().(readOnly)
// 如果不一致，就把 dirty 提升為 readOnly，
// 同時 dirty 置空，
// 因為 dirty map 也包含了 readOnly map 沒有的 key。
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()
}
// 遍歷 readOnly map 的資料，執行回撥函數。
for k, e := range read.m {
v, ok := e.load()
if !ok {
continue
}
if !f(k, v) {
break
}
}
}
好了，到這裡整個 sync.Map 就講完了，剩下的程式碼也沒多少了，套路差不多，我們總結一下：
這篇文章近 5000 字（第一篇差不多 2000 字），從構思、成文到校對，真的需要花費不少時間，希望對你有幫助！
文章來源於本人部落格，釋出於 2021-05-05，原文連結：https://imlht.com/archives/258/