github.com/orcaman/concurrent-map
这个哈希表是基于golang提供的map来实现线程安全的哈希表。map的key只能是string
下面给一个多线程操作的例子，该map的value为string
package main
import (
"fmt"
"github.com/orcaman/concurrent-map/v2"
"sync"
)
func main() {
var wg sync.WaitGroup
// Create a new map.
m := cmap.New[string]()
wg.Add(1)
// Sets item within map, sets "bar" under key "foo"
go func() {
defer wg.Done()
m.Set("foo", "bar1111")
m.Set("foo1", "bar11112")
m.Set("test1", "val1")
}()
wg.Add(1)
go func() {
defer wg.Done()
fmt.Println("====================")
m.Set("111", "2222")
for item := range m.IterBuffered() {
fmt.Println("==== kv :", item)
}
fmt.Println("====================")
} ()
// Retrieve item from map.
go func () {
bar, ok := m.Get("foo")
fmt.Println("bar : ", bar, ", ok :", ok)
// Removes item under key "foo"
m.Remove("foo")
}()
wg.Add(1)
go func() {
defer wg.Done()
m.Set("3333", "4444")
fmt.Println("-----------------")
for item := range m.IterBuffered() {
fmt.Println("---- kv :", item)
}
fmt.Println("-----------------")
}()
wg.Wait()
}
运行结果：
====================
bar : bar1111 , ok : true
-----------------
==== kv : {3333 4444}
---- kv : {3333 4444}
---- kv : {111 2222}
==== kv : {111 2222}
---- kv : {foo1 bar11112}
---- kv : {test1 val1}
==== kv : {foo1 bar11112}
-----------------
==== kv : {test1 val1}
====================
Process finished with the exit code 0
-----------------
bar : bar1111 , ok : true
====================
---- kv : {3333 4444}
---- kv : {111 2222}
---- kv : {foo1 bar11112}
---- kv : {test1 val1}
-----------------
==== kv : {3333 4444}
==== kv : {111 2222}
==== kv : {foo1 bar11112}
==== kv : {test1 val1}
====================
Process finished with the exit code 0
运行结果比较随机。
下面是对该哈希表的源码分析：
数据结构：
/*
*多少个子哈希表，这个可以修改，如果想减少冲突，可以参考之前的博客中的github.com/tidwall/shardmap哈希表的计算方式，即根据实际的物理CPU核心数来计算
*/
var SHARD_COUNT = 32
// A "thread" safe map of type string:Anything.
// To avoid lock bottlenecks this map is dived to several (SHARD_COUNT) map shards.
type ConcurrentMap[V any] []*ConcurrentMapShared[V]//对外暴露的哈希结构，是ConcurrentMapShared的指针数组，其中V any的用法类似于c++的模板，传入的结构就是value的结构
// A "thread" safe string to anything map.
type ConcurrentMapShared[V any] struct {
items map[string]V//key为string就是在这里决定的，这里使用的是golang自带的哈希表，减少重复造轮子
sync.RWMutex // Read Write mutex, guards access to internal map.//每个哈希表一个读写锁
}
创建哈希表cmap.New：
// Creates a new concurrent map.
func New[V any]() ConcurrentMap[V] {
m := make(ConcurrentMap[V], SHARD_COUNT)//创建哈希表结构
for i := 0; i < SHARD_COUNT; i++ {//循环创建子哈希表
m[i] = &ConcurrentMapShared[V]{items: make(map[string]V)}
}
return m
}
添加元素Set：
// Sets the given value under the specified key.
func (m ConcurrentMap[V]) Set(key string, value V) {
// Get map shard.
shard := m.GetShard(key)//根据传入的key，计算此key对应在哪个子哈希表中，
shard.Lock()//使用对应的锁进行加写锁
shard.items[key] = value//赋值（如果已存在，替换）
shard.Unlock()//解锁
}
哈希函数进行散列（fnv哈希算法）：
// GetShard returns shard under given key
func (m ConcurrentMap[V]) GetShard(key string) *ConcurrentMapShared[V] {
return m[uint(fnv32(key))%uint(SHARD_COUNT)]
}
func fnv32(key string) uint32 {
hash := uint32(2166136261)
const prime32 = uint32(16777619)
keyLength := len(key)
for i := 0; i < keyLength; i++ {
hash *= prime32
hash ^= uint32(key[i])
}
return hash
}
Get函数：
// Get retrieves an element from map under given key.
func (m ConcurrentMap[V]) Get(key string) (V, bool) {//基本逻辑和Set一样，不过因为是读操作，所以加读锁
// Get shard
shard := m.GetShard(key)
shard.RLock()
// Get item from shard.
val, ok := shard.items[key]
shard.RUnlock()
return val, ok
}
代码中有个遍历哈希表操作
for item := range m.IterBuffered() {......}
具体代码实现如下：
// Used by the Iter & IterBuffered functions to wrap two variables together over a channel,
type Tuple[V any] struct {//键值对结构
Key string
Val V
}
......
// IterBuffered returns a buffered iterator which could be used in a for range loop.
func (m ConcurrentMap[V]) IterBuffered() <-chan Tuple[V] {
chans := snapshot(m)//获取快照数据，其实这里就是将所有子哈希表当前数据的键值对信息，后面增加或者删除了元素不影响，
total := 0
for _, c := range chans {//遍历chan数组个数
total += cap(c)//每个元素都是一个一维数组
}
ch := make(chan Tuple[V], total)//根据计算键值对的个数分配数组
go fanIn(chans, ch)//并发去将chans的内容写入ch中，这里为什么不直接用chans，因为chans是二维数组，这里的操作是将二维数组转换成一维数组
return ch
}
我们看一下快照函数的实现：
// Returns a array of channels that contains elements in each shard,
// which likely takes a snapshot of `m`.
// It returns once the size of each buffered channel is determined,
// before all the channels are populated using goroutines.
func snapshot[V any](m ConcurrentMap[V]) (chans []chan Tuple[V]) {
//When you access map items before initializing.
if len(m) == 0 {
panic(`cmap.ConcurrentMap is not initialized. Should run New() before usage.`)
}
chans = make([]chan Tuple[V], SHARD_COUNT)//根据实际的子哈希表个数分配chan 数组
wg := sync.WaitGroup{}
wg.Add(SHARD_COUNT)//设置等待goroutine个数，因为我们需要对每个子表进行遍历操作，所以还是SHARD_COUNT
// Foreach shard.
for index, shard := range m {//遍历子表
go func(index int, shard *ConcurrentMapShared[V]) {//启动goroutine进行操作，提升性能
// Foreach key, value pair.
shard.RLock()//使用对应的锁加读锁
chans[index] = make(chan Tuple[V], len(shard.items))
wg.Done()
for key, val := range shard.items {//遍历子表，将对应键值对复制到对应的chans的子数组中
chans[index] <- Tuple[V]{key, val}
}
shard.RUnlock()
close(chans[index])
}(index, shard)
}
wg.Wait()//这里会等待所有子表都遍历完，才会返回，所以如果散列的不够均匀，等待时间会长
return chans
}
我们来看将二维数组聚合成一维数组的过程fanIn：
// fanIn reads elements from channels `chans` into channel `out`
func fanIn[V any](chans []chan Tuple[V], out chan Tuple[V]) {
wg := sync.WaitGroup{}
wg.Add(len(chans))
for _, ch := range chans {//遍历每一个chan，一个子表一个chan
go func(ch chan Tuple[V]) {
for t := range ch {//遍历子表的键值对
out <- t//全部赋值到out的一维数组中
}
wg.Done()
}(ch)
}
wg.Wait()
close(out)
}
其实之前还有一个老的Iter版本，注释说性能比较差。我们看看它的实现：
// Iter returns an iterator which could be used in a for range loop.
//
// Deprecated: using IterBuffered() will get a better performence
func (m ConcurrentMap[V]) Iter() <-chan Tuple[V] {
chans := snapshot(m)
ch := make(chan Tuple[V])
go fanIn(chans, ch)
return ch
}
唯一的区别就是没有指定ch的的大小。chan内部实现是一个环形队列，如果队列不够大，会等待处理，性能差。
例子中还用到了Remove接口，其实和set大同小异，底层实现使用的是系统自带的delete函数。其它的函数后面用到了再来追加，基本逻辑都是一样。
备注：
里面有两个函数比较好一个是MSet，这个用于将一个map数据导入到这个并发安全的map中。
第二个是Upsert，这个函数用于修改或添加键值对，提供了UpsertCb回调函数，也就是说，可以提供自己的逻辑来处理数据，比如在历史数据上面进行修改，比较方便。
下面是一部分测试代码：
......//这是上面的测试用例代码，我们只需要在main函数中调用下UpdateTest()就可以了。
func UpdateTest() {
m := cmap.New[int64]()
m.Set("test", 1)
m.Upsert("test", 3, UpsertCb)
v, ok :=m.Get("test")
if ok {
fmt.Println(v)
}
}
func UpsertCb (exist bool, valueInMap int64, newValue int64) int64{
if exist {
return valueInMap + newValue
} else {
return newValue
}
}
运行结果如下：
====================
bar : bar1111 , ok : true
-----------------
==== kv : {foo1 bar11112}
---- kv : {foo1 bar11112}
==== kv : {3333 4444}
---- kv : {3333 4444}
==== kv : {111 2222}
==== kv : {test1 val1}
====================
---- kv : {111 2222}
---- kv : {test1 val1}
-----------------
4
Process finished with the exit code 0
运行结果为4，我们的测试代码就是将新旧值相互累加后返回，如果存在旧值的话。这里就能看到它比sync.Map好的地方，这样就能进行临界控制来修改旧值。
同理RemoveCb也是一样的道理。这里就不一一分析了。
这里还有个地方需要注意就是Keys函数，它在刚进入函数的时候就获取该哈希表的总的个数，然后分配了一个一维数组chan，个数刚好就是此时哈希表的总个数，然后循环遍历哈希表将所有的key，最后转换成一个一维数组。需要考虑一个问题，如果此时往该哈希表中添加一个元素会不会出问题？程序出来这么久，应该不会问题，整个哈希表遍历完，就调用close channel，就算数据少了也会直接返回，不会有问题，如果多了，channel也是能获取到新增的数据。
|