<p>在Go 1.6之前， 内置的map类型是部分goroutine安全的，并发的读没有问题，并发的写可能有问题。自go 1.6之后， 并发地读写map会报错，这在一些知名的开源库中都存在这个问题，所以go 1.9之前的解决方案是额外绑定一个锁，封装成一个新的struct或者单独使用锁都可以。</p>    <p>本文带你深入到  <code>sync.Map</code> 的具体实现中，看看为了增加一个功能，代码是如何变的复杂的,以及作者在实现  <code>sync.Map</code> 的一些思想。 </p>    <h3>有并发问题的map</h3>    <p>官方的  <a href="/misc/goto?guid=4959750380345159084" rel="nofollow,noindex">faq</a> 已经提到内建的  <code>map</code> 不是线程(goroutine)安全的。 </p>    <p>首先，让我们看一段并发读写的代码,下列程序中一个goroutine一直读，一个goroutine一只写同一个键值，即即使读写的键不相同，而且map也没有"扩容"等操作，代码还是会报错。</p>    <pre>  <code class="language-go">package main    func main() {   m := make(map[int]int)     go func() {    for {     _ = m[1]    }   }()     go func() {    for {     m[2] =2    }   }()     select {}  }  </code></pre>    <p>错误信息是:  <code>fatal error: concurrent map read and map write</code> 。 </p>    <p>如果你查看Go的源代码:  <a href="/misc/goto?guid=4959750380435084902" rel="nofollow,noindex">hashmap_fast.go#L118</a> ,会看到读的时候会检查  <code>hashWriting</code> 标志， 如果有这个标志，就会报并发错误。 </p>    <p>写的时候会设置这个标志:  <a href="/misc/goto?guid=4959750380513414202" rel="nofollow,noindex">hashmap.go#L542</a></p>    <pre>  <code class="language-go">h.flags |= hashWriting  </code></pre>    <p><a href="/misc/goto?guid=4959750380598888434" rel="nofollow,noindex">hashmap.go#L628</a> 设置完之后会取消这个标记。 </p>    <p>当然，代码中还有好几处并发读写的检查， 比如写的时候也会检查是不是有并发的写，删除键的时候类似写，遍历的时候并发读写问题等。</p>    <p>有时候，map的并发问题不是那么容易被发现, 你可以利用  <code>-race</code> 参数来检查。 </p>    <h3>Go 1.9之前的解决方案</h3>    <p>但是，很多时候，我们会并发地使用map对象，尤其是在一定规模的项目中，map总会保存goroutine共享的数据。在Go官方blog的  <a href="/misc/goto?guid=4959750380690184421" rel="nofollow,noindex">Go maps in action</a> 一文中，提供了一种简便的解决方案。 </p>    <pre>  <code class="language-go">var counter = struct{      sync.RWMutex      m map[string]int  }{m: make(map[string]int)}  </code></pre>    <p>它使用嵌入struct为map增加一个读写锁。</p>    <p>读数据的时候很方便的加锁：</p>    <pre>  <code class="language-go">counter.RLock()  n := counter.m["some_key"]  counter.RUnlock()  fmt.Println("some_key:", n)  </code></pre>    <p>写数据的时候:</p>    <pre>  <code class="language-go">counter.Lock()  counter.m["some_key"]++  counter.Unlock()  </code></pre>    <h3>sync.Map</h3>    <p>可以说，上面的解决方案相当简洁，并且利用读写锁而不是Mutex可以进一步减少读写的时候因为锁带来的性能。</p>    <p>但是，它在一些场景下也有问题，如果熟悉Java的同学，可以对比一下java的  <code>ConcurrentHashMap</code> 的实现，在map的数据非常大的情况下，一把锁会导致大并发的客户端共争一把锁，Java的解决方案是  <code>shard</code> , 内部使用多个锁，每个区间共享一把锁，这样减少了数据共享一把锁带来的性能影响，  <a href="/misc/goto?guid=4959750380777325125" rel="nofollow,noindex">orcaman</a> 提供了这个思路的一个实现：  <a href="/misc/goto?guid=4959750380862219815" rel="nofollow,noindex">concurrent-map</a> ，他也询问了Go相关的开发人员是否在Go中也实现这种  <a href="/misc/goto?guid=4959750380948615463" rel="nofollow,noindex">方案</a> ，由于实现的复杂性，答案是  <code>Yes, we considered it.</code> ,但是除非有特别的性能提升和应用场景，否则没有进一步的开发消息。 </p>    <p>那么，在Go 1.9中  <code>sync.Map</code> 是怎么实现的呢？它是如何解决并发提升性能的呢？ </p>    <p><code>sync.Map</code> 的实现有几个优化点，这里先列出来，我们后面慢慢分析。 </p>    <ol>     <li>空间换时间。 通过冗余的两个数据结构(read、dirty),实现加锁对性能的影响。</li>     <li>使用只读数据(read)，避免读写冲突。</li>     <li>动态调整，miss次数多了之后，将dirty数据提升为read。</li>     <li>double-checking。</li>     <li>延迟删除。 删除一个键值只是打标记，只有在提升dirty的时候才清理删除的数据。</li>     <li>优先从read读取、更新、删除，因为对read的读取不需要锁。</li>    </ol>    <p>下面我们介绍  <code>sync.Map</code> 的重点代码，以便理解它的实现思想。 </p>    <p>首先，我们看一下  <code>sync.Map</code> 的数据结构： </p>    <pre>  <code class="language-go">type Map struct {   // 当涉及到dirty数据的操作的时候，需要使用这个锁   mu Mutex     // 一个只读的数据结构，因为只读，所以不会有读写冲突。   // 所以从这个数据中读取总是安全的。   // 实际上，实际也会更新这个数据的entries,如果entry是未删除的(unexpunged), 并不需要加锁。如果entry已经被删除了，需要加锁，以便更新dirty数据。   read atomic.Value // readOnly     // dirty数据包含当前的map包含的entries,它包含最新的entries(包括read中未删除的数据,虽有冗余，但是提升dirty字段为read的时候非常快，不用一个一个的复制，而是直接将这个数据结构作为read字段的一部分),有些数据还可能没有移动到read字段中。   // 对于dirty的操作哦需要加锁，因为对它的操作可能会有读写竞争。   // 当dirty为空的时候， 比如初始化或者刚提升完，下一次的写操作会复制read字段中未删除的数据到这个数据中。   dirty map[interface{}]*entry     // 当从Map中读取entry的时候，如果read中不包含这个entry,会尝试从dirty中读取，这个时候会将misses加一，   // 当misses累积到 dirty的长度的时候， 就会将dirty提升为read,避免从dirty中miss太多次。因为操作dirty需要加锁。   misses int  }  </code></pre>    <p>它的数据结构很简单，值包含四个字段：  <code>read</code> 、  <code>mu</code> 、  <code>dirty</code> 、  <code>misses</code> 。 </p>    <p>它使用了冗余的数据结构  <code>read</code> 、  <code>dirty</code> 。  <code>dirty</code> 中会包含  <code>read</code> 中为删除的entries，新增加的entries会加入到  <code>dirty</code> 中。 </p>    <p><code>read</code> 的数据结构是： </p>    <pre>  <code class="language-go">type readOnly struct {   m       map[interface{}]*entry   amended bool // 如果Map.dirty有些数据不在中的时候，这个值为true  }  </code></pre>    <p><code>amended</code> 指明  <code>Map.dirty</code> 中有  <code>readOnly.m</code> 未包含的数据，所以如果从  <code>Map.read</code> 找不到数据的话，还要进一步到  <code>Map.dirty</code> 中查找。 </p>    <p>对Map.read的修改是通过原子操作进行的。</p>    <p>虽然  <code>read</code> 和  <code>dirty</code> 有冗余数据，但这些数据是通过指针指向同一个数据，所以尽管Map的value会很大，但是冗余的空间占用还是有限的。 </p>    <p><code>readOnly.m</code> 和  <code>Map.dirty</code> 存储的值类型是  <code>*entry</code> ,它包含一个指针p, 指向用户存储的value值。 </p>    <pre>  <code class="language-go">type entry struct {   p unsafe.Pointer // *interface{}  }  </code></pre>    <p>p有三种值：</p>    <ul>     <li>nil: entry已被删除了，并且m.dirty为nil</li>     <li>expunged: entry已被删除了，并且m.dirty不为nil，而且这个entry不存在于m.dirty中</li>     <li>其它： entry是一个正常的值</li>    </ul>    <p>以上是  <code>sync.Map</code> 的数据结构，下面我们重点看看  <code>Load</code> 、  <code>Store</code> 、  <code>Delete</code> 、  <code>Range</code> 这四个方法，其它辅助方法可以参考这四个方法来理解。 </p>    <p>Load</p>    <p>加载方法，也就是提供一个键  <code>key</code> ,查找对应的值  <code>value</code> ,如果不存在，通过  <code>ok</code> 反映： </p>    <pre>  <code class="language-go">func (m *Map) Load(key interface{}) (value interface{}, ok bool) {   // 1.首先从m.read中得到只读readOnly,从它的map中查找，不需要加锁   read, _ := m.read.Load().(readOnly)   e, ok := read.m[key]     // 2. 如果没找到，并且m.dirty中有新数据，需要从m.dirty查找，这个时候需要加锁   if !ok && read.amended {    m.mu.Lock()    // 双检查，避免加锁的时候m.dirty提升为m.read,这个时候m.read可能被替换了。    read, _ = m.read.Load().(readOnly)    e, ok = read.m[key]      // 如果m.read中还是不存在，并且m.dirty中有新数据    if !ok && read.amended {     // 从m.dirty查找     e, ok = m.dirty[key]     // 不管m.dirty中存不存在，都将misses计数加一     // missLocked()中满足条件后就会提升m.dirty     m.missLocked()    }    m.mu.Unlock()   }   if !ok {    return nil, false   }   return e.load()  }  </code></pre>    <p>这里有两个值的关注的地方。一个是首先从  <code>m.read</code> 中加载，不存在的情况下，并且  <code>m.dirty</code>中有新数据，加锁，然后从  <code>m.dirty</code> 中加载。 </p>    <p>二是这里使用了双检查的处理，因为在下面的两个语句中，这两行语句并不是一个原子操作。</p>    <pre>  <code class="language-go">if !ok && read.amended {    m.mu.Lock()  </code></pre>    <p>虽然第一句执行的时候条件满足，但是在加锁之前，  <code>m.dirty</code> 可能被提升为  <code>m.read</code> ,所以加锁后还得再检查  <code>m.read</code> ，后续的方法中都使用了这个方法。 </p>    <p>双检查的技术Java程序员非常熟悉了，单例模式的实现之一就是利用双检查的技术。</p>    <p>可以看到，如果我们查询的键值正好存在于  <code>m.read</code> 中，无须加锁，直接返回，理论上性能优异。即使不存在于  <code>m.read</code> 中，经过  <code>miss</code> 几次之后，  <code>m.dirty</code> 会被提升为  <code>m.read</code> ，又会从 <code>m.read</code> 中查找。所以对于更新／增加较少，加载存在的key很多的case,性能基本和无锁的map类似。 </p>    <p>下面看看  <code>m.dirty</code> 是如何被提升的。  <code>missLocked</code> 方法中可能会将  <code>m.dirty</code> 提升。 </p>    <pre>  <code class="language-go">func (m *Map) missLocked() {   m.misses++   if m.misses < len(m.dirty) {    return   }   m.read.Store(readOnly{m: m.dirty})   m.dirty = nil   m.misses =0  }  </code></pre>    <p>上面的最后三行代码就是提升  <code>m.dirty</code> 的，很简单的将  <code>m.dirty</code> 作为  <code>readOnly</code> 的  <code>m</code> 字段，原子更新  <code>m.read</code> 。提升后  <code>m.dirty</code> 、  <code>m.misses</code> 重置， 并且  <code>m.read.amended</code> 为false。 </p>    <p>Store</p>    <p>这个方法是更新或者新增一个entry。</p>    <pre>  <code class="language-go">func (m *Map) Store(key, value interface{}) {   // 如果m.read存在这个键，并且这个entry没有被标记删除，尝试直接存储。   // 因为m.dirty也指向这个entry,所以m.dirty也保持最新的entry。   read, _ := m.read.Load().(readOnly)   if e, ok := read.m[key]; ok && e.tryStore(&value) {    return   }     // 如果`m.read`不存在或者已经被标记删除   m.mu.Lock()   read, _ = m.read.Load().(readOnly)     if e, ok := read.m[key]; ok {    if e.unexpungeLocked() { //标记成未被删除     m.dirty[key] = e //m.dirty中不存在这个键，所以加如m.dirty    }    e.storeLocked(&value) //更新   } else if e, ok := m.dirty[key]; ok { // m.dirty存在这个键，更新    e.storeLocked(&value)   } else { //新键值    if !read.amended { //m.dirty中没有新的数据，往m.dirty中增加第一个新键     m.dirtyLocked() //从m.read中复制未删除的数据     m.read.Store(readOnly{m: read.m, amended: true})    }    m.dirty[key] = newEntry(value) //将这个entry加入到m.dirty中   }   m.mu.Unlock()  }    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 {    // 将已经删除标记为nil的数据标记为expunged    if atomic.CompareAndSwapPointer(&e.p, nil, expunged) {     return true    }    p = atomic.LoadPointer(&e.p)   }   return p == expunged  }  </code></pre>    <p>你可以看到，以上操作都是先从操作  <code>m.read</code> 开始的，不满足条件再加锁，然后操作  <code>m.dirty</code>。 </p>    <p><code>Store</code> 可能会在某种情况下(初始化或者m.dirty刚被提升后)从  <code>m.read</code> 中复制数据，如果这个时候  <code>m.read</code> 中数据量非常大，可能会影响性能。 </p>    <p>Delete</p>    <p>删除一个键值。</p>    <pre>  <code class="language-go">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()   }  }  </code></pre>    <p>同样，删除操作还是从  <code>m.read</code> 中开始， 如果这个entry不存在于  <code>m.read</code> 中，并且  <code>m.dirty</code> 中有新数据，则加锁尝试从  <code>m.dirty</code> 中删除。 </p>    <p>注意，还是要双检查的。 从  <code>m.dirty</code> 中直接删除即可，就当它没存在过，但是如果是从  <code>m.read</code> 中删除，并不会直接删除，而是打标记： </p>    <pre>  <code class="language-go">func (e *entry) delete() (hadValue bool) {   for {    p := atomic.LoadPointer(&e.p)    // 已标记为删除    if p == nil || p == expunged {     return false    }    // 原子操作，e.p标记为nil    if atomic.CompareAndSwapPointer(&e.p, p, nil) {     return true    }   }  }  </code></pre>    <p>Range</p>    <p>因为  <code>for ... range map</code> 是内建的语言特性，所以没有办法使用  <code>for range</code> 遍历  <code>sync.Map</code>, 但是可以使用它的  <code>Range</code> 方法，通过回调的方式遍历。 </p>    <pre>  <code class="language-go">func (m *Map) Range(f func(key, value interface{}) bool) {   read, _ := m.read.Load().(readOnly)     // 如果m.dirty中有新数据，则提升m.dirty,然后在遍历   if read.amended {    //提升m.dirty    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 range是安全的   for k, e := range read.m {    v, ok := e.load()    if !ok {     continue    }    if !f(k, v) {     break    }   }  }  </code></pre>    <p>Range方法调用前可能会做一个  <code>m.dirty</code> 的提升，不过提升  <code>m.dirty</code> 不是一个耗时的操作。 </p>    <h3>sync.Map的性能</h3>    <p>Go 1.9源代码中提供了性能的测试：  <a href="/misc/goto?guid=4959750381035551682" rel="nofollow,noindex">map_bench_test.go</a> 、  <a href="/misc/goto?guid=4959750381117239191" rel="nofollow,noindex">map_reference_test.go</a></p>    <p>我也基于这些代码修改了一下，得到下面的测试数据，相比较以前的解决方案，性能多少回有些提升，如果你特别关注性能，可以考虑  <code>sync.Map</code> 。 </p>    <pre>  <code class="language-go">BenchmarkHitAll/*sync.RWMutexMap-4    20000000         83.8 ns/op  BenchmarkHitAll/*sync.Map-4           30000000         59.9 ns/op  BenchmarkHitAll_WithoutPrompting/*sync.RWMutexMap-4          20000000         96.9 ns/op  BenchmarkHitAll_WithoutPrompting/*sync.Map-4                 20000000         64.1 ns/op  BenchmarkHitNone/*sync.RWMutexMap-4                          20000000         79.1 ns/op  BenchmarkHitNone/*sync.Map-4                                 30000000         43.3 ns/op  BenchmarkHit_WithoutPrompting/*sync.RWMutexMap-4             20000000         81.5 ns/op  BenchmarkHit_WithoutPrompting/*sync.Map-4                    30000000         44.0 ns/op  BenchmarkUpdate/*sync.RWMutexMap-4                            5000000        328 ns/op  BenchmarkUpdate/*sync.Map-4                                  10000000        146 ns/op  BenchmarkUpdate_WithoutPrompting/*sync.RWMutexMap-4           5000000        336 ns/op  BenchmarkUpdate_WithoutPrompting/*sync.Map-4                  5000000        324 ns/op  BenchmarkDelete/*sync.RWMutexMap-4                           10000000        155 ns/op  BenchmarkDelete/*sync.Map-4                                  30000000         55.0 ns/op  BenchmarkDelete_WithoutPrompting/*sync.RWMutexMap-4          10000000        173 ns/op  BenchmarkDelete_WithoutPrompting/*sync.Map-4                 10000000        147 ns/op  </code></pre>    <h3>其它</h3>    <p><code>sync.Map</code> 没有  <code>Len</code> 方法，并且目前没有迹象要加上 (  <a href="/misc/goto?guid=4959750381207488213" rel="nofollow,noindex">issue#20680</a> ),所以如果想得到当前Map中有效的entries的数量，需要使用  <code>Range</code> 方法遍历一次， 比较X疼。 </p>    <p><code>LoadOrStore</code> 方法如果提供的key存在，则返回已存在的值(Load)，否则保存提供的键值(Store)。</p>    <p> </p>    <p>来自：http://colobu.com/2017/07/11/dive-into-sync-Map/</p>