Go中的Map和并发写入错误

1. 概述

Map是Go中最常用的数据结构，仅次于slice。但map在多个goroutine并发访问时并不安全。如果你从多个goroutine写入一个map，程序可能会因为fatal error: concurrent map writes而完全崩溃。

本文回顾map的基本操作。然后解释为什么会出现上述错误，如何用race detector检测它，以及三种修复方案：sync.Mutex、sync.RWMutex、sync.Map。示例使用Go 1.22及以上版本。

2. Go中的Map：需要记住的要点

2.1. 基本操作

package main

import "fmt"

func main() {
	stock := map[string]int{"apple": 10, "banana": 5}
	stock["orange"] = 7

	qty, ok := stock["grape"]
	fmt.Println(qty, ok)

	delete(stock, "banana")
	fmt.Println(len(stock))
}

解释：

• map[string]int{...}创建一个key为string、value为int的map。
• stock["grape"]当key不存在时返回零值（0）。变量ok告诉你这个key是否真的在map中。
• delete删除一个key。对不存在的key调用delete不会导致错误。

0 false
2

2.2. nil map

声明var m map[K]V而不初始化会给你一个nil map。从nil map读取是可以的，但写入会panic：

package main

func main() {
	var orders map[string]int
	_ = orders["A-001"] // OK，返回0
	orders["A-001"] = 1 // panic
}

panic: assignment to entry in nil map

修复的方法是总是用make(map[string]int)或literal map[string]int{}来初始化map。当map是struct中的字段且你忘记在构造函数中初始化时，这个错误很常见。

2.3. Map是引用类型，遍历顺序是随机的

当把map赋给另一个变量或传入函数时，它们都指向同一个底层数据结构。因此两个goroutine接收“同一个map”实际上是在共享同一块内存。这是下一小节问题的根源。

另外，for k, v := range m的遍历顺序被runtime故意打乱。不要写依赖于这个顺序的代码或测试。如果需要固定顺序，取出所有key、排序，然后遍历。

3. 并发写入错误

3.1. 重现错误

假设你按用户计数访问次数，每个请求在一个goroutine上运行：

package main

import (
	"fmt"
	"sync"
)

func main() {
	visits := make(map[string]int)
	var wg sync.WaitGroup

	for i := 0; i < 100; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for j := 0; j < 1000; j++ {
				visits[fmt.Sprintf("user-%d", j%10)]++
			}
		}()
	}

	wg.Wait()
	fmt.Println(len(visits))
}

多次运行，你几乎肯定会看到：

fatal error: concurrent map writes

goroutine 23 [running]:
main.main.func1()
	/tmp/visits/main.go:17 +0x...
...
exit status 2

visits[...]++包含一次读取和一次写入。当100个goroutine同时做这个操作时，多个goroutine可能同时修改map的内部结构（bucket、增长过程）。Go的runtime有一个简单的检查机制：map保持一个“正在写入”的标志。如果一个goroutine在开始写入时看到这个标志已经开启，runtime会立即停止程序。

类似的变体发生在当一个goroutine正在读取而另一个在写入时：

fatal error: concurrent map read and map write

3.2. 为什么不能用recover

很多人尝试用recover包裹代码：

go func() {
	defer func() {
		if r := recover(); r != nil {
			fmt.Println("recovered:", r)
		}
	}()
	visits["user-1"]++
}()

这种方法没有用。concurrent map writes不是panic。这是runtime调用throw的fatal error，它会终止整个进程。recover无法捕获它。这个选择是有意的：如果map已经损坏，继续运行比直接停止更危险。

注意： Runtime的检测机制只是“尽力而为”。程序没有崩溃不意味着代码是安全的。可能你只是还没遇到两个goroutine恰好碰撞的时刻，而数据已经悄悄出错了。

3.3. 用race detector提前检测

更可靠的工具是race detector，用-race标志启用：

go run -race main.go

==================
WARNING: DATA RACE
Read at 0x00c000... by goroutine 8:
  runtime.mapaccess1_faststr()
  main.main.func1()
      /tmp/visits/main.go:17 +0x...

Previous write at 0x00c000... by goroutine 7:
  runtime.mapassign_faststr()
  main.main.func1()
      /tmp/visits/main.go:17 +0x...
==================

Race detector准确指出了冲突的读取和写入代码行。你应该在CI中运行测试时启用-race。用-race运行的程序会更慢、消耗更多内存，所以通常不会在production构建中使用这个标志。

4. 第一种修复方案：sync.Mutex

最常见和最易理解的方法是把map包裹在一个有mutex的struct中，然后只通过method访问map：

package visits

import (
	"maps"
	"sync"
)

type VisitCounter struct {
	mu     sync.Mutex
	counts map[string]int
}

func NewVisitCounter() *VisitCounter {
	return &VisitCounter{counts: make(map[string]int)}
}

func (c *VisitCounter) Inc(user string) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.counts[user]++
}

func (c *VisitCounter) Get(user string) int {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.counts[user]
}

func (c *VisitCounter) Snapshot() map[string]int {
	c.mu.Lock()
	defer c.mu.Unlock()
	return maps.Clone(c.counts)
}

解释：

• counts字段是小写的，所以包外的代码无法直接访问map。
• Inc和Get都锁住mutex，所以在任何时刻只有一个goroutine操作map。
• Snapshot用maps.Clone返回一个副本（Go 1.21引入）。如果直接返回c.counts，调用者会在没有锁的情况下遍历map，你又回到了原始的错误。
• 构造函数NewVisitCounter初始化map，避免第2.2小节的nil map错误。

4.1. 用-race写测试

Go不使用JUnit或AssertJ。标准的testing包对这个例子来说已经足够了：

package visits

import (
	"sync"
	"testing"
)

func TestVisitCounter_ConcurrentInc(t *testing.T) {
	counter := NewVisitCounter()
	var wg sync.WaitGroup

	for i := 0; i < 100; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for j := 0; j < 1000; j++ {
				counter.Inc("alice")
			}
		}()
	}
	wg.Wait()

	if got := counter.Get("alice"); got != 100_000 {
		t.Fatalf("Get(alice) = %d, want 100000", got)
	}
}

go test -race ./...

ok  	example.com/visits	1.214s

测试检查两个事情。首先，结果是100,000，即没有丢失任何增量。其次，race detector没有报告DATA RACE。尝试从Inc中移除c.mu.Lock()，然后再运行，测试会立即失败。

4.2. mutex的常见错误

复制包含mutex的struct。 如果你声明一个value receiver的method，如func (c VisitCounter) Inc(...)，每次调用method时会使用一个mutex的副本。对副本上的锁定无法保护任何东西。总是使用pointer receiver。go vet会通过copylocks检查来警告这种情况：

./visits.go:18:9: Inc passes lock by value: visits.VisitCounter contains sync.Mutex

持有锁太久。 不要在持有锁时调用HTTP、查询数据库或写文件。取出你需要的数据，释放锁，然后再做慢速操作。

检查然后执行分离。 下面的代码有两个单独的锁定，所以两次之间其他goroutine可能插入：

if counter.Get("bob") == 0 {
	counter.Inc("bob") // 可能已经有其他goroutine先增加了
}

如果需要“如果还没有就创建”的逻辑，把它写成一个method，在整个method期间持有锁。

5. 第二种修复方案：sync.RWMutex

对于读取非常频繁但很少写入的数据，比如缓存配置或产品价格表，sync.RWMutex允许多个goroutine并行读取：

package pricing

import "sync"

type PriceTable struct {
	mu     sync.RWMutex
	prices map[string]int64
}

func NewPriceTable() *PriceTable {
	return &PriceTable{prices: make(map[string]int64)}
}

func (p *PriceTable) Price(sku string) (int64, bool) {
	p.mu.RLock()
	defer p.mu.RUnlock()
	v, ok := p.prices[sku]
	return v, ok
}

func (p *PriceTable) SetPrice(sku string, cents int64) {
	p.mu.Lock()
	defer p.mu.Unlock()
	p.prices[sku] = cents
}

RLock允许多个goroutine同时读取。Lock是排它性的：它等待所有reader完成，并在写完之前阻止新的reader。

注意： RWMutex并不总是比Mutex更快。它本身有更高的管理开销。当写入比例相当可观，或锁内的代码段非常短时，Mutex通常与之相当或更快。在改变之前用benchmark来测量。

一个常见的错误：调用p.mu.RLock()然后在其中调用另一个也RLock的method。如果两者之间有一个writer在等待，程序可能会deadlock。原因是当已经有writer排队时RWMutex会阻止新的reader。

6. 第三种修复方案：sync.Map

sync包有sync.Map，一个线程安全的map，不需要自己管理锁：

package main

import (
	"fmt"
	"sync"
)

type Session struct {
	UserID string
}

func main() {
	var sessions sync.Map

	sessions.Store("token-abc", Session{UserID: "alice"})

	actual, loaded := sessions.LoadOrStore("token-abc", Session{UserID: "bob"})
	fmt.Println(actual.(Session).UserID, loaded)

	if v, ok := sessions.Load("token-abc"); ok {
		fmt.Println(v.(Session).UserID)
	}

	sessions.Range(func(key, value any) bool {
		fmt.Println(key, value.(Session).UserID)
		return true
	})
}

alice true
alice
token-abc alice

解释：

• Store写入一个key/value对。
• LoadOrStore是一个原子操作。如果key已经存在，method返回旧值并loaded = true，同时不覆盖。这是用sync.Map处理检查然后执行情况的正确方式。
• Range遍历map。回调函数返回false可以提前停止。

sync.Map有一些你需要了解的限制：

• Key和value类型为any，所以你失去了编译时类型检查，到处都要做type assertion。
• 没有Len()。要计数元素，必须用Range遍历。
• sync.Map针对两个具体情况进行了优化。一个是每个key写入一次然后读取很多次，比如只增加的缓存。二个是各goroutine读写不相交的key集合。

注意： 除了上述两种情况外，一个map加上Mutex或RWMutex通常更快、更易读、类型更安全。不要默认把所有map改成sync.Map。

7. 选择哪个方案？

情况	推荐选择
读写交替，有条件逻辑	map + sync.Mutex
读为主，写很少	map + sync.RWMutex（先测量）
只增加的缓存，或每个goroutine用独立key集	sync.Map
Map只在单个goroutine中使用	普通map，无需锁

还有一种方法：让单个goroutine拥有map，其他goroutine通过channel发送请求。当map的更新与现有的处理循环绑定时这种方式合适。对于简单的计数器，这比mutex更冗长，没有明显的好处。

关于只读map：如果map在goroutine启动之前初始化完毕，之后没有人再写，那么多个goroutine同时读取是安全的，不需要锁。

8. 总结

Go中的map很快也很方便，但对并发不安全。当多个goroutine同时写入，或既读又写时，runtime可能会因为fatal error: concurrent map writes而停止程序。这个错误无法recover。程序不crash也不能证明代码正确，所以要用go test -race运行测试来尽早发现data race。

在大多数情况下，你只需要把map包裹在一个有sync.Mutex的struct里，通过pointer receiver的method访问，并返回副本而不是原始map。当读取负载真的压倒性地占优时使用sync.RWMutex，只在sync.Map被设计的情况下才用它。