Go实现并发安全字典

KV 内存数据库的核心是并发安全的哈希表，常见的设计有几种：

• sync.map: golang 官方提供的并发哈希表, 适合读多写少的场景。但是在 m.dirty 刚被提升后会将 m.read 复制到新的 m.dirty 中，在数据量较大的情况下复制操作会阻塞所有协程，存在较大的隐患。关于 sync.map 的详细讨论推荐阅读鸟窝:sync.Map揭秘

• juc.ConcurrentHashMap: java 的并发哈希表采用分段锁实现。在进行扩容时访问哈希表线程都将协助进行 rehash 操作，在 rehash 结束前所有的读写操作都会阻塞。因为缓存数据库中键值对数量巨大且对读写操作响应时间要求较高，使用juc的策略是不合适的。

• memcached hashtable: 在后台线程进行 rehash 操作时，主线程会判断要访问的哈希槽是否已被 rehash 从而决定操作 old_hashtable 还是操作 primary_hashtable。

memcached hashtable 的渐进式 Rehash 策略使主线程和rehash线程之间的 data race 限制在哈希槽内，最小化rehash操作对读写操作的影响，这是最理想的实现方式。但由于才疏学浅无法使用 golang 实现该策略故忍痛放弃（主要原因在于 golang 没有 volatile 关键字, 保证可见性的操作非常复杂），欢迎各位读者讨论。

这里 golang 社区广泛使用的分段锁策略。我们将 key 分散到固定数量的 shard 中避免 rehash 操作。shard 是有锁保护的 map, 当 shard 进行 rehash 时会阻塞shard内的读写，但不会对其他 shard 造成影响。

接口

// Consumer is used to traversal dict, if it returns false the traversal will be breaktype 
Consumer func(key string, val interface{}) bool  
  
// Dict is interface of a key-value data structuretype Dict interface {  
   Get(key string) (val interface{}, exists bool)  
   Len() int  
   Put(key string, val interface{}) (result int)  
   PutIfAbsent(key string, val interface{}) (result int)  
   PutIfExists(key string, val interface{}) (result int)  
   Remove(key string) (result int)  
   ForEach(consumer Consumer)  
   Keys() []string  
   RandomKeys(limit int) []string  
   RandomDistinctKeys(limit int) []string  
   Clear()  
}

实现类：

// ConcurrentDict is thread safe map using sharding lock
type ConcurrentDict struct {  
   table      []*shard  
   count      int32  
   shardCount int  
}  

// 分段哈希槽
type shard struct {  
   m     map[string]interface{}  
   mutex sync.RWMutex  
}  
// 计算容量
func computeCapacity(param int) (size int) {  
   if param <= 16 {  
      return 16  
   }  
   n := param - 1  
   n |= n >> 1  
   n |= n >> 2  
   n |= n >> 4  
   n |= n >> 8  
   n |= n >> 16  
   if n < 0 {  
      return math.MaxInt32  
   }  
   return n + 1  
}  
  
// MakeConcurrent creates ConcurrentDict with the given shard count
// 初始化
func MakeConcurrent(shardCount int) *ConcurrentDict {  
   shardCount = computeCapacity(shardCount)  
   table := make([]*shard, shardCount)  
   for i := 0; i < shardCount; i++ {  
      table[i] = &shard{  
         m: make(map[string]interface{}),  
      }  
   }  
   d := &ConcurrentDict{  
      count:      0,  
      table:      table,  
      shardCount: shardCount,  
   }  
   return d  
}  
  
const prime32 = uint32(16777619)  
// fnv哈希算法
func fnv32(key string) uint32 {  
   hash := uint32(2166136261)  
   for i := 0; i < len(key); i++ {  
      hash *= prime32  
      hash ^= uint32(key[i])  
   }  
   return hash  
}  

// 根据哈希值计算哈希槽索引位置
func (dict *ConcurrentDict) spread(hashCode uint32) uint32 {  
   if dict == nil {  
      panic("dict is nil")  
   }  
   tableSize := uint32(len(dict.table))  
   return (tableSize - 1) & uint32(hashCode)  
}  

// 根据索引获取哈希槽对象
func (dict *ConcurrentDict) getShard(index uint32) *shard {  
   if dict == nil {  
      panic("dict is nil")  
   }  
   return dict.table[index]  
}  
  
// Get returns the binding value and whether the key is exist
// 根据key获取值，返回值和是否存在的bool值
func (dict *ConcurrentDict) Get(key string) (val interface{}, exists bool) {  
   if dict == nil {  
      panic("dict is nil")  
   }  
   hashCode := fnv32(key)  
   index := dict.spread(hashCode)  
   shard := dict.getShard(index)  
   shard.mutex.RLock()  
   defer shard.mutex.RUnlock()  
   val, exists = shard.m[key]  
   return  
}  
  
// Len returns the number of dict
// 获取字典长度
func (dict *ConcurrentDict) Len() int {  
   if dict == nil {  
      panic("dict is nil")  
   }  
   return int(atomic.LoadInt32(&dict.count))  
}  
  
// Put puts key value into dict and returns the number of new inserted key-value
// 添加值
func (dict *ConcurrentDict) Put(key string, val interface{}) (result int) {  
   if dict == nil {  
      panic("dict is nil")  
   }  
   hashCode := fnv32(key)  
   index := dict.spread(hashCode)  
   shard := dict.getShard(index)  
   shard.mutex.Lock()  
   defer shard.mutex.Unlock()  
  
   if _, ok := shard.m[key]; ok {  
      shard.m[key] = val  
      return 0  
   }  
   shard.m[key] = val  
   dict.addCount()  
   return 1  
}  
  
// PutIfAbsent puts value if the key is not exists and returns the number of updated key-value
// 如果key不存在，则添加该key
// 如果key存在，则什么也不做
func (dict *ConcurrentDict) PutIfAbsent(key string, val interface{}) (result int) {  
   if dict == nil {  
      panic("dict is nil")  
   }  
   hashCode := fnv32(key)  
   index := dict.spread(hashCode)  
   shard := dict.getShard(index)  
   shard.mutex.Lock()  
   defer shard.mutex.Unlock()  
  
   if _, ok := shard.m[key]; ok {  
      return 0  
   }  
   shard.m[key] = val  
   dict.addCount()  
   return 1  
}  
  
// PutIfExists puts value if the key is exist and returns the number of inserted key-value
// 如果key存在，才更新
// 如果key不存在，则什么都不做
func (dict *ConcurrentDict) PutIfExists(key string, val interface{}) (result int) {  
   if dict == nil {  
      panic("dict is nil")  
   }  
   hashCode := fnv32(key)  
   index := dict.spread(hashCode)  
   shard := dict.getShard(index)  
   shard.mutex.Lock()  
   defer shard.mutex.Unlock()  
  
   if _, ok := shard.m[key]; ok {  
      shard.m[key] = val  
      return 1  
   }  
   return 0  
}  
  
// Remove removes the key and return the number of deleted key-value
// 删除key
func (dict *ConcurrentDict) Remove(key string) (result int) {  
   if dict == nil {  
      panic("dict is nil")  
   }  
   hashCode := fnv32(key)  
   index := dict.spread(hashCode)  
   shard := dict.getShard(index)  
   shard.mutex.Lock()  
   defer shard.mutex.Unlock()  
  
   if _, ok := shard.m[key]; ok {  
      delete(shard.m, key)  
      dict.decreaseCount()  
      return 1  
   }  
   return 0  
}  
// 原子性增加字典长度
func (dict *ConcurrentDict) addCount() int32 {  
   return atomic.AddInt32(&dict.count, 1)  
}  
// 原子性减少字典长度
func (dict *ConcurrentDict) decreaseCount() int32 {  
   return atomic.AddInt32(&dict.count, -1)  
}  
  
// ForEach traversal the dict// it may not visits new entry inserted during traversal  
// 传入函数类型Consumer，对字典中的每一个key都执行此函数
func (dict *ConcurrentDict) ForEach(consumer Consumer) {  
   if dict == nil {  
      panic("dict is nil")  
   }  
  
   for _, shard := range dict.table {  
      shard.mutex.RLock()  
      func() {  
         defer shard.mutex.RUnlock()  
         for key, value := range shard.m {  
            continues := consumer(key, value)  
            if !continues {  
               return  
            }  
         }  
      }()  
   }  
}  
  
// Keys returns all keys in dict
// 返回该字典所有的key
func (dict *ConcurrentDict) Keys() []string {  
   keys := make([]string, dict.Len())  
   i := 0  
   dict.ForEach(func(key string, val interface{}) bool {  
      if i < len(keys) {  
         keys[i] = key  
         i++  
      } else {  
         keys = append(keys, key)  
      }  
      return true  
   })  
   return keys  
}  
  
// RandomKey returns a key randomly
// 随即获取一个key
func (shard *shard) RandomKey() string {  
   if shard == nil {  
      panic("shard is nil")  
   }  
   shard.mutex.RLock()  
   defer shard.mutex.RUnlock()  
  
   for key := range shard.m {  
      return key  
   }  
   return ""  
}  
  
// RandomKeys randomly returns keys of the given number, may contain duplicated key
// 随即获取limit个key
func (dict *ConcurrentDict) RandomKeys(limit int) []string {  
   size := dict.Len()  
   if limit >= size {  
      return dict.Keys()  
   }  
   shardCount := len(dict.table)  
  
   result := make([]string, limit)  
   for i := 0; i < limit; {  
      shard := dict.getShard(uint32(rand.Intn(shardCount)))  
      if shard == nil {  
         continue  
      }  
      key := shard.RandomKey()  
      if key != "" {  
         result[i] = key  
         i++  
      }  
   }  
   return result  
}  
  
// RandomDistinctKeys randomly returns keys of the given number, won't contain duplicated key
// 随即获取limit个密钥，这些密钥是不重复的
func (dict *ConcurrentDict) RandomDistinctKeys(limit int) []string {  
   size := dict.Len()  
   if limit >= size {  
      return dict.Keys()  
   }  
  
   shardCount := len(dict.table)  
   result := make(map[string]bool)  
   for len(result) < limit {  
      shardIndex := uint32(rand.Intn(shardCount))  
      shard := dict.getShard(shardIndex)  
      if shard == nil {  
         continue  
      }  
      key := shard.RandomKey()  
      if key != "" {  
         result[key] = true  
      }  
   }  
   arr := make([]string, limit)  
   i := 0  
   for k := range result {  
      arr[i] = k  
      i++  
   }  
   return arr  
}  
  
// Clear removes all keys in dict
// 清除

func (dict *ConcurrentDict) Clear() {  
   *dict = *MakeConcurrent(dict.shardCount)  
}