河南国控建设集团网站,投资者教育网站建设,网站设计设计目的,网站建设企业开发公司涉及概念并发安全Map分段锁sync.MapCAS ( Compare And Swap )双检查分断锁type SimpleCache struct {mu sync.RWMutexitems map[interface{}]*simpleItem}在日常开发中#xff0c; 上述这种数据结构肯定不少见#xff0c;因为golang的原生map是非并发安全的#xff0c;所以为…涉及概念并发安全Map分段锁sync.MapCAS ( Compare And Swap )双检查分断锁type SimpleCache struct {mu sync.RWMutexitems 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 是一个加锁的并发安全Maptype 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 hashfunc fnv32(key string) uint32 {hash : uint32(2166136261)const prime32 uint32(16777619)for i : 0; i len(key); i {hash * prime32hash ^ uint32(key[i])}return hash}之后对于map的GET SET 就简单顺利成章的完成Set And Getfunc (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 : 0for i : 0; i SHARD_COUNT; i {shard : m[i]shard.RLock()count len(shard.items)shard.RUnlock()}return count}// 获取所有的keyfunc (m ConcurrentMap) Keys() []string {count : m.Count()ch : make(chan string, count)// 每一个分片启动一个协程 遍历keygo func() {wg : sync.WaitGroup{}wg.Add(SHARD_COUNT)for _, shard : range m {go func(shard *ConcurrentMapShared) {defer wg.Done()shard.RLock()// 每个分片中的key遍历后都写入统计用的channelfor key : range shard.items {ch }shard.RUnlock()}(shard)}wg.Wait()close(ch)}()keys : make([]string, count)// 统计各个协程并发读取Map分片的keyfor k : range ch {keys append(keys, k)}return keys}这里写了一个benchMark来对该分段锁Map和原生的Map加锁方式进行压测 场景为将一万个不重复的键值对同时以100万次写和100万次读分别进行5次压测 如下压测代码func BenchmarkMapShared(b *testing.B) {num : 10000testCase : 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 1000000wg : 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.Mapgo1.9之后加入了支持并发安全的Map sync.Map, sync.Map 通过一份只使用原子操作的数据和一份冗余了只读数据的加锁数据实现一定程度上的读写分离使得大多数读操作和更新操作是原子操作写入新数据才加锁的方式来提升性能。以下是 sync.Map源码剖析, 结构体中的注释都会在具体实现代码中提示相呼应type Map struct {// 保护dirty的锁mu Mutex// 只读数据(修改采用原子操作)read atomic.Value// 包含只读中所有数据(冗余)写入新数据时也在dirty中操作dirty map[interface{}]*entry// 当原子操作访问只读read时找不到数据时会去dirty中寻找此时misses1dirty及作为存储新写入的数据又冗余了只读结构中的数据所以当misses dirty 的长度时 会将dirty升级为read同时将老的dirty置nilmisses int}// Map struct 中的 read 就是readOnly 的指针type readOnly struct {// 基础Mapm map[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 Writefunc (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()// 双检查获取readread, _ m.read.Load().(readOnly)// 如果data在read中更新entryif e, ok : read.m[key]; ok {// 如果原子操作读到的数据是被标记删除的 则视为新数据写入dirtyif e.unexpungeLocked() {m.dirty[key] e}// 原子操作写新数据e.storeLocked(value)} else if e, ok : m.dirty[key]; ok {// 原子操作写新数据e.storeLocked(value)} else {// 新数据// 当dirty中没有新数据时将read中数据冗余到dirtyif !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 Readfunc (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 {// 只读中没有读取到的次数1e, ok m.dirty[key]// 检查是否达到触发dirty升级read的条件m.missLocked()}m.mu.Unlock()}if !ok {return nil, false}// atomic.Load 但被标记为删除的会返回nilreturn e.load()}func (m *Map) missLocked() {m.missesif m.misses len(m.dirty) {return}m.read.Store(readOnly{m: m.dirty})m.dirty nilm.misses 0}sync.Map DELETEfunc (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代码复制出来一份把私有成员改成可外部访问的打印指针)
