golang并发安全Map以及分段锁的实现方法

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

// Map 分片
type ConcurrentMap []*ConcurrentMapShared

// 每一个Map 是一个加锁的并发安全Map
type ConcurrentMapShared struct {
  items map[string]interface{}
  sync.RWMutex  // 各个分片Map各自的锁
}

func New() ConcurrentMap {
  // SHARD_COUNT 默认32个分片
  m := make(ConcurrentMap, SHARD_COUNT)
  for i := 0; i 

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 

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中item的个数
func (m ConcurrentMap) Count() int {
  count := 0
  for i := 0; i 

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 

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
  m  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{}))

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
}

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 

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