golang中三种线程安全的MAP

一、map 是什么

map 是 Go 中用于存储 key-value 关系数据的数据结构，类似 C++ 中的 map，Python 中的 dict。Go 中 map 的使用很简单，但是对于初学者，经常会犯两个错误：没有初始化，并发读写。

1、未初始化的 map 都是 nil，直接赋值会报 panic。map 作为结构体成员的时候，很容易忘记对它的初始化。

2、并发读写是我们使用 map 中很常见的一个错误。多个协程并发读写同一个 key 的时候，会出现冲突，导致 panic。

Go 内置的 map 类型并没有对并发场景场景进行优化，但是并发场景又很常见，如何实现线程安全（并发安全）的 map就很重要了

二、三种线程安全的 map

1、加读写锁（RWMutex）

这是最容易想到的一种方式。常见的 map 的操作有增删改查和遍历，这里面查和遍历是读操作，增删改是写操作，因此对查和遍历需要加读锁，对增删改需要加写锁。

以 map[int]int 为例，借助 RWMutex，具体的实现方式如下:

type RWMap struct { // 一个读写锁保护的线程安全的map

sync.RWMutex // 读写锁保护下面的map字段

m map[int]int

}

// 新建一个RWMap

func NewRWMap(n int) *RWMap {

return &RWMap{

m: make(map[int]int, n),

}

func (m *RWMap) Get(k int) (int, bool) { //从map中读取一个值

m.RLock()

defer m.RUnlock()

v, existed := m.m[k] // 在锁的保护下从map中读取

return v, existed

}

func (m *RWMap) Set(k int, v int) { // 设置一个键值对

m.Lock() // 锁保护

defer m.Unlock()

m.m[k] = v

}

func (m *RWMap) Delete(k int) { //删除一个键

m.Lock() // 锁保护

defer m.Unlock()

delete(m.m, k)

}

func (m *RWMap) Len() int { // map的长度

m.RLock() // 锁保护

defer m.RUnlock()

return len(m.m)

}

func (m *RWMap) Each(f func(k, v int) bool) { // 遍历map

m.RLock() //遍历期间一直持有读锁

defer m.RUnlock()

for k, v := range m.m {

if !f(k, v) {

return

}

2、分片加锁

通过读写锁 RWMutex 实现的线程安全的 map，功能上已经完全满足了需要，但是面对高并发的场景，仅仅功能满足可不行，性能也得跟上。锁是性能下降的万恶之源之一。所以并发编程的原则就是尽可能减少锁的使用。当锁不得不用的时候，可以减小锁的粒度和持有的时间。

在第一种方法中，加锁的对象是整个 map，协程 A 对 map 中的 key 进行修改操作，会导致其它协程无法对其它 key 进行读写操作。一种解决思路是将这个 map 分成 n 块，每个块之间的读写操作都互不干扰，从而降低冲突的可能性。

Go 比较知名的分片 map 的实现是 orcaman/concurrent-map，它的定义如下：

var SHARD_COUNT = 32

// 分成SHARD_COUNT个分片的map

type ConcurrentMap []*ConcurrentMapShared

// 通过RWMutex保护的线程安全的分片，包含一个map

type ConcurrentMapShared struct {

items map[string]interface{}

sync.RWMutex // Read Write mutex, guards access to internal map.

}

// 创建并发map

func New() ConcurrentMap {

m := make(ConcurrentMap, SHARD_COUNT)

for i := 0; i < SHARD_COUNT; i++ {

m[i] = &ConcurrentMapShared{items: make(map[string]interface{})}

}

return m

}

// 根据key计算分片索引

func (m ConcurrentMap) GetShard(key string) *ConcurrentMapShared {

return m[uint(fnv32(key))%uint(SHARD_COUNT)]

}

ConcurrentMap 其实就是一个切片，切片的每个元素都是第一种方法中携带了读写锁的 map。

这里面 GetShard 方法就是用来计算每一个 key 应该分配到哪个分片上。

再来看一下 Set 和 Get 操作。

func (m ConcurrentMap) Set(key string, value interface{}) {

// 根据key计算出对应的分片

shard := m.GetShard(key)

shard.Lock() //对这个分片加锁，执行业务操作

shard.items[key] = value

shard.Unlock()

}

func (m ConcurrentMap) Get(key string) (interface{}, bool) {

// 根据key计算出对应的分片

shard := m.GetShard(key)

shard.RLock()

// 从这个分片读取key的值

val, ok := shard.items[key]

shard.RUnlock()

return val, ok

}

Get 和 Set 方法类似，都是根据 key 用 GetShard 计算出分片索引，找到对应的 map 块，执行读写操作。

3、sync 中的 map

分片加锁的思路是将大块的数据切分成小块的数据，从而减少冲突导致锁阻塞的可能性。如果在一些特殊的场景下，将读写数据分开，是不是能在进一步提升性能呢？

在内置的 sync 包中（Go 1.9+）也有一个线程安全的 map，通过将读写分离的方式实现了某些特定场景下的性能提升。

其实在生产环境中，sync.map 用的很少，官方文档推荐的两种使用场景是：

a) when the entry for a given key is only ever written once but read many times, as in caches that only grow.
b) when multiple goroutines read, write, and overwrite entries for disjoint sets of keys.

两种场景都比较苛刻，要么是一写多读，要么是各个协程操作的 key 集合没有交集（或者交集很少）。所以官方建议先对自己的场景做性能测评，如果确实能显著提高性能，再使用 sync.map。

sync.map 的整体思路就是用两个数据结构（只读的 read 和可写的 dirty）尽量将读写操作分开，来减少锁对性能的影响。

下面详细看下 sync.map 的定义和增删改查实现。

sync.map 数据结构定义

type Map struct {

mu Mutex

// 基本上你可以把它看成一个安全的只读的map

// 它包含的元素其实也是通过原子操作更新的，但是已删除的entry就需要加锁操作了

read atomic.Value // readOnly

// 包含需要加锁才能访问的元素

// 包括所有在read字段中但未被expunged（删除）的元素以及新加的元素

dirty map[interface{}]*entry

// 记录从read中读取miss的次数，一旦miss数和dirty长度一样了，就会把dirty提升为read，并把dirty置空

misses int

}

type readOnly struct {

m map[interface{}]*entry

amended bool // 当dirty中包含read没有的数据时为true，比如新增一条数据

}

// expunged是用来标识此项已经删掉的指针

// 当map中的一个项目被删除了，只是把它的值标记为expunged，以后才有机会真正删除此项

var expunged = unsafe.Pointer(new(interface{}))

// entry代表一个值

type entry struct {

p unsafe.Pointer // *interface{}

}

Map 的定义中，read 字段通过 atomic.Values 存储被高频读的 readOnly 类型的数据。dirty 存储

Store 方法

Store 方法用来设置一个键值对，或者更新一个键值对。

func (m *Map) Store(key, value interface{}) {

read, _ := m.read.Load().(readOnly)

// 如果read字段包含这个项，说明是更新，cas更新项目的值即可

if e, ok := read.m[key]; ok && e.tryStore(&value) {

return

}

// read中不存在，或者cas更新失败，就需要加锁访问dirty了

m.mu.Lock()

read, _ = m.read.Load().(readOnly)

if e, ok := read.m[key]; ok { // 双检查，看看read是否已经存在了

if e.unexpungeLocked() {

// 此项目先前已经被删除了，需要添加到 dirty 中

m.dirty[key] = e

}

e.storeLocked(&value) // 更新

} else if e, ok := m.dirty[key]; ok { // 如果dirty中有此项

e.storeLocked(&value) // 直接更新

} else { // 否则就是一个新的key

if !read.amended { //如果dirty为nil

// 需要创建dirty对象，并且标记read的amended为true,

// 说明有元素它不包含而dirty包含

m.dirtyLocked()

m.read.Store(readOnly{m: read.m, amended: true})

}

m.dirty[key] = newEntry(value) //将新值增加到dirty对象中

}

m.mu.Unlock()

}

// tryStore利用 cas 操作来更新value。

// 更新之前会判断这个键值对有没有被打上删除的标记

func (e *entry) tryStore(i *interface{}) bool {

for {

p := atomic.LoadPointer(&e.p)

if p == expunged {

return false

}

if atomic.CompareAndSwapPointer(&e.p, p, unsafe.Pointer(i)) {

return true

}

// 将值设置成 nil，表示没有被删除

func (e *entry) unexpungeLocked() (wasExpunged bool) {

return atomic.CompareAndSwapPointer(&e.p, expunged, nil)

}

// 通过复制 read 生成 dirty

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 {

if !e.tryExpungeLocked() {

m.dirty[k] = e

}

// 标记删除

func (e *entry) tryExpungeLocked() (isExpunged bool) {

p := atomic.LoadPointer(&e.p)

for p == nil {

if atomic.CompareAndSwapPointer(&e.p, nil, expunged) {

return true

}

p = atomic.LoadPointer(&e.p)

}

return p == expunged

}

第2-6行，通过 cas 进行键值对更新，更新成功直接返回。

第8-28行，通过互斥锁加锁来处理处理新增键值对和更新失败的场景（键值对被标记删除）。

第11行，再次检查 read 中是否已经存在要 Store 的 key（双检查是因为之前检查的时候没有加锁，中途可能有协程修改了 read）。

如果该键值对之前被标记删除，先将这个键值对写到 dirty 中，同时更新 read。

如果 dirty 中已经有这一项了，直接更新 read。

如果是一个新的 key。dirty 为空的情况下通过复制 read 创建 dirty，不为空的情况下直接更新 dirty。

Load 方法

Load 方法比较简单，先是从 read 中读数据，读不到，再通过互斥锁锁从 dirty 中读数据。

func (m *Map) Load(key interface{}) (value interface{}, ok bool) {

// 首先从read处理

read, _ := m.read.Load().(readOnly)

e, ok := read.m[key]

if !ok && read.amended { // 如果不存在并且dirty不为nil(有新的元素)

m.mu.Lock()

// 双检查，看看read中现在是否存在此key

read, _ = m.read.Load().(readOnly)

e, ok = read.m[key]

if !ok && read.amended {//依然不存在，并且dirty不为nil

e, ok = m.dirty[key]// 从dirty中读取

// 不管dirty中存不存在，miss数都加1

m.missLocked()

}

m.mu.Unlock()

}

if !ok {

return nil, false

}

return e.load() //返回读取的对象，e既可能是从read中获得的，也可能是从dirty中获得的

}

func (m *Map) missLocked() {

m.misses++ // misses计数加一

if m.misses < len(m.dirty) { // 如果没达到阈值(dirty字段的长度),返回

return

}

m.read.Store(readOnly{m: m.dirty}) //把dirty字段的内存提升为read字段

m.dirty = nil // 清空dirty

m.misses = 0 // misses数重置为0

}

这里需要注意的是，如果出现多次从 read 中读不到数据，得到 dirty 中读取的情况，就直接把 dirty 升级成 read，以提高 read 效率。

Delete 方法

下面是 Go1.13 中 Delete 的实现方式，如果 key 在 read 中，就将值置成 nil；如果在 dirty 中，直接删除 key。

func (m *Map) Delete(key interface{}) {

read, _ := m.read.Load().(readOnly)

e, ok := read.m[key]

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

}

补充说明一下，delete() 执行完之后，e.p 变成 nil，下次 Store 的时候，执行到 dirtyLocked() 这一步的时候，会被标记成 enpunged。因此在 read 中 nil 和 enpunged 都表示删除状态。

sync.map 总结

上面对源码粗略的梳理了一遍，最后在总结一下 sync.map 的实现思路：

读写分离。读（更新）相关的操作尽量通过不加锁的 read 实现，写（新增）相关的操作通过 dirty 加锁实现。
动态调整。新写入的 key 都只存在 dirty 中，如果 dirty 中的 key 被多次读取，dirty 就会上升成不需要加锁的 read。
延迟删除。Delete 只是把被删除的 key 标记成 nil，新增 key-value 的时候，标记成 enpunged；dirty 上升成 read 的时候，标记删除的 key 被批量移出 map。这样的好处是 dirty 变成 read 之前，这些 key 都会命中 read，而 read 不需要加锁，无论是读还是更新，性能都很高。

总结了 sync.map 的设计思路后，我们就能理解官方文档推荐的 sync.map 的两种应用场景了。