golang 併發安全Map以及分段鎖的實現

摘要： 涉及概念 併發安全Map 分段鎖 sync.Map CAS ( Compare And Swap ) 雙檢查 分斷鎖 type SimpleCache struct { musync.RWMutex items map[...

涉及概念

1. 併發安全Map
2. 分段鎖
3. sync.Map
4. CAS ( Compare And Swap )
5. 雙檢查

分斷鎖

type SimpleCache struct {
musync.RWMutex
items map[interface{}]*simpleItem
}

在日常開發中， 上述這種資料結構肯定不少見，因為golang的原生map是非併發安全的，所以為了保證map的併發安全，最簡單的方式就是給map加鎖。

之前使用過兩個本地記憶體快取的開源庫， gcache , cache2go ，其中儲存快取物件的結構都是這樣，對於輕量級的快取庫，為了設計簡潔（包含清理過期物件等 ) 再加上當需要快取大量資料時有redis,memcache等明星專案解決。 但是如果拋開這些因素遇到真正數量巨大的資料量時，直接對一個map加鎖，當map中的值越來越多，訪問map的請求越來越多，大家都競爭這一把鎖顯得併發訪問控制變重。 在go1.9引入sync.Map 之前，比較流行的做法就是使用分段鎖，顧名思義就是將鎖分段，將鎖的粒度變小，將儲存的物件分散到各個分片中，每個分片由一把鎖控制，這樣使得當需要對在A分片上的資料進行讀寫時不會影響B分片的讀寫。

分段鎖的實現

// Map 分片
type ConcurrentMap []*ConcurrentMapShared

// 每一個Map 是一個加鎖的併發安全Map
type ConcurrentMapShared struct {
items map[string]interface{}
sync.RWMutex// 各個分片Map各自的鎖
}

主流的分段鎖，即通過hash取模的方式找到當前訪問的key處於哪一個分片之上，再對該分片進行加鎖之後再讀寫。分片定位時，常用有BKDR, FNV32等hash演算法得到key的hash值。

func New() ConcurrentMap {
// SHARD_COUNT預設32個分片
m := make(ConcurrentMap, SHARD_COUNT)
for i := 0; i < SHARD_COUNT; i++ {
m[i] = &ConcurrentMapShared{
items: make(map[string]interface{}),
}
}
return m
}

在初始化好分片後， 對分片上的資料進行讀寫時就需要用hash取模進行分段定位來確認即將要讀寫的分片。

獲取段定位

func (m ConcurrentMap) GetShard(key string) *ConcurrentMapShared {
return m[uint(fnv32(key))%uint(SHARD_COUNT)]
}

// FNV hash
func fnv32(key string) uint32 {
hash := uint32(2166136261)
const prime32 = uint32(16777619)
for i := 0; i < len(key); i++ {
hash *= prime32
hash ^= uint32(key[i])
}
return hash
}

之後對於map的GET SET 就簡單順利成章的完成

Set And Get

func (m ConcurrentMap) Set(key string, value interface{}) {
shard := m.GetShard(key) // 段定位找到分片
shard.Lock()// 分片上鎖
shard.items[key] = value // 分片操作 
shard.Unlock()// 分片解鎖
}

func (m ConcurrentMap) Get(key string) (interface{}, bool) {
shard := m.GetShard(key)
shard.RLock()
val, ok := shard.items[key]
shard.RUnlock()
return val, ok
}

由此一個分段鎖Map就實現了， 但是比起普通的Map, 常用到的方法比如獲取所有key, 獲取所有Val 操作是要比原生Map複雜的，因為要遍歷每一個分片的每一個數據， 好在golang的併發特性使得解決這類問題變得非常簡單

Keys

// 統計當前分段map中item的個數
func (m ConcurrentMap) Count() int {
count := 0
for i := 0; i < SHARD_COUNT; i++ {
shard := m[i]
shard.RLock()
count += len(shard.items)
shard.RUnlock()
}
return count
}

// 獲取所有的key
func (m ConcurrentMap) Keys() []string {
count := m.Count()
ch := make(chan string, count)

// 每一個分片啟動一個協程 遍歷key
go func() {
wg := sync.WaitGroup{}
wg.Add(SHARD_COUNT)
for _, shard := range m {

go func(shard *ConcurrentMapShared) {
defer wg.Done()

shard.RLock()

// 每個分片中的key遍歷後都寫入統計用的channel
for key := range shard.items {
ch <- key
}

shard.RUnlock()
}(shard)
}
wg.Wait()
close(ch)
}()

keys := make([]string, count)
// 統計各個協程併發讀取Map分片的key
for k := range ch {
keys = append(keys, k)
}
return keys
}

這裡寫了一個benchMark來對該分段鎖Map和原生的Map加鎖方式進行壓測， 場景為將一萬個不重複的鍵值對同時以100萬次寫和100萬次讀，分別進行5次壓測， 如下壓測程式碼

func BenchmarkMapShared(b *testing.B) {
num := 10000
testCase := genNoRepetTestCase(num) // 10000個不重複的鍵值對
m := New()
for _, v := range testCase {
m.Set(v.Key, v.Val)
}
b.ResetTimer()

for i := 0; i < 5; i++ {
b.Run(strconv.Itoa(i), func(b *testing.B) {

b.N = 1000000

wg := sync.WaitGroup{}
wg.Add(b.N * 2)
for i := 0; i < b.N; i++ {
e := testCase[rand.Intn(num)]

go func(key string, val interface{}) {
m.Set(key, val)
wg.Done()
}(e.Key, e.Val)

go func(key string) {
_, _ = m.Get(key)
wg.Done()
}(e.Key)

}
wg.Wait()
})
}
}

原生Map加鎖壓測結果

分段鎖壓測結果

可以看出在將鎖的粒度細化後再面對大量需要控制併發安全的訪問時，分段鎖Map的耗時比原生Map加鎖要快3倍有餘

Sync.Map

go1.9之後加入了支援併發安全的Map sync.Map , sync.Map 通過一份只使用原子操作的資料和一份冗餘了只讀資料的加鎖資料實現一定程度上的讀寫分離，使得大多數讀操作和更新操作是原子操作，寫入新資料才加鎖的方式來提升效能。以下是 sync.Map原始碼剖析, 結構體中的註釋都會在具體實現程式碼中提示相呼應

type Map struct {
// 保護dirty的鎖
mu Mutex
// 只讀資料（修改採用原子操作)
read atomic.Value
// 包含只讀中所有資料（冗餘），寫入新資料時也在dirty中操作
dirty map[interface{}]*entry
// 當原子操作訪問只讀read時找不到資料時會去dirty中尋找，此時misses+1，dirty及作為儲存新寫入的資料，又冗餘了只讀結構中的資料，所以當misses > dirty 的長度時， 會將dirty升級為read，同時將老的dirty置nil
misses int 
}

// Map struct 中的 read 就是readOnly 的指標
type readOnly struct {
// 基礎Map
mmap[interface{}]*entry 
// 用於表示當前dirty中是否有read中不存在的資料， 在寫入資料時， 如果發現dirty中沒有新資料且dirty為nil時，會將read中未被刪除的資料拷貝一份冗餘到dirty中， 過程與Map struct中的 misses相呼應
amended bool 
}

// 資料項
type entry struct {
p unsafe.Pointer 
}

// 用於標記資料項已被刪除（主要保證資料冗餘時的併發安全）
// 上述Map結構中說到有一個將read資料拷貝冗餘至dirty的過程， 因為刪除資料項是將*entry置nil, 為了避免冗餘過程中因併發問題導致*entry改變而影響到拷貝後的dirty正確性，所以sync.Map使用expunged來標記entry是否被刪除
var expunged = unsafe.Pointer(new(interface{}))

在下面sync.Map具體實現中將會看到很多“雙檢查”程式碼，因為通過原子操作獲取的值可能在進行其他非原子操作過程中已改變，所以再非原子操作後需要使用之前原子操作獲取的值需要再次進行原子操作獲取。

compareAndSwap 交換並比較， 用於在多執行緒程式設計中實現不被打斷的資料交換操作，從而避免多執行緒同時改寫某一資料時導致資料不一致問題。

sync.Map Write

func (m *Map) Store(key, value interface{}) {
// 先不上鎖，而是從只讀資料中按key讀取， 如果已存在以compareAndSwap操作進行覆蓋(update)
read, _ := m.read.Load().(readOnly)
if e, ok := read.m[key]; ok && e.tryStore(&value) {
return
}

m.mu.Lock()
// 雙檢查獲取read
read, _ = m.read.Load().(readOnly)
// 如果data在read中，更新entry
if e, ok := read.m[key]; ok {
// 如果原子操作讀到的資料是被標記刪除的， 則視為新資料寫入dirty
if e.unexpungeLocked() {
m.dirty[key] = e
}
// 原子操作寫新資料
e.storeLocked(&value)
} else if e, ok := m.dirty[key]; ok {
// 原子操作寫新資料
e.storeLocked(&value)
} else {
// 新資料 
// 當dirty中沒有新資料時，將read中資料冗餘到dirty
if !read.amended {
m.dirtyLocked()
m.read.Store(readOnly{m: read.m, amended: true})
}

m.dirty[key] = newEntry(value)
}
m.mu.Unlock()
}

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
}
}
}


// 在dirty中沒有比read多出的新資料時觸發冗餘
func (m *Map) dirtyLocked() {
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是否被刪除， 被刪除的資料不冗餘
if !e.tryExpungeLocked() {
m.dirty[k] = e
}
}
}

func (e *entry) tryExpungeLocked() (isExpunged bool) {
p := atomic.LoadPointer(&e.p)
for p == nil {
// 將被刪除（置nil）的資料以cas原子操作標記為expunged（防止因併發情況下其他操作導致冗餘進dirty的資料不正確）
if atomic.CompareAndSwapPointer(&e.p, nil, expunged) {
return true
}
p = atomic.LoadPointer(&e.p)
}
return p == expunged
}

sync.Map Read

func (m *Map) Load(key interface{}) (value interface{}, ok bool) {
read, _ := m.read.Load().(readOnly)
e, ok := read.m[key]

// 只讀資料中沒有，並且dirty有比read多的資料，加鎖在dirty中找
if !ok && read.amended {
m.mu.Lock()
// 雙檢查， 因為上鎖之前的語句是非原子性的
read, _ = m.read.Load().(readOnly)
e, ok = read.m[key]
if !ok && read.amended {
// 只讀中沒有讀取到的次數+1
e, ok = m.dirty[key]
// 檢查是否達到觸發dirty升級read的條件
m.missLocked()
}
m.mu.Unlock()
}
if !ok {
return nil, false
}
// atomic.Load 但被標記為刪除的會返回nil
return e.load()
}

func (m *Map) missLocked() {
m.misses++
if m.misses < len(m.dirty) {
return
}
m.read.Store(readOnly{m: m.dirty})
m.dirty = nil
m.misses = 0
}

sync.Map DELETE

func (m *Map) Delete(key interface{}) {
read, _ := m.read.Load().(readOnly)
e, ok := read.m[key]
// 只讀中不存在需要到dirty中去刪除
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 {
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
}
}
}

同樣以剛剛壓測原生加鎖Map和分段鎖的方式來壓測sync.Map

壓測平均下來sync.Map和分段鎖差別不大，但是比起分段鎖， sync.Map則將鎖的粒度更加的細小到對資料的狀態上，使得大多資料可以無鎖化操作， 同時比分段鎖擁有更好的拓展性，因為分段鎖使用前總是要定一個分片數量， 在做擴容或者縮小時很麻煩, 但要達到sync.Map這種效能既好又能動態擴容的程度，程式碼就相對複雜很多。

還有注意在使用sync.Map時切忌不要將其拷貝， go原始碼中有對sync.Map註釋到” A Map must not be copied after first use.”因為當sync.Map被拷貝之後， Map型別的dirty還是那個map 但是read 和 鎖卻不是之前的read和鎖(都不在一個世界你拿什麼保護我), 所以必然導致併發不安全（為了寫博我把sync.Map程式碼複製出來一份把私有成員改成可外部訪問的列印指標）