go线程安全哈希表concurrent-map

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也是能获取到新增的数据。