前言

Go 语言原生 map 并不是线程安全的，要对它进行并发读写操作时，一般有两种选择：

1. 原生map搭配Mutex或RWMutex
2. 使用sync.Map

和原生map搭配Mutex或RWMutex相比，sync.Map在以下场景更有优势：

• 读多写少

• 修改，删除已存在key对应的value较多

本文将介绍sync.map的整体结构，及查，增，删，改，遍历的实现原理，以及为啥要设置expunge这个特殊值

原理

流程

sync.map的增删改查的流程大体类似，基于只读结构read，和可写结构dirty

先看key在只读结构read中是否存在，如果存在直接进行操作。否则加锁去dirty结构中检查

结构

sync.map的数据结构比较简单，涉及3个结构体：

type Map struct {
   // 锁，用于保护dirty的访问
   mu Mutex
   // 只读的map，实际存储readOnly结构体 
   read atomic.Value 
   // 可写的map
   dirty map[any]*entry
   // 从read中查询失败的次数
   misses int
}

type readOnly struct {
   m       map[any]*entry
   // 为true时，代表dirty中存在read中没有的键值对
   amended bool 
}

type entry struct {
   p unsafe.Pointer 
}

• entry.p

  • 一般存储某个key对于的value值
  • 同时也有两个特殊的取值：nil，expunged，的Delete操作有关，后面详细介绍

• read和dirty中，相同key底层引用了同一个entry，因此对read中的entry修改，也会影响到dirty

下面分析sync.Map关键方法的代码细节

Load

func (m *Map) Load(key any) (value any, ok bool) {
   read, _ := m.read.Load().(readOnly)
   e, ok := read.m[key]
   // 如果key在read中不存在，且dirty数据比read多，则去dirty中找
   if !ok && read.amended {
      m.mu.Lock()
      // 双重检查，再去read中找一次
      read, _ = m.read.Load().(readOnly)
      e, ok = read.m[key]
      // 如果read中还是没有，就去dirty中找
      if !ok && read.amended {
         e, ok = m.dirty[key]
         m.missLocked()
      }
      m.mu.Unlock()
   }
   if !ok {
      return nil, false
   }
   // 如果read中有该key，返回该value。从代码可读性角度来说，其实这一步可以在第4行直接返回
   return e.load()
}

Load整体流程为：

• 先从read中尝试获取，如果存在直接返回

• 否则加锁，再次从read中获取一次

  • 这里是经典的双重检查做法，在sync.Map中大量使用。因为在从read读和加锁期间，可能有其他线程对map进行了操作，使read中有该键值对了

• 如果还是没有，就从dirty中获取

• 在missLocked方法中，不管是否获取成功都对m.misses++，如果达到阈值，就将dirty提升为read

  • 提升dirty的目的：将全量的数据提升到read中，使得后续的操作能在read中完成，无需加锁

其中涉及到的子方法：

func (m *Map) missLocked() {
   // read中没有的次数++
   m.misses++
   // 若misses不够多，直接返回
   if m.misses < len(m.dirty) {
      return
   }
   // 否则重建read，做法为将dirty赋值给read，并将dirty,misses置空
   m.read.Store(readOnly{m: m.dirty})
   m.dirty = nil
   m.misses = 0
}

func (e *entry) load() (value any, ok bool) {
   p := atomic.LoadPointer(&e.p)
   if p == nil || p == expunged {
      return nil, false
}
   return *(*any)(p), true
}

• entry.load()即检查entry.p是否为nil或expunged，如果是说明键值对已经被删除，返回空

Store

func (m *Map) Store(key, value any) {
   read, _ := m.read.Load().(readOnly)
   // 如果read中存在该键值对，cas更新其value
   if e, ok := read.m[key]; ok && e.tryStore(&value) {
      return
   }
   // 接下来就是当前时刻read中没有该键值对的逻辑
   m.mu.Lock()
   read, _ = m.read.Load().(readOnly)
   // 如果加锁后发现read中有了
   if e, ok := read.m[key]; ok {
      // 如果e是被删除状态，将其更新为nil
      if e.unexpungeLocked() {
         // 并且给dirty中增加该键值对，因为此时dirty中没有
         m.dirty[key] = e
      }
      // 更新value
      e.storeLocked(&value)
   // read没有，但dirty有，更新dirty中该entry的值   
   } else  if e, ok := m.dirty[key]; ok {
      e.storeLocked(&value)
   // dirty，read都没有   
   } else {
      // 如果刚刚把dirty提升到read
      if !read.amended {
         // 将read浅拷贝到dirty中
         m.dirtyLocked()
         // 修改read.amended为true
         m.read.Store(readOnly{m: read.m, amended: true})
      }
      // 只将键值对加到dirty中
      m.dirty[key] = newEntry(value)
   }
   m.mu.Unlock()
}

Store整体流程为：

• 如果read中存在该键值对，CAS更新其value

• 若不存在，加锁，执行后面的逻辑：

  • 如果加锁后发现read中有了，该e是被删除状态，将其更新为nil，并且给dirty中增加该键值对，因为此时dirty中没有。然后更新e的值

  • 如果read没有，但dirty有，更新dirty中该entry的值，返回

  • 如果dirty，read都没有

    • 如果是刚提升dirty到read，此时dirty为空，需要将read浅拷贝到dirty中
    • 如果不是，则只在dirty中增加键值对

总的来说就是分各种情况处理：

• read有：无锁更新read中的数据
• read没有但dirty有：更新dirty中该entry的值
• read没有dirty也没有：将新的键值对添加到dirty中

来看一些小函数：

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
		}
	}
}

• tryStore：当entry.p不是expunged时，通过CAS的方式设置value

func (m *Map) dirtyLocked() {
   if m.dirty != nil {
      return
   }
   // 将read浅拷贝到dirty中
   read, _ := m.read.Load().(readOnly)
   m.dirty = make(map[any]*entry, len(read.m))
   for k, e := range read.m {
      if !e.tryExpungeLocked() {
         m.dirty[k] = e
      }
   }
}

• dirtyLocked：

  • 刚刚将dirty提升为read后，dirty为空，因此需要欧诺个read中浅拷贝一份。
  • 将read浅拷贝到dirty中，如果read中entry为空，该键值对就不会被拷贝到dirty，并将该entry置为expunged

Delete

func (m *Map) Delete(key any) {
   m.LoadAndDelete(key)
}

func (m *Map) LoadAndDelete(key any) (value any, loaded bool) {
   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 {
         // 如果该key不在read中，在dirty中，调用map原生的删除方法删除
         e, ok = m.dirty[key]
         delete(m.dirty, key)
         // 更新misses值
         m.missLocked()
      }
      m.mu.Unlock()
   }
   // 如果该key存在于read中，执行e.delete删除
   if ok {
      return e.delete()
   }
   return nil, false
}

• 其中e.delete方法如下：

  • 如果已经是被删除状态，直接返回
  • 否则将e.p更新为nil

func (e *entry) delete() (value any, ok bool) {
   for {
      p := atomic.LoadPointer(&e.p)
      if p == nil || p == expunged {
         return nil, false
      }
      if atomic.CompareAndSwapPointer(&e.p, p, nil) {
         return *(*any)(p), true
      }
   }
}

删除流程比较简单，如果在read里，就将其entry置位nil，如果不在read，就加锁去dirty删

为啥read的删除不像dirty一样，调用内置delete函数删除？

• 因为read是只读结构，不能对hash表的结构做修改，而只能做逻辑删除，即将entry.p设为nil

由于这里已经被删除，重建ditry时（从read浅拷贝），如果发现该key对应的entry已经被删除，即等于nil，就不把该键值对复制到dirty

• 为啥不复制该键值对？

  • 如果复制过去，但后续没有再对这个被删除的键值对进行操作，就会浪费内存空间

• read中该被删除的key，啥时候真正删除？

  • 假设后续没有对该key进行操作，等后续misses达到阈值，将dirty提升为read时，就能真正的从sync.map中删除该键值对

• 如果后续对该key进行操作咋办？

  • 回到Store流程里：

  • // 如果加锁后发现read中有了
       if e, ok := read.m[key]; ok {
          // 如果该e是被删除状态，将其更新为nil
          if e.unexpungeLocked() {
             // 并且给dirty中增加该键值对，因为此时dirty中没有
             m.dirty[key] = e
          }
          // 更新value
          e.storeLocked(&value)
    

  • 若发现read中该entry为expunge，说明此时dirty中没有该键值对，因此需要去dirty中进行添加，同时将这次Store的新value放入entry中

  • 这也是sync.map设置expunge这个特殊值的意义所在：

    • 区分这个entry为空的键值对是否存在于dirty中，若为expunge，说明不在

Range

func (m *Map) Range(f func(key, value any) bool) {
   read, _ := m.read.Load().(readOnly)
   if read.amended {
      m.mu.Lock()
      read, _ = m.read.Load().(readOnly)
      if read.amended {
         read = readOnly{m: m.dirty}
         m.read.Store(read)
         m.dirty = nil
         m.misses = 0
      }
      m.mu.Unlock()
   }

   for k, e := range read.m {
      v, ok := e.load()
      if !ok {
         continue
      }
      if !f(k, v) {
         break
      }
   }
}

Range方法比较简单，如果dirty数据比read多，执行一次提升操作，然后遍历read

因为read不可变，所以这次遍历不会有并发安全问题，这也是copy on write思想的应用

总结

• sync.Map 是线程安全的

• 通过只读和可写分离，使得查询，更新已存在key的value不需要加锁

• 随着程序的运行，dirty和read的差距会越来越大，使得需要加锁访问dirty的概率变大，效率也下降。因此当misses达到阈值时，将dirty提升为read，减低加锁的概率

• 提升后第一次新增键值对时，会将read浅拷贝一份成为dirty，但会过滤掉entry为nil的键值对

• 当 dirty 为 nil 的时候，read 就代表 map 所有的数据；当 dirty 不为 nil 的时候，dirty 才代表 map 所有的数据