go sync.Map源碼分析

概述

go 語言中的map並非併發安全的,在Go 1.6以前,併發讀寫map會致使讀取到髒數據,在1.6以後則程序直接panic. 所以以前的解決方案通常都是經過引入RWMutex(讀寫鎖)進行處理, 關於go爲何不支持map的原子操做,概況來講,對map原子操做必定程度上下降了只有併發讀,或不存在併發讀寫等場景的性能. 但做爲服務端來講,使用go編寫服務,大部分狀況下都會存在gorutine併發訪問map的狀況,所以,1.9以後,go 在sync包下引入了併發安全的map. 這裏將從源碼對其進行解讀.

1. sync.Map提供的方法

• 存儲數據,存入key以及value能夠爲任意類型.

func (m *Map) Store(key, value interface{})
複製代碼

• 刪除對應key

func (m *Map) Delete(key interface{})
複製代碼

• 讀取對應key的值,ok表示是否在map中查詢到key

func (m *Map) Load(key interface{}) (value interface{}, ok bool)
複製代碼

• 針對某個key的存在讀取不存在就存儲,loaded爲true表示存在值,false表示不存在值.

func (m *Map) LoadOrStore(key, value interface{}) (actual interface{}, loaded bool)
複製代碼

• 表示對全部key進行遍歷,並將遍歷出的key,value傳入回調函數進行函數調用,回調函數返回false時遍歷結束,不然遍歷完全部key.

func (m *Map) Range(f func(key, value interface{}) bool)
複製代碼

2. 原理

經過引入兩個map,將讀寫分離到不一樣的map,其中read map只提供讀,而dirty map則負責寫. 這樣read map就能夠在不加鎖的狀況下進行併發讀取,當read map中沒有讀取到值時,再加鎖進行後續讀取,並累加未命中數,當未命中數到達必定數量後,將dirty map上升爲read map.

另外，雖然引入了兩個map，可是底層數據存儲的是指針，指向的是同一份值．

具體流程： 如插入key 1,2,3時均插入了dirty map中，此時read map沒有key值，讀取時從dirty map中讀取，並記錄miss數

當miss數大於等於dirty map的長度時，將dirty map直接升級爲read map,這裏直接 對dirty map進行地址拷貝．

當有新的key 4插入時，將read map中的key值拷貝到dirty map中，這樣dirty map就含有全部的值，下次升級爲read map時直接進行地址拷貝．

3. 源碼分析

3.1 主要結構

entry結構,用於保存value的interface指針,經過atomic進行原子操做.

type entry struct {
	p unsafe.Pointer // *interface{}
}
複製代碼

Map結構, 主結構,提供對外的方法,以及數據存儲.

type Map struct {
	mu Mutex    

    //存儲readOnly,不加鎖的狀況下,對其進行併發讀取
	read atomic.Value // readOnly

    //dirty map用於存儲寫入的數據,能直接升級成read map.
	dirty map[interface{}]*entry

    //misses 主要記錄read讀取不到數據加鎖讀取read map以及dirty map的次數.
	misses int
}
複製代碼

readOnly 結構, 主要用於存儲

// readOnly 經過原子操做存儲在Map.read中, 
type readOnly struct {
	m       map[interface{}]*entry
	amended bool // true if the dirty map contains some key not in m.
}

複製代碼

3.1 Load方法

func (m *Map) Load(key interface{}) (value interface{}, ok bool) {
	read, _ := m.read.Load().(readOnly)
	e, ok := read.m[key]
	if !ok && read.amended {
		m.mu.Lock()
		//加鎖,而後再讀取一遍read map中內容,主要防止在加鎖的過程當中,dirty map轉換成read map,從而致使讀取不到數據.
        read, _ = m.read.Load().(readOnly)
		e, ok = read.m[key]
		if !ok && read.amended {
			e, ok = m.dirty[key]
			//記錄miss數, 在dirty map提高爲read map以前,
            //這個key值都必須在加鎖的狀況下在dirty map中讀取到.
			m.missLocked()
		}
		m.mu.Unlock()
	}
	if !ok {
		return nil, false
	}
	return e.load()
}
複製代碼

3.2 Store方法

// Store sets the value for a key.
func (m *Map) Store(key, value interface{}) {
    //若是在read map讀取到值,則嘗試使用原子操做直接對值進行更新,更新成功則返回
	read, _ := m.read.Load().(readOnly)
	if e, ok := read.m[key]; ok && e.tryStore(&value) {
		return
	}

    //若是未在read map中讀取到值或讀取到值進行更新時更新失敗,則加鎖進行後續處理
	m.mu.Lock()
	read, _ = m.read.Load().(readOnly)
	if e, ok := read.m[key]; ok {
        //在檢查一遍read,若是讀取到的值處於刪除狀態,將值寫入dirty map中
		if e.unexpungeLocked() {
			m.dirty[key] = e
		}
        //使用原子操做更新key對應的值
		e.storeLocked(&value)
	} else if e, ok := m.dirty[key]; ok {
        //若是在dirty map中讀取到值，則直接使用原子操做更新值
		e.storeLocked(&value)
	} else {
        //若是dirty map中不含有值，則說明dirty map已經升級爲read map,或者第一次進入
        //須要初始化dirty map，並將read map的key添加到新建立的dirty map中．
		if !read.amended {
			m.dirtyLocked()
			m.read.Store(readOnly{m: read.m, amended: true})
		}
		m.dirty[key] = newEntry(value)
	}
	m.mu.Unlock()
}
複製代碼

3.3 LoadOrStore方法

代碼邏輯和Store相似

func (m *Map) LoadOrStore(key, value interface{}) (actual interface{}, loaded bool) {
	// 不加鎖的狀況下讀取read map
	read, _ := m.read.Load().(readOnly)
	if e, ok := read.m[key]; ok {
        //若是讀取到值則嘗試對值進行更新或讀取
		actual, loaded, ok := e.tryLoadOrStore(value)
		if ok {
			return actual, loaded
		}
	}

	m.mu.Lock()
	read, _ = m.read.Load().(readOnly)
    // 在加鎖的請求下在肯定一次read map
	if e, ok := read.m[key]; ok {
		if e.unexpungeLocked() {
			m.dirty[key] = e
		}
		actual, loaded, _ = e.tryLoadOrStore(value)
	} else if e, ok := m.dirty[key]; ok {
		actual, loaded, _ = e.tryLoadOrStore(value)
		m.missLocked()
	} else {
		if !read.amended {
			m.dirtyLocked()
			m.read.Store(readOnly{m: read.m, amended: true})
		}
		m.dirty[key] = newEntry(value)
		actual, loaded = value, false
	}
	m.mu.Unlock()

	return actual, loaded
}
複製代碼

3.4 Range 方法

func (m *Map) Range(f func(key, value interface{}) bool) {

    //先獲取read map中值
	read, _ := m.read.Load().(readOnly)
    //若是dirty map中還有值，則進行加鎖檢測
	if read.amended {
		m.mu.Lock()
		read, _ = m.read.Load().(readOnly)
        
		if read.amended {
            //將dirty map中賦給read,由於dirty map包含了全部的值
			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
		}
	}
}
複製代碼

3.5 Delete 方法

func (m *Map) Delete(key interface{}) {
    //首先獲取read map
	read, _ := m.read.Load().(readOnly)
	e, ok := read.m[key]
	if !ok && read.amended {
		m.mu.Lock()
        //加鎖二次檢測
		read, _ = m.read.Load().(readOnly)
		e, ok = read.m[key]
        //沒有在read map中獲取到值，到dirty map中刪除
		if !ok && read.amended {
			delete(m.dirty, key)
		}
		m.mu.Unlock()
	}
	if ok {
		e.delete()
	}
}
複製代碼

4. 侷限性

從以上的源碼可知,sync.map並不適合同時存在大量讀寫的場景,大量的寫會致使read map讀取不到數據從而加鎖進行進一步讀取,同時dirty map不斷升級爲read map. 從而致使總體性能較低,特別是針對cache場景.針對append-only以及大量讀,少許寫場景使用sync.map則相對比較合適.

對於map,還有一種基於hash的實現思路,具體就是對map加讀寫鎖,可是分配n個map,根據對key作hash運算肯定是分配到哪一個map中. 這樣鎖的消耗就降到了1/n(理論值).具體實現可見:concurrent-map

相比之下, 基於hash的方式更容易理解,總體性能較穩定. sync.map在某些場景性能可能差一些,但某些場景卻能取得更好的效果. 因此仍是要根據具體的業務場景進行取捨.

5. 參考

go
sync.map