「Golang」并发场景下的Map使用方式及避坑指南

昨天挖了个并发读写map的坑，今天给填上

非线程安全的map

map
是一个在开发过程中常用的内建类型，也是各位很熟悉的一个数据结构，他可以很方便的让我们做一些关于针对某些key-value结构的CRUD操作，但是在官方设计中，内建类型的map并不是一个可以在并发场景下进行并发读写的线程安全类型，此时我们就需要考虑对其进行一些线程安全的改造，首先我们要了解一下为什么```map``是一个非线程安全的类型，首先看下列代码：

func main() {
 m := make(map[string]int, 2)
 m["dd"] = 22
 go func() {
  for {
   m["ff"] = 1
  }
 }()
 go func() {
  for {
   _ = m["dd"]
  }
 }()
 time.Sleep(1 * time.Hour)
}

// out:
// fatal error: concurrent map read and map write

上面的代码在运行期抛出了：fatal error: concurrent map read and map write
异常，代表在读写过程中出现了同时读写的问题，那么为什么会出现这个问题呢？在map
底层源代码的针对读和写的函数中有下列两段代码：

// runtime.mapassign 即在map进行写入操作时调用的函数
func mapassign(t *maptype, h *hmap, key unsafe.Pointer) unsafe.Pointer {
 // ..............

 // 在写入时会对h.flags字段进行正在写入标记以防止并发读写
 h.flags ^= hashWriting
 // ..........
 
 // 检查是否有并发写的问题
 if h.flags&hashWriting == 0 {
  throw("concurrent map writes")
 }
 // 写入完毕回复标记字段
 h.flags &^= hashWriting
 //.........
}
// runtime.mapaccess1 在 value:=m[key]时调用的函数
func mapaccess1(t *maptype, h *hmap, key unsafe.Pointer) unsafe.Pointer {
 // .............
 // 检测标记字段 是否不等于0 如果不等于0则代表发生并发读写
 if h.flags&hashWriting != 0 {
  throw("concurrent map read and map write")
 }
 // .............
}
// runtime.mapaccess2 在 value,ok:=m[key]时调用的函数
func mapaccess2(t *maptype, h *hmap, key unsafe.Pointer) (unsafe.Pointer, bool) {
 // .............
 // 检测标记字段 是否不等于0 如果不等于0则代表发生并发读写
 if h.flags&hashWriting != 0 {
  throw("concurrent map read and map write")
 }
 // .............
} 

上述源代码做了很大一部分的省略，仅留下了针对并发读写标记的设置、取消、以及检查的语句，在此能看出map
的底层根据并发场景做了非线程安全的处理。

如何使得map
变成线程安全

MutexMap
的实现

上面讲了map
在并发环境下非线程安全的例子以及原因，那么目前我们如何解决这个问题呢？从之前作者的文章：「Golang」 sync.Mutex源码讲解 中可以看到使用互斥锁可以使得在并发环境下某些对象只能在某些事件内只有一个goroutine持有对其进行CRUD操作，那么我们可以使用互斥锁对map
对其进行封装，例子代码如下：

type MutexMap struct {
 lock sync.Mutex
 m    map[string]int
}

func (receiver *MutexMap) Get(key string) (int, bool) {
 receiver.lock.Lock()
 value, ok := receiver.m[key]
 receiver.lock.Unlock()
 return value, ok
}

func (receiver *MutexMap) Set(key string, value int) {
 receiver.lock.Lock()
 receiver.m[key] = value
 receiver.lock.Unlock()
}
func (receiver *MutexMap) Del(key string) {
 receiver.lock.Lock()
 delete(receiver.m, key)
 receiver.lock.Unlock()
}

上述代码中作者使用了一个结构体对sync.Mutex
和map
进行了封装，并且对外提供了一系列的方法以供使用，从上述方法中可以看到，在Get(),Set(),Del()
方法中进行map
操作前均对其进行了互斥锁加锁操作，这样使得该结构体对象可以在并发环境下保证只有一个goroutine持有对其map
进行CRUD操作，这样也就消除了并发读写问题的影响。

MutexMap
的优化，RWMutexMap
的诞生

在上一节中，我们进行了MutexMap
的实现，但是有一个问题也就产生了，由于互斥锁的特性，虽然可以提供线程安全的 map
，但是在大量并发读写的情况下，锁的竞争会非常激烈。尤其在读的并发非常大的时候，互斥锁会严重影响读取的性能，因为在通过map
底层源码发现，并发的读并不会产生panic
只有并发读写时，才会发生，因此，互斥锁会导致整体读的效率下降很多，此时我们就应该使用读写锁来进行优化，读写锁相关解析请阅读: 「Golang」sync.RWMutex源码讲解，该文章中指出：

读写锁就是一个可以并发读但是不可以并发写的锁，由于互斥锁的特性，会导致将所有goroutine（下称协程）串行化，从而影响整体程序运行的性能，如果写的数量大于读的数量时，性能损耗暂且可以忽略不计（建议不要忽略），但是当读的数量大于写的数量时，互斥锁对性能的影响是很大的，所以此刻我们需要采用读写锁来进行读操作和写操作的分开。

现在我们就对MutexMap
进行读写锁的优化，优化后的代码如下：

type RWMutexMap struct {
 lock sync.RWMutex
 m    map[string]int
}

func (receiver *RWMutexMap) Get(key string) (int, bool) {
 receiver.lock.RLock()
 value, ok := receiver.m[key]
 receiver.lock.RUnlock()
 return value, ok
}

func (receiver *RWMutexMap) Set(key string, value int) {
 receiver.lock.Lock()
 receiver.m[key] = value
 receiver.lock.Unlock()
}
func (receiver *RWMutexMap) Del(key string) {
 receiver.lock.Lock()
 delete(receiver.m, key)
 receiver.lock.Unlock()
}

上述代码中，将lock sync.Mutex
字段替换成为了lock sync.RWMutex
字段，并且将Get()
方法转而使用读锁进行加锁，此时，当前的 RWMutexMap
结构体就针对读多写少的并发情况有了很好的性能提升，

标准库的sync.Map

标准库中提供了一种官方实现的线程安全的map
结构，即sync.Map
，该包虽然是线程安全的，但是使用场景很少，这个 sync.Map
并不是用来替换内建的 map
类型的，它只能被应用在一些特殊的场景里。官方文档指出可以用在以下场景：

1.只会增长的缓存系统中，一个 key 只写入一次而被读很多次，即读多写少 2.多个 goroutine 为不相交的键集读、写和重写键值对，即多个goroutineCRUD操作不同的key-value

但是，最好在使用的时候根据场景首先进行性能测试后，再根据性能测试结果选用sync.Map
或者RWMutexMap
。sync.Map
的实现，其针对RWMutexMap
做了下面几个优化。

1. 空间换时间。通过冗余的两个数据结构（只读的 read 字段、可写的 dirty），来减少加锁对性能的影响。对只读字段（read）的操作不需要加锁。
2. 优先从 read 字段读取、更新、删除，因为对 read 字段的读取不需要锁。
3. 动态调整。miss次数多了之后，将 dirty 数据提升为 read，避免总是从 dirty 中加锁读取。
4. double-checking。加锁之后先还要再检查 read 字段，确定真的不存在才操作 dirty 字段。
5. 延迟删除。删除一个键值只是打标记，只有在提升 dirty字段为 read 字段的时候才清理删除的数据。

既然sync.Map
这么好，那么就来测试一下MutexMap
，RWMutexMap
，sync.Map
的性能差距吧，下面我针对这两个结构体进行性能测试看看其在大量只读，只写，读写的情况下的性能如何（代码引用于goalng1.14不同场景下sync.Map、Mutex、RWMutex锁性能测试对比）：

package main

import (
 "fmt"
 "sync"
 "testing"
 "time"
)

var (
 num  = 1000 * 10
 gnum = 1000
)

func Test_main(t *testing.T) {
 count := 10000
 div := int(50) //抽样写比例 1/5
 fmt.Println("only read")
 testRwmutexReadOnly(count)
 testMutexReadOnly(count)
 //test sync.map
 testSyncMapReadOnly(count)

 fmt.Println("write and read")
 testRwmutexWriteRead(count, div)
 testMutexWriteRead(count, div)
 testSyncMapWriteRead(count, div)

 fmt.Println("write only")
 testRwmutexWriteOnly(count)
 testMutexWriteOnly(count)
 testSyncMapWriteOnly(count)

}

func testRwmutexReadOnly(count int) {
 var w = &sync.WaitGroup{}
 var rwmutexTmp = newRwmutex(count)
 w.Add(gnum)
 t1 := time.Now()
 for i := 0; i < gnum; i++ {
  go func() {
   defer w.Done()
   for in := 0; in < num; in++ {
    rwmutexTmp.get(in)
   }
  }()
 }
 w.Wait()
 fmt.Println("testRwmutexReadOnly cost:", time.Now().Sub(t1).String())
}

func testRwmutexWriteOnly(count int) {
 var w = &sync.WaitGroup{}
 var rwmutexTmp = newRwmutex(count)
 w.Add(gnum)
 t1 := time.Now()
 for i := 0; i < gnum; i++ {
  go func() {
   defer w.Done()
   for in := 0; in < num; in++ {
    rwmutexTmp.set(in, in)
   }
  }()
 }
 w.Wait()
 fmt.Println("testRwmutexWriteOnly cost:", time.Now().Sub(t1).String())
}

func testRwmutexWriteRead(count, div int) {
 var w = &sync.WaitGroup{}
 var rwmutexTmp = newRwmutex(count)
 w.Add(gnum)
 t1 := time.Now()
 for i := 0; i < gnum; i++ {
  if i%div != 0 {
   go func() {
    defer w.Done()
    for in := 0; in < num; in++ {
     rwmutexTmp.get(in)
    }
   }()
  } else {
   go func() {
    defer w.Done()
    for in := 0; in < num; in++ {
     rwmutexTmp.set(in, in)
    }
   }()
  }
 }
 w.Wait()
 fmt.Println("testRwmutexWriteRead cost:", time.Now().Sub(t1).String())
}

func testMutexReadOnly(count int) {
 var w = &sync.WaitGroup{}
 var mutexTmp = newMutex(count)
 w.Add(gnum)

 t1 := time.Now()
 for i := 0; i < gnum; i++ {
  go func() {
   defer w.Done()
   for in := 0; in < num; in++ {
    mutexTmp.get(in)
   }
  }()
 }
 w.Wait()
 fmt.Println("testMutexReadOnly cost:", time.Now().Sub(t1).String())
}

func testMutexWriteOnly(count int) {
 var w = &sync.WaitGroup{}
 var mutexTmp = newMutex(count)
 w.Add(gnum)

 t1 := time.Now()
 for i := 0; i < gnum; i++ {
  go func() {
   defer w.Done()
   for in := 0; in < num; in++ {
    mutexTmp.set(in, in)
   }
  }()
 }
 w.Wait()
 fmt.Println("testMutexWriteOnly cost:", time.Now().Sub(t1).String())
}

func testMutexWriteRead(count, div int) {
 var w = &sync.WaitGroup{}
 var mutexTmp = newMutex(count)
 w.Add(gnum)
 t1 := time.Now()
 for i := 0; i < gnum; i++ {
  if i%div != 0 {
   go func() {
    defer w.Done()
    for in := 0; in < num; in++ {
     mutexTmp.get(in)
    }
   }()
  } else {
   go func() {
    defer w.Done()
    for in := 0; in < num; in++ {
     mutexTmp.set(in, in)
    }
   }()
  }

 }
 w.Wait()
 fmt.Println("testMutexWriteRead cost:", time.Now().Sub(t1).String())
}

func testSyncMapReadOnly(count int) {
 var w = &sync.WaitGroup{}
 var mutexTmp = newSyncMap(count)
 w.Add(gnum)

 t1 := time.Now()
 for i := 0; i < gnum; i++ {
  go func() {
   defer w.Done()
   for in := 0; in < num; in++ {
    mutexTmp.Load(in)
   }
  }()
 }
 w.Wait()
 fmt.Println("testSyncMapReadOnly cost:", time.Now().Sub(t1).String())
}

func testSyncMapWriteOnly(count int) {
 var w = &sync.WaitGroup{}
 var mutexTmp = newSyncMap(count)
 w.Add(gnum)

 t1 := time.Now()
 for i := 0; i < gnum; i++ {
  go func() {
   defer w.Done()
   for in := 0; in < num; in++ {
    mutexTmp.Store(in, in)
   }
  }()
 }
 w.Wait()
 fmt.Println("testSyncMapWriteOnly cost:", time.Now().Sub(t1).String())
}

func testSyncMapWriteRead(count, div int) {
 var w = &sync.WaitGroup{}
 var mutexTmp = newSyncMap(count)
 w.Add(gnum)
 t1 := time.Now()
 for i := 0; i < gnum; i++ {
  if i%div != 0 {
   go func() {
    defer w.Done()
    for in := 0; in < num; in++ {
     mutexTmp.Load(in)
    }
   }()
  } else {
   go func() {
    defer w.Done()
    for in := 0; in < num; in++ {
     mutexTmp.Store(in, in)
    }
   }()
  }

 }
 w.Wait()
 fmt.Println("testSyncMapWriteRead cost:", time.Now().Sub(t1).String())
}

func newRwmutex(count int) *rwmutex {
 var t = &rwmutex{}
 t.mu = &sync.RWMutex{}
 t.ipmap = make(map[int]int, count)

 for i := 0; i < count; i++ {
  t.ipmap[i] = 0
 }
 return t
}

type rwmutex struct {
 mu    *sync.RWMutex
 ipmap map[int]int
}

func (t *rwmutex) get(i int) int {
 t.mu.RLock()
 defer t.mu.RUnlock()

 return t.ipmap[i]
}

func (t *rwmutex) set(k, v int) {
 t.mu.Lock()
 defer t.mu.Unlock()

 t.ipmap[k] = v
}

func newMutex(count int) *mutex {
 var t = &mutex{}
 t.mu = &sync.Mutex{}
 t.ipmap = make(map[int]int, count)

 for i := 0; i < count; i++ {
  t.ipmap[i] = 0
 }
 return t
}

func newSyncMap(count int) *sync.Map {
 var t = &sync.Map{}

 for i := 0; i < count; i++ {
  t.Store(i, 0)
 }
 return t
}

type mutex struct {
 mu    *sync.Mutex
 ipmap map[int]int
}

func (t *mutex) get(i int) int {
 t.mu.Lock()
 defer t.mu.Unlock()

 return t.ipmap[i]
}

func (t *mutex) set(k, v int) {
 t.mu.Lock()
 defer t.mu.Unlock()

 k = k % 100
 t.ipmap[k] = v
}
// only read
// testRwmutexReadOnly cost: 346.72604ms
// testMutexReadOnly cost: 1.56353484s
// testSyncMapReadOnly cost: 94.436269ms
// write and read
// testRwmutexWriteRead cost: 511.433544ms
// testMutexWriteRead cost: 1.439301381s
// testSyncMapWriteRead cost: 106.220371ms
// write only
// testRwmutexWriteOnly cost: 2.579502195s
// testMutexWriteOnly cost: 1.922190051s
// testSyncMapWriteOnly cost: 4.177583278s

结论：

只读场景：sync.map > rwmutex >> mutex 读写场景（边读边写）：rwmutex > mutex >> sync.map 读写场景（读80% 写20%）：sync.map > rwmutex > mutex 读写场景（读98% 写2%）：sync.map > rwmutex >> mutex 只写场景：sync.map >> mutex > rwmutex

使用环境不负责任推荐

关于MutexMap
，RWMutexMap
，sync.Map
的选择问题，其实不一定只能看上述的测试结论，还需要根据应用场合以及实际情况来使用，根据我自己的情况来看的话，有以下几个方面可以考虑，只是我个人见解。

1. 由于sync.Map
   的通用性的问题，所以返回的值均为interface{}
   类型，这样在我们进行一些操作的时候就会需要不断的去断言他，导致麻烦，此时如果性能要求不是特别高的情况下可以使用RWMutexMap
   用作专用性替换，这样可以免去了断言的一些性能损耗和麻烦。
2. 之前看过一篇文章，RWMutexMap
   在并发量异常庞大的时候性能下降会很快，比如1000万，所以如果高并发的时候尽可能采用sync.Map
   
3. 当读写均多的时候sync.Map
   就显得力不从心，因为其内部实现原因需要针对只读map
   进行适时更换，如果一旦写入过多，就会导致更换的过于频繁，但是RWMutexMap
   就不会产生这个问题，除扩容和写锁的性能下降外，基本上没有什么其他损耗。
4. 如果map
   遍历情况过多请使用RWMutexMap
   ，因为sync.Map
   的时间复杂度最低为O(n)，所以此时sync.Map
   的遍历性能就很差了。