Go中sync.map使用小结
<ul><li>sync.map
<ul>
<li>前言</li>
<li>深入了解下</li>
<li>查看下具体的实现
<ul>
<li>Load</li>
<li>Store</li>
<li>Delete</li>
<li>LoadOrStore</li>
</ul>
</li>
<li>总结</li>
<li>流程图片</li>
<li>参考</li>
</ul>
</li>
</ul>
<h2 id="syncmap">sync.map</h2>
<h3 id="前言">前言</h3>
<p>Go中的map不是并发安全的，在Go1.9之后，引入了<code>sync.Map</code>,并发安全的map。</p>
<h3 id="深入了解下">深入了解下</h3>
<p>对于map，我们常用的做法就是加锁。</p>
<p>对于<code>sync.Map</code>这些操作则是不需要的，来看下<code>sync.Map</code>的特点:</p>
<ul>
<li>1、以空间换效率，通过read和dirty两个map来提高读取效率</li>
<li>2、优先从read map中读取(无锁)，否则再从dirty map中读取(加锁)</li>
<li>3、动态调整，当misses次数过多时，将dirty map提升为read map</li>
<li>4、延迟删除，删除只是为value打一个标记，在dirty map提升时才执行真正的删除</li>
</ul>
<p>简单的使用栗子：</p>
<pre><code class="language-go">func syncMapDemo() {
var smp sync.Map
// 数据写入
smp.Store("name", "小红")
smp.Store("age", 18)
// 数据读取
name, _ := smp.Load("name")
fmt.Println(name)
age, _ := smp.Load("age")
fmt.Println(age)
// 遍历
smp.Range(func(key, value interface{}) bool {
fmt.Println(key, value)
return true
})
// 删除
smp.Delete("age")
age, ok := smp.Load("age")
fmt.Println("删除后的查询")
fmt.Println(age, ok)
// 读取或写入,存在就读取，不存在就写入
smp.LoadOrStore("age", 100)
age, _ = smp.Load("age")
fmt.Println("不存在")
fmt.Println(age)
smp.LoadOrStore("age", 99)
age, _ = smp.Load("age")
fmt.Println("存在")
fmt.Println(age)
}
</code></pre>
<h3 id="查看下具体的实现">查看下具体的实现</h3>
<pre><code class="language-go">// sync/map.go
type Map struct {
// 当写read map 或读写dirty map时 需要上锁
mu Mutex
// read map的 k v(entry) 是不变的，删除只是打标记，插入新key会加锁写到dirty中
// 因此对read map的读取无需加锁
read atomic.Value // 保存readOnly结构体
// dirty map 对dirty map的操作需要持有mu锁
dirty map*entry
// 当Load操作在read map中未找到，尝试从dirty中进行加载时(不管是否存在)，misses+1
// 当misses达到diry map len时，dirty被提升为read 并且重新分配dirty
misses int
}
// read map数据结构
type readOnly struct {
m map*entry
// 为true时代表dirty map中含有m中没有的元素
amended bool
}
type entry struct {
// 指向实际的interface{}
// p有三种状态:
// p == nil: 键值已经被删除，此时，m.dirty==nil 或 m.dirty指向该entry
// p == expunged: 键值已经被删除， 此时, m.dirty!=nil 且 m.dirty不存在该键值
// 其它情况代表实际interface{}地址 如果m.dirty!=nil 则 m.read 和 m.dirty 指向同一个entry
// 当删除key时，并不实际删除，先CAS entry.p为nil 等到每次dirty map创建时(dirty提升后的第一次新建Key)，会将entry.p由nil CAS为expunged
p unsafe.Pointer // *interface{}
}
</code></pre>
<p><code>read map</code> 和 <code>dirty map</code> 的存储方式是不一致的。</p>
<p>前者使用 <code>atomic.Value</code>，后者只是单纯的使用 map。原因是 <code>read map</code> 使用 <code>lock free</code> 操作，必须保证 <code>load/store</code> 的原子性；而 <code>dirty map</code> 的 <code>load+store</code> 操作是由 lock（就是 mu）来保护的。</p>
<p>1、read和dirty通过entry包装value，这样使得value的变化和map的变化隔离，前者可以用atomic无锁完成<br>
2、Map的read字段结构体定义为readOnly，这只是针对map*entry而言的，entry内的内容以及amended字段都是可以变的<br>
3、大部分情况下，对已有key的删除(entry.p置为nil)和更新可以直接通过修改entry.p来完成</p>
<h4 id="load">Load</h4>
<pre><code class="language-go">func (m *Map) Load(key interface{}) (value interface{}, ok bool) {
// 首先在通过atomic的原子操作读取内容
read, _ := m.read.Load().(readOnly)
e, ok := read.m
// 如果没在 read 中找到，并且 amended 为 true，即 dirty 中存在 read 中没有的 key
if !ok && read.amended {
// read调用了atomic的原子性，所以不用加锁，但是dirty map*entry就需要了，用的互斥锁
m.mu.Lock()
// double check，避免在加锁的时候dirty map提升为read map
read, _ = m.read.Load().(readOnly)
e, ok = read.m
// 还是没有找到
if !ok && read.amended {
// 从 dirty 中找
e, ok = m.dirty
// 不管dirty中有没有找到 都增加misses计数 该函数可能将dirty map提升为readmap
m.missLocked()
}
m.mu.Unlock()
}
if !ok {
return nil, false
}
return e.load()
}
// 从entry中atomic 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
}
</code></pre>
<p>梳理下处理的逻辑：</p>
<p>1、首先是 fast path，直接在 read 中找，如果找到了直接调用 entry 的 load 方法，取出其中的值。<br>
2、如果 read 中没有这个 key，且 amended 为 fase，说明 dirty 为空，那直接返回 空和 false。<br>
3、如果 read 中没有这个 key，且 amended 为 true，说明 dirty 中可能存在我们要找的 key。当然要先上锁，再尝试去 dirty 中查找。在这之前，仍然有一个 double check 的操作。若还是没有在 read 中找到，那么就从 dirty 中找。不管 dirty 中有没有找到，都要"记一笔"，因为在 dirty 被提升为 read 之前，都会进入这条路径</p>
<pre><code class="language-go">// 增加misses计数，并在必要的时候提升dirty map
// 如果 misses 值小于 m.dirty 的长度，就直接返回。否则，将 m.dirty 晋升为 read，并清空 dirty，清空 misses 计数值。这样，之前
// 一段时间新加入的 key 都会进入到 read 中，从而能够提升 read 的命中率。
func (m *Map) missLocked() {
m.misses++
if m.misses < len(m.dirty) {
return
}
// 提升过程很简单，直接将m.dirty赋给m.read.m
// 提升完成之后 amended == false m.dirty == nil
// m.dirty并不立即创建被拷贝元素，而是延迟创建
m.read.Store(readOnly{m: m.dirty})
m.dirty = nil
m.misses = 0
}
</code></pre>
<p>对于<code>missLocked</code>会直接将 misses 的值加 1，表示一次未命中，如果 misses 值小于 m.dirty 的长度，就直接返回。否则，将 m.dirty 晋升为 read，并清空 dirty，清空 misses 计数值。这样，之前一段时间新加入的 key 都会进入到 read 中，从而能够提升 read 的命中率。</p>
<h4 id="store">Store</h4>
<pre><code class="language-go">// Store sets the value for a key.
func (m *Map) Store(key, value interface{}) {
// 如果read map中存在该key则尝试直接更改(由于修改的是entry内部的pointer，因此dirty map也可见)
read, _ := m.read.Load().(readOnly)
if e, ok := read.m; ok && e.tryStore(&value) {
return
}
m.mu.Lock()
read, _ = m.read.Load().(readOnly)
if e, ok := read.m; ok {
if e.unexpungeLocked() {
// 如果 read map 中存在该 key，但 p == expunged，则说明 m.dirty != nil 并且 m.dirty 中不存在该 key 值 此时:
// a. 将 p 的状态由 expunged更改为 nil
// b. dirty map 插入 key
m.dirty = e
}
// 更新 entry.p = value (read map 和 dirty map 指向同一个 entry)
e.storeLocked(&value)
} else if e, ok := m.dirty; ok {
// 如果 read map 中不存在该 key，但 dirty map 中存在该 key，直接写入更新 entry(read map 中仍然没有这个 key)
e.storeLocked(&value)
} else {
// 如果read map和dirty map中都不存在该key，则:
// a. 如果dirty map为空，则需要创建dirty map，并从read map中拷贝未删除的元素
// b. 更新amended字段，标识dirty map中存在read map中没有的key
// c. 将k v写入dirty map中，read.m不变
if !read.amended {
m.dirtyLocked()
m.read.Store(readOnly{m: read.m, amended: true})
}
m.dirty = newEntry(value)
}
m.mu.Unlock()
}
// 尝试直接更新entry 如果p == expunged 返回false
func (e *entry) tryStore(i *interface{}) bool {
p := atomic.LoadPointer(&e.p)
if p == expunged {
return false
}
for {
if atomic.CompareAndSwapPointer(&e.p, p, unsafe.Pointer(i)) {
return true
}
p = atomic.LoadPointer(&e.p)
if p == expunged {
return false
}
}
}
func (e *entry) unexpungeLocked() (wasExpunged bool) {
return atomic.CompareAndSwapPointer(&e.p, expunged, nil)
}
// 如果 dirty map为nil，则从read map中拷贝元素到dirty map
func (m *Map) dirtyLocked() {
if m.dirty != nil {
return
}
read, _ := m.read.Load().(readOnly)
m.dirty = make(map*entry, len(read.m))
for k, e := range read.m {
// a. 将所有为 nil的 p 置为 expunged
// b. 只拷贝不为expunged 的 p
if !e.tryExpungeLocked() {
m.dirty = 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
}
</code></pre>
<p>梳理下流程：</p>
<p>1、首先还是去read map中查询，存在并且p!=expunged,直接修改。（由于修改的是 entry 内部的 pointer，因此 dirty map 也可见）<br>
2、如果read map中存在该key，但p == expunged。加锁更新p的状态，然后直接更新该entry (此时<code>m.dirty==nil</code>或<code>m.dirty==e</code>)<br>
3、如果read map中不存在该Key，但dirty map中存在该key，直接写入更新entry(read map中仍然没有)<br>
4、如果read map和dirty map都不存在该key</p>
<ul>
<li>a. 如果dirty map为空，则需要创建dirty map，并从read map中拷贝未删除的元素</li>
<li>b. 更新amended字段，标识dirty map中存在read map中没有的key</li>
<li>c. 将k v写入dirty map中，read.m不变</li>
</ul>
<h4 id="delete">Delete</h4>
<pre><code class="language-go">// Delete deletes the value for a key.
func (m *Map) Delete(key interface{}) {
// 从read map中寻找
read, _ := m.read.Load().(readOnly)
e, ok := read.m
// 没找到
if !ok && read.amended { // read.amended为true代表dirty map中含有m中没有的元素
m.mu.Lock()
// double check
read, _ = m.read.Load().(readOnly)
e, ok = read.m
// 第二次仍然没找到，但dirty map中存在，则直接从dirty map删除
if !ok && read.amended {
delete(m.dirty, key)
}
m.mu.Unlock()
}
// 如果read存在，将entry.p 置为 nil
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
}
}
}
</code></pre>
<p>梳理下流程：<br>
1、先去read map中寻找，如果存在就直接删除<br>
2、如果没找到，并且 read.amended为true代表dirty map中存在，依照传统进行 double check。<br>
3、read map找到就删除，没找到判断dirty map是否存在，存在了就删除</p>
<h4 id="loadorstore">LoadOrStore</h4>
<pre><code class="language-go">func (m *Map) LoadOrStore(key, value interface{}) (actual interface{}, loaded bool) {
// 首先还是先去read map中查询
read, _ := m.read.Load().(readOnly)
if e, ok := read.m; ok {
actual, loaded, ok := e.tryLoadOrStore(value)
if ok {
return actual, loaded
}
}
m.mu.Lock()
// double check
read, _ = m.read.Load().(readOnly)
if e, ok := read.m; ok {
if e.unexpungeLocked() {
// 如果 read map 中存在该 key，但 p == expunged，则说明 m.dirty != nil 并且 m.dirty 中不存在该 key 值 此时:
// a. 将 p 的状态由 expunged更改为 nil
// b. dirty map 插入 key
m.dirty = e
}
actual, loaded, _ = e.tryLoadOrStore(value)
} else if e, ok := m.dirty; ok {
// 如果 read map 中不存在该 key，但 dirty map 中存在该 key
actual, loaded, _ = e.tryLoadOrStore(value)
// 不管dirty中有没有找到 都增加misses计数 该函数可能将dirty map提升为readmap
m.missLocked()
} else {
if !read.amended {
// 如果read map和dirty map中都不存在该key，则:
// a. 如果dirty map为空，则需要创建dirty map，并从read map中拷贝未删除的元素
// b. 更新amended字段，标识dirty map中存在read map中没有的key
// c. 将k v写入dirty map中，read.m不变
m.dirtyLocked()
m.read.Store(readOnly{m: read.m, amended: true})
}
m.dirty = newEntry(value)
actual, loaded = value, false
}
m.mu.Unlock()
return actual, loaded
}
// 如果entry is expunged则不处理，返回false
func (e *entry) tryLoadOrStore(i interface{}) (actual interface{}, loaded, ok bool) {
p := atomic.LoadPointer(&e.p)
if p == expunged {
return nil, false, false
}
if p != nil {
return *(*interface{})(p), true, true
}
// Copy the interface after the first load to make this method more amenable
// to escape analysis: if we hit the "load" path or the entry is expunged, we
// shouldn't bother heap-allocating.
ic := i
for {
if atomic.CompareAndSwapPointer(&e.p, nil, unsafe.Pointer(&ic)) {
return i, false, true
}
p = atomic.LoadPointer(&e.p)
if p == expunged {
return nil, false, false
}
if p != nil {
return *(*interface{})(p), true, true
}
}
}
</code></pre>
<p>这个函数结合了 Load 和 Store 的功能，如果 map 中存在这个 key，那么返回这个 key 对应的 value；否则，将 key-value 存入 map。<br>
这在需要先执行 Load 查看某个 key 是否存在，之后再更新此 key 对应的 value 时很有效，因为 LoadOrStore 可以并发执行。</p>
<h3 id="总结">总结</h3>
<p>除了Load/Store/Delete之外，sync.Map还提供了LoadOrStore/Range操作，但没有提供Len()方法，这是因为要统计有效的键值对只能先提<br>
升dirty map(dirty map中可能有read map中没有的键值对)，再遍历m.read(由于延迟删除，不是所有的键值对都有效)，这其实就是Range做的事情，<br>
因此在不添加新数据结构支持的情况下，sync.Map的长度获取和Range操作是同一复杂度的。这部分只能看官方后续支持。</p>
<p>1、sync.map 是线程安全的，读取，插入，删除也都保持着常数级的时间复杂度。</p>
<p>2、通过读写分离，降低锁时间来提高效率，适用于读多写少的场景。</p>
<p>3、Range 操作需要提供一个函数，参数是 k,v，返回值是一个布尔值：f func(key, value interface{}) bool。</p>
<p>4、调用 Load 或 LoadOrStore 函数时，如果在 read 中没有找到 key，则会将 misses 值原子地增加 1，当 misses 增加到和 dirty 的长度相等时，会将 dirty 提升为 read。以期减少“读 miss”。</p>
<p>5、新写入的 key 会保存到 dirty 中，如果这时 dirty 为 nil，就会先新创建一个 dirty，并将 read 中未被删除的元素拷贝到 dirty。</p>
<p>6、当 dirty 为 nil 的时候，read 就代表 map 所有的数据；当 dirty 不为 nil 的时候，dirty 才代表 map 所有的数据。</p>
<h3 id="流程图片">流程图片</h3>
<p>最后附上一张操作的流程图</p>
<img src="https://img2022.cnblogs.com/blog/1237626/202203/1237626-20220316202031420-1204200449.png">
<h3 id="参考">参考</h3>
<p>【Go sync.Map 实现】https://wudaijun.com/2018/02/go-sync-map-implement/<br>
【深度解密 Go 语言之 sync.map】https://www.cnblogs.com/qcrao-2018/p/12833787.html<br>
【golang源码阅读笔记】https://github.com/boilingfrog/Go-POINT/tree/master/golang<br>
【go中sync.Map源码刨铣】https://boilingfrog.github.io/2022/03/16/go中sync.map源码刨铣/</p><br><br>
来源：https://www.cnblogs.com/ricklz/p/13659397.html
頁:
[1]