go中的map是并发不安全的,同时多个协程读取不会出现问题,但是多个协程 同时读写就会出现 fatal error:concurrent map read and map write的错误。通用的解决办法如下:
1. 加锁
1.1 通用锁
import "sync" type SafeMap struct { data map[string]string lock sync.Mutex } func (this *SafeMap) get(key string) string{ this.lock.Lock() defer this.lock.Unlock() return this.data[key] } func (this *SafeMap) set(key, value string) { this.lock.Lock() defer this.lock.Unlock() this.data[key] = value }
1.2 读写锁
import "sync" type SafeMap struct { data map[string]string lock sync.RWMutex } func (this *SafeMap) get(key string) string{ this.lock.RLock() defer this.lock.RUnlock() return this.data[key] } func (this *SafeMap) set(key, value string) { this.lock.Lock() defer this.lock.Unlock() this.data[key] = value }
1.3 在go1.9之后,go引入了并发安全的map: sync.map
sync.map的原理可以概括为:
1. 通过read和dirty两个字段将读写分离,读的数据存在于read字段的,最新写的数据位于dirty字段上。
2. 读取时先查询read,不存在时查询dirty,写入时只写入dirty
3. 读取read不需要加锁,而读或写dirty需要加锁
4. 使用misses字段来统计read被穿透的次数,超过一定次数将数据从dirty同步到read上
5. 删除数据通过标记来延迟删除
sync.Map结构如下所示:
type Map struct { mu Mutex //加锁,宝座dirty字段 read atomic.Value // 只读数据,实例类型为 readOnly dirty map[interface{}]*entry //最新写入的数据 misses int //read被穿透的次数 }
readOnly结构
type readOnly struct { m map[interface{}]*entry amended bool // true if the dirty map contains some key not in m. }
entery结构
type entry struct { // p == nil entry已经被删除且 dirty == nil
// p == expunged entry已经被删除,但是dirty != nil且dirty中不存在该元素,这种情况出现于重建dirty时,将read复制到dirty中,复制的过程中将nil标记为expunged,不将其复制到dirty
// 除此之外,entry存在于read中,如果dirty != nil则也存在于dirty中
p unsafe.Pointer // *interface{} }
Load()方法
func (m *Map) Load(key interface{}) (value interface{}, ok bool) {
//首先尝试从read中读取 readOnly对象 read, _ := m.read.Load().(readOnly) e, ok := read.m[key]
//如果不存在则尝试从dirty中读取 if !ok && read.amended { m.mu.Lock() //再读取一次read中内容,主要是用于防止上一步加锁过程中dirty map转换为read map导致dirty中读取不到数据 read, _ = m.read.Load().(readOnly) e, ok = read.m[key]
//如果确实不存在,则从dirty中读取 if !ok && read.amended { e, ok = m.dirty[key] // 不管dirty中存不存在,都将miss + 1, 如果misses值等于dirty中元素个数,就会把dirty中元素迁移到read中 m.missLocked() } m.mu.Unlock() } if !ok { return nil, false } return e.load() }
Store()方法
// Store sets the value for a key. func (m *Map) Store(key, value interface{}) {
//直接再read中查找 read, _ := m.read.Load().(readOnly)
//如果找到了,直接更新read中值,返回 if e, ok := read.m[key]; ok && e.tryStore(&value) { return } //如不存在,去dirty中读 m.mu.Lock()
//二次检测 read, _ = m.read.Load().(readOnly)
//如果此时读到,read中不允许直接的添加删除值,此种情况说明加锁之前存在dirty升级为read的操作 if e, ok := read.m[key]; ok {
//如果读到的值为expunged, 说明生成dirty时,复制read中的元素,对于nil的元素,搞成了expunged,所以意味着dirty不为nil,且dirty中没有该元素 if e.unexpungeLocked() { // The entry was previously expunged, which implies that there is a // non-nil dirty map and this entry is not in it.
//更新dirty中的值 m.dirty[key] = e }
//更新read中的值 e.storeLocked(&value)
//此时,read中没有该元素,需要更新dirty中的值 } else if e, ok := m.dirty[key]; ok { e.storeLocked(&value) } else {
// 如果 !read.amended, 说明dirty为nil, 需要将read map复制一份到dirty map if !read.amended { // We're adding the first new key to the dirty map. // Make sure it is allocated and mark the read-only map as incomplete. m.dirtyLocked()
//设置read.amended == true m.read.Store(readOnly{m: read.m, amended: true}) } m.dirty[key] = newEntry(value) } m.mu.Unlock() }
LoadOrStoce()
// LoadOrStore returns the existing value for the key if present. // Otherwise, it stores and returns the given value. // The loaded result is true if the value was loaded, false if stored. func (m *Map) LoadOrStore(key, value interface{}) (actual interface{}, loaded bool) { // Avoid locking if it's a clean hit.
//读取read中是否存在该key read, _ := m.read.Load().(readOnly) if e, ok := read.m[key]; ok {
//如果存在(是否标识为删除由tryLoadOrStore处理),尝试获取该元素的值,或者将值写入 actual, loaded, ok := e.tryLoadOrStore(value) if ok { return actual, loaded } } m.mu.Lock()
//二次检测 read, _ = m.read.Load().(readOnly)
//如果此时读到,read中不允许直接的添加删除值,此种情况说明加锁之前存在dirty升级为read的操作 if e, ok := read.m[key]; ok {
//如果读到的值为expunged, 说明生成dirty时,复制read中的元素,对于nil的元素,搞成了expunged,所以意味着dirty不为nil,且dirty中没有该元素 if e.unexpungeLocked() { m.dirty[key] = e }
//如果存在(是否标识为删除由tryLoadOrStore处理),尝试获取该元素的值,或者将值写入 actual, loaded, _ = e.tryLoadOrStore(value)
// 此时,read中没有元素,需要 tryLoadOrStore dirty中值 } else if e, ok := m.dirty[key]; ok { actual, loaded, _ = e.tryLoadOrStore(value) m.missLocked() } else {
// 如果 !read.amended, 说明dirty为nil, 需要将read map复制一份到dirty map if !read.amended { // We're adding the first new key to the dirty map. // Make sure it is allocated and mark the read-only map as incomplete. m.dirtyLocked() m.read.Store(readOnly{m: read.m, amended: true}) }
// 将值写入dirty中 m.dirty[key] = newEntry(value) actual, loaded = value, false } m.mu.Unlock() return actual, loaded }
// tryLoadOrStore atomically loads or stores a value if the entry is not // expunged. // // If the entry is expunged, tryLoadOrStore leaves the entry unchanged and // returns with ok==false. // 如果元素是 expunged, tryLoadOrStore 保持entry不变并直接返回false func (e *entry) tryLoadOrStore(i interface{}) (actual interface{}, loaded, ok bool) { p := atomic.LoadPointer(&e.p) // 标识删除,直接返回 if p == expunged { return nil, false, false } // 如果元素存在真实值,则直接返回该真实值 if p != nil { return *(*interface{})(p), true, true } // Copy the interface after the first load to make this method more amenable // to escape analysis: if we hit the "load" path or the entry is expunged, we // shouldn't bother heap-allocating. // 如果 p == nil, 则更新该元素值 ic := i for { if atomic.CompareAndSwapPointer(&e.p, nil, unsafe.Pointer(&ic)) { return i, false, true } p = atomic.LoadPointer(&e.p) if p == expunged { return nil, false, false } if p != nil { return *(*interface{})(p), true, true } } }
Delete()方法
// Delete deletes the value for a key. func (m *Map) Delete(key interface{}) {
// 检查read中是否存在 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 {
// dirty中删除 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 } } }
Range()方法
// Range calls f sequentially for each key and value present in the map. // If f returns false, range stops the iteration. // // Range does not necessarily correspond to any consistent snapshot of the Map's // contents: no key will be visited more than once, but if the value for any key // is stored or deleted concurrently, Range may reflect any mapping for that key // from any point during the Range call. // // Range may be O(N) with the number of elements in the map even if f returns // false after a constant number of calls. func (m *Map) Range(f func(key, value interface{}) bool) { // We need to be able to iterate over all of the keys that were already // present at the start of the call to Range. // If read.amended is false, then read.m satisfies that property without // requiring us to hold m.mu for a long time. read, _ := m.read.Load().(readOnly) // 如果 amended == true, 说明dirty中存在元素,且包含所有有效元素,此时,使用dirty map if read.amended { // m.dirty contains keys not in read.m. Fortunately, Range is already O(N) // (assuming the caller does not break out early), so a call to Range // amortizes an entire copy of the map: we can promote the dirty copy // immediately! m.mu.Lock() read, _ = m.read.Load().(readOnly) if read.amended {
//使用dirty map并将其升级为 read map read = readOnly{m: m.dirty} m.read.Store(read) m.dirty = nil m.misses = 0 } m.mu.Unlock() } // 使用read map读 for k, e := range read.m { v, ok := e.load()
// 被删除的不计入 if !ok { continue } if !f(k, v) { break } } }
当sync.Map中存在大量写操作的情况下,会导致read中读不到数据,依然会频繁加锁,同时dirty升级为read,整体性能就会很低,所以sync.Map更加适合大量读、少量写的场景。