由淺入深聊聊Golang的sync.Map

時間 2019-11-17

標籤由淺入深聊聊 golang sync.map sync map 欄目 Go 简体版

原文原文鏈接

前言

今天在技術羣中有小夥伴討論併發安全的東西，其實以前就有寫過map相關文章：由淺入深聊聊Golang的map。可是沒有詳細說明sync.Map是怎麼一回事。回想了一下，居然腦中只剩下「兩個map、一個只讀一個讀寫，xxxxx」等，關鍵詞。有印象能扯，可是有點亂，仍是寫一遍簡單記錄一下吧。mysql

1.爲何須要sync.Map？ 2.sync.Map如何使用？ 3.理一理sync.Map源碼實現？ 4.sync.Map的優缺點？ 5.思惟擴散？git

正文

1.爲何須要sync.Map？

關於map能夠直接查看由淺入深聊聊Golang的map,再也不贅述。github

爲何須要呢？緣由很簡單，就是：map在併發狀況虛啊，只讀是線程安全的，同時寫線程不安全，因此爲了併發安全 & 高效，官方實現了一把。sql

1.1 併發寫map會有什麼問題？

來看看不使用sync.Map的map是如何實現併發安全的：安全

func main() {
	m := map[int]int {1:1}
	go do(m)
	go do(m)

	time.Sleep(1*time.Second)
	fmt.Println(m)
}

func do (m map[int]int) {
	i := 0
	for i < 10000 {
		m[1]=1
		i++
	}
}
複製代碼

輸出：bash

fatal error: concurrent map writes
複製代碼

oh，no。報錯說的很明顯，這哥們不能同時寫。數據結構

1.2 低配版解決方案

加一把大鎖。併發

// 你們好，我是那把大鎖
var s sync.RWMutex
func main() {
	m := map[int]int {1:1}
	go do(m)
	go do(m)

	time.Sleep(1*time.Second)
	fmt.Println(m)
}

func do (m map[int]int) {
	i := 0
	for i < 10000 {
	    // 加鎖
		s.Lock()
		m[1]=1
		// 解鎖
		s.Unlock()
		i++
	}
}
複製代碼

輸出：性能

map[1:1]
複製代碼

這回終於正常了，可是會有什麼問題呢？加大鎖大機率都不是最優解，通常都會有效率問題。通俗說就是加大鎖影響其餘的元素操做了。優化

解決思路：減小加鎖時間。
方法： 1.空間換時間。  2.下降影響範圍。
複製代碼

sync.Map就是用了以上的思路。繼續往下看。

2.sync.Map如何使用？

上代碼：

func main() {
    // 關鍵人物出場
	m := sync.Map{}
	m.Store(1,1)
	go do(m)
	go do(m)

	time.Sleep(1*time.Second)
	fmt.Println(m.Load(1))
}

func do (m sync.Map) {
	i := 0
	for i < 10000 {
		m.Store(1,1)
		i++
	}
}
複製代碼

輸出：

1 true
複製代碼

運行ok。這把秀了。

3.理一理sync.Map源碼實現？

先白話文說下大概邏輯。讓下文看的更快。（大概只有是這樣流程就好） 寫：直寫。讀：先讀read，沒有再讀dirty。

從「基礎結構 + 增刪改查」的思路來詳細過一遍源碼。

3.1 基礎結構

sync.Map的核心數據結構:

type Map struct {
	mu Mutex
	read atomic.Value // readOnly
	dirty map[interface{}]*entry
	misses int
}
複製代碼

說明	類型	做用
mu	Mutex	加鎖做用。保護後文的dirty字段
read	atomic.Value	存讀的數據。由於是atomic.Value類型，只讀，因此併發是安全的。實際存的是readOnly的數據結構。
misses	int	計數做用。每次從read中讀失敗，則計數+1。
dirty	map[interface{}]*entry	包含最新寫入的數據。當misses計數達到必定值，將其賦值給read。

這裏有必要簡單描述一下，大概的邏輯，

readOnly的數據結構：

type readOnly struct {
    m  map[interface{}]*entry
    amended bool 
}
複製代碼

說明	類型	做用
m	map[interface{}]*entry	單純的map結構
amended	bool	Map.dirty的數據和這裏的 m 中的數據不同的時候，爲true

entry的數據結構：

type entry struct {
    //可見value是個指針類型，雖然read和dirty存在冗餘狀況（amended=false），可是因爲是指針類型，存儲的空間應該不是問題
    p unsafe.Pointer // *interface{}
}
複製代碼

這個結構體主要是想說明。雖然前文read和dirty存在冗餘的狀況，可是因爲value都是指針類型，其實存儲的空間其實沒增長多少。

3.2 查詢

func (m *Map) Load(key interface{}) (value interface{}, ok bool) {
    // 因read只讀，線程安全，優先讀取
    read, _ := m.read.Load().(readOnly)
    e, ok := read.m[key]
    
    // 若是read沒有，而且dirty有新數據，那麼去dirty中查找
    if !ok && read.amended {
        m.mu.Lock()
        // 雙重檢查（緣由是前文的if判斷和加鎖非原子的，懼怕這中間發生故事）
        read, _ = m.read.Load().(readOnly)
        e, ok = read.m[key]
        
        // 若是read中仍是不存在，而且dirty中有新數據
        if !ok && read.amended {
            e, ok = m.dirty[key]
            // m計數+1
            m.missLocked()
        }
        
        m.mu.Unlock()
    }
    
    if !ok {
        return nil, false
    }
    return e.load()
}

func (m *Map) missLocked() {
    m.misses++
    if m.misses < len(m.dirty) {
        return
    }
    
    // 將dirty置給read，由於穿透機率太大了(原子操做，耗時很小)
    m.read.Store(readOnly{m: m.dirty})
    m.dirty = nil
    m.misses = 0
}
複製代碼

流程圖：

這邊有幾個點須要強調一下：

如何設置閥值？

這裏採用miss計數和dirty長度的比較，來進行閥值的設定。

爲何dirty能夠直接換到read？

由於寫操做只會操做dirty，因此保證了dirty是最新的，而且數據集是確定包含read的。（可能有同窗疑問，dirty不是下一步就置爲nil了，爲什麼還包含？後文會有解釋。）

爲何dirty置爲nil？

我不肯定這個緣由。猜想：一方面是當read徹底等於dirty的時候，讀的話read沒有就是沒有了，即便穿透也是同樣的結果，因此存的沒啥用。另外一方是當存的時候，若是元素比較多，影響插入效率。

3.3 刪

func (m *Map) Delete(key interface{}) {
    // 讀出read，斷言爲readOnly類型
    read, _ := m.read.Load().(readOnly)
    e, ok := read.m[key]
    // 若是read中沒有，而且dirty中有新元素，那麼就去dirty中去找。這裏用到了amended，當read與dirty不一樣時爲true，說明dirty中有read沒有的數據。
    
    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 {
    // 若是read中存在該key，則將該value 賦值nil（採用標記的方式刪除！）
        e.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
        }
    }
}
複製代碼

流程圖：

這邊有幾個點須要強調一下：

1.爲何dirty是直接刪除，而read是標記刪除？

read的做用是在dirty前頭優先度，遇到相同元素的時候爲了避免穿透到dirty，因此採用標記的方式。同時正是由於這樣的機制+amended的標記，能夠保證read找不到&&amended=false的時候，dirty中確定找不到

2.爲何dirty是能夠直接刪除，而沒有先進行讀取存在後刪除？

刪除成本低。讀一次須要尋找，刪除也須要尋找，無需重複操做。

3.如何進行標記的？

將值置爲nil。（這個很關鍵）

3.4 增(改)

func (m *Map) Store(key, value interface{}) {
    // 若是m.read存在這個key，而且沒有被標記刪除，則嘗試更新。
    read, _ := m.read.Load().(readOnly)
    if e, ok := read.m[key]; ok && e.tryStore(&value) {
        return
    }
    
    // 若是read不存在或者已經被標記刪除
    m.mu.Lock()
    read, _ = m.read.Load().(readOnly)
   
    if e, ok := read.m[key]; ok { // read 存在該key
    // 若是entry被標記expunge，則代表dirty沒有key，可添加入dirty，並更新entry。
        if e.unexpungeLocked() { 
            // 加入dirty中，這兒是指針
            m.dirty[key] = e
        }
        // 更新value值
        e.storeLocked(&value) 
        
    } else if e, ok := m.dirty[key]; ok { // dirty 存在該key，更新
        e.storeLocked(&value)
        
    } else { // read 和 dirty都沒有
        // 若是read與dirty相同，則觸發一次dirty刷新（由於當read重置的時候，dirty已置爲nil了）
        if !read.amended { 
            // 將read中未刪除的數據加入到dirty中
            m.dirtyLocked() 
            // amended標記爲read與dirty不相同，由於後面即將加入新數據。
            m.read.Store(readOnly{m: read.m, amended: true})
        }
        m.dirty[key] = newEntry(value) 
    }
    m.mu.Unlock()
}

// 將read中未刪除的數據加入到dirty中
func (m *Map) dirtyLocked() {
    if m.dirty != nil {
        return
    }
    
    read, _ := m.read.Load().(readOnly)
    m.dirty = make(map[interface{}]*entry, len(read.m))
    
    // 遍歷read。
    for k, e := range read.m {
        // 經過這次操做，dirty中的元素都是未被刪除的，可見標記爲expunged的元素不在dirty中！！！
        if !e.tryExpungeLocked() {
            m.dirty[k] = e
        }
    }
}

// 判斷entry是否被標記刪除，而且將標記爲nil的entry更新標記爲expunge
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
}

// 對entry嘗試更新 （原子cas操做）
func (e *entry) tryStore(i *interface{}) bool {
    p := atomic.LoadPointer(&e.p)
    if p == expunged {
        return false
    }
    for {
        if atomic.CompareAndSwapPointer(&e.p, p, unsafe.Pointer(i)) {
            return true
        }
        p = atomic.LoadPointer(&e.p)
        if p == expunged {
            return false
        }
    }
}

// read裏 將標記爲expunge的更新爲nil
func (e *entry) unexpungeLocked() (wasExpunged bool) {
    return atomic.CompareAndSwapPointer(&e.p, expunged, nil)
}

// 更新entry
func (e *entry) storeLocked(i *interface{}) {
    atomic.StorePointer(&e.p, unsafe.Pointer(i))
}
複製代碼

流程圖：

這邊有幾個點須要強調一下：

read中的標記爲已刪除的區別？

標記爲nil，說明是正常的delete操做，此時dirty中不必定存在 a. dirty賦值給read後，此時dirty不存在 b. dirty初始化後，確定存在

標記爲expunged，說明是在dirty初始化的時候操做的，此時dirty中確定不存在。

可能存在性能問題？

初始化dirty的時候，雖然都是指針賦值，但read若是較大的話，可能會有些影響。

4.sync.Map的優缺點？

先說結論，後來證實。

優勢：是官方出的，是親兒子；經過讀寫分離，下降鎖時間來提升效率；缺點：不適用於大量寫的場景，這樣會致使read map讀不到數據而進一步加鎖讀取，同時dirty map也會一直晉升爲read map，總體性能較差。適用場景：大量讀，少許寫

這裏主要證實一下，爲何適合大量讀，少許寫。代碼的大概思路：經過比較單純的map和sync.Map，在併發安全的狀況下，只寫和讀寫的效率

var s sync.RWMutex
var w sync.WaitGroup
func main() {
	mapTest()
	syncMapTest()
}
func mapTest() {
	m := map[int]int {1:1}
	startTime := time.Now().Nanosecond()
	w.Add(1)
	go writeMap(m)
	w.Add(1)
	go writeMap(m)
	//w.Add(1)
	//go readMap(m)

	w.Wait()
	endTime := time.Now().Nanosecond()
	timeDiff := endTime-startTime
	fmt.Println("map:",timeDiff)
}

func writeMap (m map[int]int) {
	defer w.Done()
	i := 0
	for i < 10000 {
		// 加鎖
		s.Lock()
		m[1]=1
		// 解鎖
		s.Unlock()
		i++
	}
}

func readMap (m map[int]int) {
	defer w.Done()
	i := 0
	for i < 10000 {
		s.RLock()
		_ = m[1]
		s.RUnlock()
		i++
	}
}

func syncMapTest() {
	m := sync.Map{}
	m.Store(1,1)
	startTime := time.Now().Nanosecond()
	w.Add(1)
	go writeSyncMap(m)
	w.Add(1)
	go writeSyncMap(m)
	//w.Add(1)
	//go readSyncMap(m)

	w.Wait()
	endTime := time.Now().Nanosecond()
	timeDiff := endTime-startTime
	fmt.Println("sync.Map:",timeDiff)
}

func writeSyncMap (m sync.Map) {
	defer w.Done()
	i := 0
	for i < 10000 {
		m.Store(1,1)
		i++
	}
}

func readSyncMap (m sync.Map) {
	defer w.Done()
	i := 0
	for i < 10000 {
		m.Load(1)
		i++
	}
}
複製代碼