golang 并發(fā)安全Map以及分段鎖的實(shí)現(xiàn)方法
涉及概念
- 并發(fā)安全Map
- 分段鎖
- sync.Map
- CAS ( Compare And Swap )
- 雙檢查
分?jǐn)噫i
type SimpleCache struct {
mu sync.RWMutex
items map[interface{}]*simpleItem
}
在日常開發(fā)中， 上述這種數(shù)據(jù)結(jié)構(gòu)肯定不少見，因?yàn)間olang的原生map是非并發(fā)安全的，所以為了保證map的并發(fā)安全，最簡單的方式就是給map加鎖。
之前使用過兩個(gè)本地內(nèi)存緩存的開源庫， gcache, cache2go，其中存儲(chǔ)緩存對象的結(jié)構(gòu)都是這樣，對于輕量級(jí)的緩存庫，為了設(shè)計(jì)簡潔（包含清理過期對象等 ) 再加上當(dāng)需要緩存大量數(shù)據(jù)時(shí)有redis,memcache等明星項(xiàng)目解決。 但是如果拋開這些因素遇到真正數(shù)量巨大的數(shù)據(jù)量時(shí)，直接對一個(gè)map加鎖，當(dāng)map中的值越來越多，訪問map的請求越來越多，大家都競爭這一把鎖顯得并發(fā)訪問控制變重。 在go1.9引入sync.Map 之前，比較流行的做法就是使用分段鎖，顧名思義就是將鎖分段，將鎖的粒度變小，將存儲(chǔ)的對象分散到各個(gè)分片中，每個(gè)分片由一把鎖控制，這樣使得當(dāng)需要對在A分片上的數(shù)據(jù)進(jìn)行讀寫時(shí)不會(huì)影響B(tài)分片的讀寫。
分段鎖的實(shí)現(xiàn)
// Map 分片
type ConcurrentMap []*ConcurrentMapShared
// 每一個(gè)Map 是一個(gè)加鎖的并發(fā)安全Map
type ConcurrentMapShared struct {
items map[string]interface{}
sync.RWMutex // 各個(gè)分片Map各自的鎖
}
主流的分段鎖，即通過hash取模的方式找到當(dāng)前訪問的key處于哪一個(gè)分片之上，再對該分片進(jìn)行加鎖之后再讀寫。分片定位時(shí)，常用有BKDR, FNV32等hash算法得到key的hash值。
func New() ConcurrentMap {
// SHARD_COUNT 默認(rèn)32個(gè)分片
m := make(ConcurrentMap, SHARD_COUNT)
for i := 0; i < SHARD_COUNT; i++ {
m[i] = &ConcurrentMapShared{
items: make(map[string]interface{}),
}
}
return m
}
在初始化好分片后， 對分片上的數(shù)據(jù)進(jìn)行讀寫時(shí)就需要用hash取模進(jìn)行分段定位來確認(rèn)即將要讀寫的分片。
獲取段定位
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
}
由此一個(gè)分段鎖Map就實(shí)現(xiàn)了， 但是比起普通的Map, 常用到的方法比如獲取所有key, 獲取所有Val 操作是要比原生Map復(fù)雜的，因?yàn)橐闅v每一個(gè)分片的每一個(gè)數(shù)據(jù)， 好在golang的并發(fā)特性使得解決這類問題變得非常簡單
Keys
// 統(tǒng)計(jì)當(dāng)前分段map中item的個(gè)數(shù)
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)
// 每一個(gè)分片啟動(dòng)一個(gè)協(xié)程 遍歷key
go func() {
wg := sync.WaitGroup{}
wg.Add(SHARD_COUNT)
for _, shard := range m {
go func(shard *ConcurrentMapShared) {
defer wg.Done()
shard.RLock()
// 每個(gè)分片中的key遍歷后都寫入統(tǒng)計(jì)用的channel
for key := range shard.items {
ch <- key
}
shard.RUnlock()
}(shard)
}
wg.Wait()
close(ch)
}()
keys := make([]string, count)
// 統(tǒng)計(jì)各個(gè)協(xié)程并發(fā)讀取Map分片的key
for k := range ch {
keys = append(keys, k)
}
return keys
}
這里寫了一個(gè)benchMark來對該分段鎖Map和原生的Map加鎖方式進(jìn)行壓測， 場景為將一萬個(gè)不重復(fù)的鍵值對同時(shí)以100萬次寫和100萬次讀，分別進(jìn)行5次壓測， 如下壓測代碼
func BenchmarkMapShared(b *testing.B) {
num := 10000
testCase := genNoRepetTestCase(num) // 10000個(gè)不重復(fù)的鍵值對
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加鎖壓測結(jié)果
分段鎖壓測結(jié)果
可以看出在將鎖的粒度細(xì)化后再面對大量需要控制并發(fā)安全的訪問時(shí)，分段鎖Map的耗時(shí)比原生Map加鎖要快3倍有余
Sync.Map
go1.9之后加入了支持并發(fā)安全的Map sync.Map, sync.Map 通過一份只使用原子操作的數(shù)據(jù)和一份冗余了只讀數(shù)據(jù)的加鎖數(shù)據(jù)實(shí)現(xiàn)一定程度上的讀寫分離，使得大多數(shù)讀操作和更新操作是原子操作，寫入新數(shù)據(jù)才加鎖的方式來提升性能。以下是 sync.Map源碼剖析, 結(jié)構(gòu)體中的注釋都會(huì)在具體實(shí)現(xiàn)代碼中提示相呼應(yīng)
type Map struct {
// 保護(hù)dirty的鎖
mu Mutex
// 只讀數(shù)據(jù)（修改采用原子操作)
read atomic.Value
// 包含只讀中所有數(shù)據(jù)（冗余），寫入新數(shù)據(jù)時(shí)也在dirty中操作
dirty map[interface{}]*entry
// 當(dāng)原子操作訪問只讀read時(shí)找不到數(shù)據(jù)時(shí)會(huì)去dirty中尋找，此時(shí)misses+1，dirty及作為存儲(chǔ)新寫入的數(shù)據(jù)，又冗余了只讀結(jié)構(gòu)中的數(shù)據(jù)，所以當(dāng)misses > dirty 的長度時(shí)， 會(huì)將dirty升級(jí)為read，同時(shí)將老的dirty置nil
misses int
}
// Map struct 中的 read 就是readOnly 的指針
type readOnly struct {
// 基礎(chǔ)Map
m map[interface{}]*entry
// 用于表示當(dāng)前dirty中是否有read中不存在的數(shù)據(jù)， 在寫入數(shù)據(jù)時(shí)， 如果發(fā)現(xiàn)dirty中沒有新數(shù)據(jù)且dirty為nil時(shí)，會(huì)將read中未被刪除的數(shù)據(jù)拷貝一份冗余到dirty中， 過程與Map struct中的 misses相呼應(yīng)
amended bool
}
// 數(shù)據(jù)項(xiàng)
type entry struct {
p unsafe.Pointer
}
// 用于標(biāo)記數(shù)據(jù)項(xiàng)已被刪除（主要保證數(shù)據(jù)冗余時(shí)的并發(fā)安全）
// 上述Map結(jié)構(gòu)中說到有一個(gè)將read數(shù)據(jù)拷貝冗余至dirty的過程， 因?yàn)閯h除數(shù)據(jù)項(xiàng)是將*entry置nil, 為了避免冗余過程中因并發(fā)問題導(dǎo)致*entry改變而影響到拷貝后的dirty正確性，所以sync.Map使用expunged來標(biāo)記entry是否被刪除
var expunged = unsafe.Pointer(new(interface{}))
在下面sync.Map具體實(shí)現(xiàn)中將會(huì)看到很多“雙檢查”代碼，因?yàn)橥ㄟ^原子操作獲取的值可能在進(jìn)行其他非原子操作過程中已改變，所以再非原子操作后需要使用之前原子操作獲取的值需要再次進(jìn)行原子操作獲取。
compareAndSwap 交換并比較， 用于在多線程編程中實(shí)現(xiàn)不被打斷的數(shù)據(jù)交換操作，從而避免多線程同時(shí)改寫某一數(shù)據(jù)時(shí)導(dǎo)致數(shù)據(jù)不一致問題。
sync.Map Write
func (m *Map) Store(key, value interface{}) {
// 先不上鎖，而是從只讀數(shù)據(jù)中按key讀取， 如果已存在以compareAndSwap操作進(jìn)行覆蓋(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 {
// 如果原子操作讀到的數(shù)據(jù)是被標(biāo)記刪除的， 則視為新數(shù)據(jù)寫入dirty
if e.unexpungeLocked() {
m.dirty[key] = e
}
// 原子操作寫新數(shù)據(jù)
e.storeLocked(&value)
} else if e, ok := m.dirty[key]; ok {
// 原子操作寫新數(shù)據(jù)
e.storeLocked(&value)
} else {
// 新數(shù)據(jù)
// 當(dāng)dirty中沒有新數(shù)據(jù)時(shí)，將read中數(shù)據(jù)冗余到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多出的新數(shù)據(jù)時(shí)觸發(fā)冗余
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是否被刪除， 被刪除的數(shù)據(jù)不冗余
if !e.tryExpungeLocked() {
m.dirty[k] = e
}
}
}
func (e *entry) tryExpungeLocked() (isExpunged bool) {
p := atomic.LoadPointer(&e.p)
for p == nil {
// 將被刪除（置nil）的數(shù)據(jù)以cas原子操作標(biāo)記為expunged（防止因并發(fā)情況下其他操作導(dǎo)致冗余進(jìn)dirty的數(shù)據(jù)不正確）
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]
// 只讀數(shù)據(jù)中沒有，并且dirty有比read多的數(shù)據(jù)，加鎖在dirty中找
if !ok && read.amended {
m.mu.Lock()
// 雙檢查， 因?yàn)樯湘i之前的語句是非原子性的
read, _ = m.read.Load().(readOnly)
e, ok = read.m[key]
if !ok && read.amended {
// 只讀中沒有讀取到的次數(shù)+1
e, ok = m.dirty[key]
// 檢查是否達(dá)到觸發(fā)dirty升級(jí)read的條件
m.missLocked()
}
m.mu.Unlock()
}
if !ok {
return nil, false
}
// atomic.Load 但被標(biāo)記為刪除的會(huì)返回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()
// 雙檢查， 因?yàn)樯湘i之前的語句是非原子性的
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則將鎖的粒度更加的細(xì)小到對數(shù)據(jù)的狀態(tài)上，使得大多數(shù)據(jù)可以無鎖化操作， 同時(shí)比分段鎖擁有更好的拓展性，因?yàn)榉侄捂i使用前總是要定一個(gè)分片數(shù)量， 在做擴(kuò)容或者縮小時(shí)很麻煩, 但要達(dá)到sync.Map這種性能既好又能動(dòng)態(tài)擴(kuò)容的程度，代碼就相對復(fù)雜很多。
還有注意在使用sync.Map時(shí)切忌不要將其拷貝， go源碼中有對sync.Map注釋到” A Map must not be copied after first use.”因?yàn)楫?dāng)sync.Map被拷貝之后， Map類型的dirty還是那個(gè)map 但是read 和 鎖卻不是之前的read和鎖(都不在一個(gè)世界你拿什么保護(hù)我), 所以必然導(dǎo)致并發(fā)不安全（為了寫博我把sync.Map代碼復(fù)制出來一份把私有成員改成可外部訪問的打印指針）
以上就是本文的全部內(nèi)容，希望對大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。
相關(guān)文章
這篇文章主要介紹了Golang中的time.Duration類型用法說明，具有很好的參考價(jià)值，希望對大家有所幫助。一起跟隨小編過來看看吧2020-12-12
Go語言提供了豐富的日期與時(shí)間處理函數(shù),涵蓋了從獲取當(dāng)前時(shí)間到格式化、時(shí)區(qū)轉(zhuǎn)換、定時(shí)器和計(jì)時(shí)器的功能,這篇文章主要給大家介紹了關(guān)于Go語言中日期與時(shí)間用法的相關(guān)資料,需要的朋友可以參考下2024-06-06
這篇文章主要介紹了golang 中的 nil,本文通過多種場景分析給大家介紹的非常詳細(xì)，對大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友可以參考下2021-03-03
在Golang中,切片的淺拷貝只復(fù)制指向?qū)ο蟮闹羔?而深拷貝則復(fù)制數(shù)據(jù)本身,文中通過示例代碼介紹的非常詳細(xì),對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2024-10-10
這篇文章主要為大家介紹了Go語言快速入門指針Map示例教程，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪2022-08-08
一文搞懂Golang文件操作增刪改查功能(基礎(chǔ)篇)
這篇文章主要介紹了一文搞懂Golang文件操作增刪改查功能(基礎(chǔ)篇),Golang 可以認(rèn)為是服務(wù)器開發(fā)語言發(fā)展的趨勢之一，特別是在流媒體服務(wù)器開發(fā)中，已經(jīng)占有一席之地，今天我們不聊特別深?yuàn)W的機(jī)制和內(nèi)容，就來聊一聊 Golang 對于文件的基本操作2021-04-04
由于最近偏愛Ubuntu，在加上作為一門開源語言，在Linux上從源代碼開始搭建環(huán)境更讓人覺得有趣味性。讓我們直接先從Go語言的環(huán)境搭建開始2015-04-04
這篇文章主要介紹了第一個(gè)Go程序-Hello?World，在編寫第一個(gè)go程序之前，我們要將系統(tǒng)的環(huán)境變量配好，下面來看具體的編一過程吧，需要的小伙伴可以參考一下2022-01-01
這篇文章主要介紹了Golang 1.16 中 Modules的主要變化更新,本文給大家介紹的非常詳細(xì)，對大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友可以參考下2021-02-02
這篇文章主要介紹了Go語言二維數(shù)組的傳參方式，具有很好的參考價(jià)值，希望對大家有所幫助。一起跟隨小編過來看看吧2021-04-04