Google资深工程师深度讲解Go语言

2022-09-15 月初影视解说

Download：https://www.itwangzi.cn/2262.html

本文基于Go版本：1.17.8

go version go1.17.8 darwin/amd64

原生Map 并发场景

package main
import (
"fmt"
"sync"
"time"
)
func main() {
var wg sync.WaitGroup
m := make(map[int]struct{}, 0)
wg.Add(2)
go func() {
wg.Done()
for i := 0; i < 100000; i++ {
m[i] = struct{}{}
}
}()
go func() {
wg.Done()
for i := 0; i < 100000; i++ {
fmt.Println(m[i])
}
}()
wg.Wait()
time.Sleep(time.Second * 20)
}

多个协程同时写入出现fatal error: concurrent map read and map write的错误。如何解决该问题呢？那就需要給Map加上一把读写互斥锁sync.RWMutex。

测试代码如下:

package main
import (
"fmt"
"sync"
"time"
)
type Demo struct {
Data map[int]struct{}
Lock *sync.RWMutex
}
func (d Demo) Get(k int) {
d.Lock.RLock()
fmt.Println(d.Data[k])
d.Lock.RUnlock()
}
func (d Demo) Set(k int) {
d.Lock.Lock()
defer d.Lock.Unlock()
d.Data[k] = struct{}{}
}
func main() {
d := Demo{
Data: make(map[int]struct{}, 0),
Lock: new(sync.RWMutex),
}
for i := 0; i < 2000; i++ {
go d.Set(i)
go d.Get(i)
}
time.Sleep(time.Second * 20)
}

Sync.Map 概述

map是一种特殊的数据结构; 是一种元素对的无序集合,一个元素是key而对应的另一个元素是value,所以这个结构也称为关联数组或字典,是一种快速寻找值的理想结构;給定key可快速定位到value。

Go语言中Map(映射、字典)是一种内置的数据结构,它是一个无序的(key-value)对的集合。

*map初始化与存储的结构化*

Go语言原生Map并不是线程安全的,在解决并发读写Map的思路需要使用读写互斥锁(sync.RWMutex),这种方案简约直接,但是缺点也很明显,性能不会太高。而sync.Map在Go 1.9的版本中引入,它是一种并发安全的map,它的设计非常巧妙,充分利用原子操作(atomic)和互斥锁(mutex)的配合, 在使用sync.Map之后,对map的读写,不需要加入一大把锁,它通过空间换时间的方式,使用read(atomic.Value)和dirty(map[interface{}]*entry)两个原生map来进行读写分离,降低锁时间来提升效率。

核心思想

sync.Map核心原则就是尽量使用原子操作,最大程度上减少了锁的使用,从而接近lock free的效果。

使用了两个原生的map作为存储介质,分别为: 只读字典read(atomic.Value)和脏字典dirty(map[interface{}]*entry)。
只读字典使用atomic.Value来承载,保证原子性与高性能;脏字典则使用互斥锁sync.Mutex来进行保护,保证读写之间的互斥关系。
只读字典和脏字典中的键值对集合并不是实时同步的,它们在某些时间段内可能会有不同。
只读字典和脏字典其实在本质都是map[interface{}]*entry类型, entry就是Map的value容器。
entry是表示具体值的指针类型,也可以表示key已删除的状态。

通过这种设计,规避了原生map无法并发安全删除的问题,同时在变更某个键对应的value时也可以使用原子操作。

如何使用

package main
import (
"fmt"
"sync"
)
func main() {
var sm sync.Map
//1. 写入
sm.Store("张三", 18)
sm.Store("李四", 20)
//2. 读取
age, _ := sm.Load("张三")
fmt.Println(age.(int))
//3. 遍历
sm.Range(func(key, value interface{}) bool {
name := key.(string)
age := value.(int)
fmt.Println(name, age)
return true
})
//4. 删除
sm.Delete("李四")
age, ok := sm.Load("李四")
fmt.Println(age, ok)
//5. 读取或写入
sm.LoadOrStore("王二麻子", 100)
age, ok = sm.Load("王二麻子")
fmt.Println(age, ok)
}

sync.Map适用于读多写少的场景,

对于写多的场景,会导致read map缓存失效,需要加锁、导致冲突变多;
由于未命中read map次数过多,导致dirty map提升为read map,这是一个O(N)时间复杂度的操作,会进一步降低性能。

源码分析

先看一下sync.Map的数据结构:

type Map struct {
// 互斥锁保护read 与 dirty
mu Mutex
// read map的 k,v(dirty) 是不变的, 删除只是打标记,插入新key会加锁写到dirty map 中
// 因此对read map的读取无需加锁。
read atomic.Value
// dirty map 对dirty map的操作需要持有互斥锁
dirty map[interface{}]*entry
// 当Load操作 read map中未找到,就会尝试从dirty中进行加载(不管是否存在), misses+1
// 当misses 达到dirty map 长度时,dirty被提升为read,并且重新分配dirty。
misses int
}
type readOnly struct {
m map[interface{}]*entry
// 为true时代表 dirty map中包含m中没有的元素
amended bool
}
type entry struct {
p unsafe.Pointer // *interface{}
}

read是atomic.Value类型,可以并发的读,但是如果需要更新read,则需要加入互斥锁保护。
read中存储的entry指针,可能会被并发的CAS(比较并交换)更新。但是如果更新一个之前已被删除entry,则需要先将其状态从删除状态改为nil,再拷贝到dirty map中去,然后再执行更新。
dirty 是一个非线程安全的原始map。包含新写入的key,并且包含read中所有未被删除的key。这样可以快速的将dirty提升为read对外提供服务。如果dirty为nil,那么下一次写入时,会新建一个新的dirty,这个初始的dirty是read的一个拷贝,但除掉了其中已经被删除的key。
每当从read中读取失败,都会将 misses的计数值+1,dirty被提升为read`。

*entry.p的三种状态*

read和dirty 原生map都存储都包含entry,它是一个指针,指向value read和dirty 各自维护一套key,而它指向都同一个value,只要修改了这个entry,对read和dirty都是可见的。而这个指针的状态有三个状态:
- 当p == nil时,说明这个键值对已被删除,并且m.dirty==nil 或 m.dirty[k]指向该entry。
- 当p == expunged时,说明这条键值对已被,并且m.dirty !=nil且m.dirty中没有这个key。
- 其它情况,p指向一个正常值,表示实际interface{}的地址,并且被记录在m.read.m[key]中、如果这时m.dirty不为nil,那么它也被记录在m.dirty[key]中。两者实际上指向的是同一个值.
- 当删除key时,并不实际删除。一个entry可以通过原子地设置p为nil删除。如果之后创建m.dirty,nil又会被原子地设置expunged,且不会拷贝dirty中。
- 如果p不为expunged和entry相关联的这个value可以被原子地更新;p == expunged那么当它初次设置到m.entry之后才可以被更新。

sync.Map的数据结构

Load 读取

Load 函数具体实现方式

func (m *Map) Load(key interface{}) (value interface{}, ok bool) {
// 优先从read map中读取数据(无锁)
read, _ := m.read.Load().(readOnly)
e, ok := read.m[key]
// 如果不存在并且 read.amended 字段指明 dirty map中有read map中不存在的字段,则加锁尝试
// 从dirty map中加载
if !ok && read.amended {
// dirty map 不是线程安全的,所以需要加上互斥锁
m.mu.Lock()
// double check 避免在加锁的时候dirty map提升为read map
read, _ = m.read.Load().(readOnly)
e, ok = read.m[key]
// 任然没有在 read中找到这个key ,并且 amended 为true
if !ok && read.amended {
e, ok = m.dirty[key]
// 不管dirty中有没有找到,都增加misses 计数,该函数可能将dirty map提升为readmap
m.missLocked()
}
m.mu.Unlock()
}
if !ok {
return nil, false
}
return e.load()
}
// 从entry中原子地操作 load 实际interface{}
func (e *entry) load() (value interface{}, ok bool) {
p := atomic.LoadPointer(&e.p)
if p == nil || p == expunged {
return nil, false
}
return *(*interface{})(p), true
}
// 增加misses计数,并在必要的时候提升 dirty map
func (m *Map) missLocked() {
m.misses++
if m.misses < len(m.dirty) {
return
}
// dirty map 晋升
m.read.Store(readOnly{m: m.dirty})
m.dirty = nil
m.misses = 0
}

如果read 中没有这个key,且amended为false,说明dirty为空,那就直接返回空或false。
如果read中没有这个key,且amended为true,说明dirty中可能存在我们要找的key,先上互斥锁在尝试去dirty中查找,在此之前,仍然有一个double check的操作,若还是没有在read中找到,那么就从dirty中找,不管dirty中有没有找到,都需要"记录一笔",因为在dirty被提升为read之前,都会进入这条路径。

Store 存储

Store 函数实现

func (m *Map) Store(key, value interface{}) {
// 如果read map 中存在该key 则尝试直接更改(由于修改的是 entry
//内部的pointer, 因此 dirty map 也可见)
read, _ := m.read.Load().(readOnly)
if e, ok := read.m[key]; ok && e.tryStore(&value) {
return
}
m.mu.Lock()
read, _ = m.read.Load().(readOnly)
if e, ok := read.m[key]; ok {
if e.unexpungeLocked() {
//如果 read map 中存在该key , 当p == expunged,则说明m.dirty != nil
// 并且 m.dirty 中不存在该key 值此时:
// a. 将p的状态由expunged 更改为nil
// b. dirty map 插入key
// c. 更新 entry.p = value (read map 和 dirty map 指向同一个entry )
m.dirty[key] = e
}
//如果read map中存在该key，且 p != expunged，直接更新该entry
//(此时m.dirty==nil或m.dirty[key]==e)
e.storeLocked(&value)
} else if e, ok := m.dirty[key]; ok {
// 如果 read map 中不存在该key, 但是dirty map 中存在该key,
//直接写入更新 entry(read map 中仍然没有这个key)
e.storeLocked(&value)
} else {
if !read.amended {
// 如果 read map 和dirty map 中都不存在该key,则:
// a. 如果dirty map 为空,则需要创建 dirty map,并从read map 中拷贝未删除的元素
// b. 更新amended 为true,并标记dirty map中存在read map中没有的key
// c. 将k ,v 写入dirty map中, read map不做改变
m.dirtyLocked()
m.read.Store(readOnly{m: read.m, amended: true})
}
m.dirty[key] = newEntry(value)
}
m.mu.Unlock()
}

Delete 删除

Delete 函数实现

func (m *Map) Delete(key interface{}) {
m.LoadAndDelete(key)
}
func (m *Map) LoadAndDelete(key interface{}) (value interface{}, loaded bool) {
// 从read map 中查找, 如果存在,则设置为nil
read, _ := m.read.Load().(readOnly)
e, ok := read.m[key]
if !ok && read.amended {
m.mu.Lock()
// double check
read, _ = m.read.Load().(readOnly)
e, ok = read.m[key]
// 如果 read map 中不存在,但dirty map中存在, 则直接从dirty map 删除
if !ok && read.amended {
e, ok = m.dirty[key]
delete(m.dirty, key)
m.missLocked()
}
m.mu.Unlock()
}
if ok {
//将 entry.p 设置为nil
return e.delete()
}
return nil, false
}
func (e *entry) delete() (value interface{}, ok bool) {
for {
p := atomic.LoadPointer(&e.p)
if p == nil || p == expunged {
return nil, false
}
//CAS操作
if atomic.CompareAndSwAppointer(&e.p, p, nil) {
return *(*interface{})(p), true
}
}
}

总结

sync.Map是线程安全的,读取,插入,删除也都是保持着常数级O(1)的时间复杂度。
通过读写分离,降低锁时间来提高效率,适用于读多写少的场景。

测试性能代码

package _map
import (
"sync"
)
var (
myMap *MyMap
syncMap *sync.Map
)
type MyMap struct {
sync.RWMutex
m map[int]struct{}
}
func init() {
myMap = &MyMap{
m: make(map[int]struct{}, 0),
}
syncMap = new(sync.Map)
}
func mutexMapStore(k int) {
myMap.Lock()
myMap.m[k] = struct{}{}
myMap.Unlock()
}
func mutexMapLoad(k int) int {
myMap.RLock()
defer myMap.RUnlock()
if _, ok := myMap.m[k]; ok {
return 1
}
return 0
}
func mutexMapDelete(k int) {
myMap.Lock()
delete(myMap.m, k)
myMap.Unlock()
}
func syncMapStore(k, v int) {
syncMap.Store(k, v)
}
func syncMapLoad(k int) int {
if _, ok := syncMap.Load(k); ok {
return 1
}
return 0
}
func syncMapDelete(k int) {
syncMap.Delete(k)
}

Benchmark Test

package _map
import (
"math/rand"
"testing"
"time"
)
func BenchmarkMutexStoreParallel(b *testing.B) {
b.RunParallel(func(pb *testing.PB) {
r := rand.New(rand.NewSource(time.Now().Unix()))
for pb.Next() {
k := r.Intn(1000000)
mutexMapStore(k)
}
})
}
func BenchmarkMutexMapStoreParalell(b *testing.B) {
b.RunParallel(func(pb *testing.PB) {
r := rand.New(rand.NewSource(time.Now().Unix()))
for pb.Next() {
k := r.Intn(1000000)
mutexMapStore(k)
}
})
}
func BenchmarkSyncMapStoreParalell(b *testing.B) {
b.RunParallel(func(pb *testing.PB) {
r := rand.New(rand.NewSource(time.Now().Unix()))
for pb.Next() {
// The loop body is executed b.N times total across all goroutines.
k := r.Intn(100000000)
syncMapStore(k, k)
}
})
}
func BenchmarkMutexMapLoadParalell(b *testing.B) {
b.RunParallel(func(pb *testing.PB) {
r := rand.New(rand.NewSource(time.Now().Unix()))
for pb.Next() {
k := r.Intn(100000000)
mutexMapLoad(k)
}
})
}
func BenchmarkSyncMapLoadParalell(b *testing.B) {
b.RunParallel(func(pb *testing.PB) {
r := rand.New(rand.NewSource(time.Now().Unix()))
for pb.Next() {
k := r.Intn(100000000)
syncMapLoad(k)
}
})
}
func BenchmarkMutexMapDeleteParalell(b *testing.B) {
b.RunParallel(func(pb *testing.PB) {
r := rand.New(rand.NewSource(time.Now().Unix()))
for pb.Next() {
// The loop body is executed b.N times total across all goroutines.
k := r.Intn(100000000)
mutexMapDelete(k)
}
})
}
func BenchmarkSyncMapDeleteParalell(b *testing.B) {
b.RunParallel(func(pb *testing.PB) {
r := rand.New(rand.NewSource(time.Now().Unix()))
for pb.Next() {
k := r.Intn(100000000)
syncMapDelete(k)
}
})
}

测试结果

goos: darwin
goarch: amd64
pkg: lib/sync/map
cpu: Intel(R) Core(TM) i7-9750H CPU @ 2.60GHz
BenchmarkMutexStoreParallel-12 9638028 136.0 ns/op 2 B/op 0 allocs/op
BenchmarkMutexMapStoreParalell-12 9125176 148.9 ns/op 0 B/op 0 allocs/op
BenchmarkSyncMapStoreParalell-12 4548318 294.9 ns/op 44 B/op 3 allocs/op
BenchmarkMutexMapLoadParalell-12 20580038 51.57 ns/op 0 B/op 0 allocs/op
BenchmarkSyncMapLoadParalell-12 91573840 12.51 ns/op 0 B/op 0 allocs/op
BenchmarkMutexMapDeleteParalell-12 9240199 152.4 ns/op 0 B/op 0 allocs/op
BenchmarkSyncMapDeleteParalell-12 99910771 12.19 ns/op 0 B/op 0 allocs/op
PASS