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也是能获取到新增的数据。
这篇好文章是转载于:学新通技术网
- 版权申明: 本站部分内容来自互联网,仅供学习及演示用,请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系,请提供相关证据及您的身份证明,我们将在收到邮件后48小时内删除。
- 本站站名: 学新通技术网
- 本文地址: /boutique/detail/tanhfiffbb
-
photoshop保存的图片太大微信发不了怎么办
PHP中文网 06-15 -
Android 11 保存文件到外部存储,并分享文件
Luke 10-12 -
《学习通》视频自动暂停处理方法
HelloWorld317 07-05 -
word里面弄一个表格后上面的标题会跑到下面怎么办
PHP中文网 06-20 -
photoshop扩展功能面板显示灰色怎么办
PHP中文网 06-14 -
微信公众号没有声音提示怎么办
PHP中文网 03-31 -
excel下划线不显示怎么办
PHP中文网 06-23 -
excel打印预览压线压字怎么办
PHP中文网 06-22 -
怎样阻止微信小程序自动打开
PHP中文网 06-13 -
TikTok加速器哪个好免费的TK加速器推荐
TK小达人 10-01