11.1 Mutex/RWMutex 与 atomic
第 10 章我们给 TaskAPI 加了后台扫描器和 worker pool。它们都在同一个进程里跑，也都在读写同一份数据——但到目前为止，那份数据要么只被一个 goroutine 碰，要么干脆是个局部变量。
真实项目不会这么温柔。worker pool 里的 4 个 worker 同时调 store.Save，扫描器同时在调 store.List，HTTP 处理器（第 13 章）又在读同一个 map。只要有两个 goroutine 在没有同步的情况下访问同一块内存，且至少一个是写，数据竞态就成立了——它可能今天不发作，明天在线上以「一个任务莫名其妙消失」的形式爆发。
本节把 TaskAPI 推进到：
MemStore从「只能单 goroutine 用」升级为并发安全，并用atomic.Int64统计处理量。
11.1.1 数据竞态长什么样
先看一个最小例子。1000 个 goroutine 同时对同一个 map 做 counts[i%10]++：
counts := map[int64]int{}
var wg sync.WaitGroup
for i := int64(0); i < 1000; i++ {
wg.Go(func() {
counts[i%10]++ // 并发读写同一个 map
})
}
wg.Wait()
fmt.Println("完成")
实测输出（go1.27.0，Apple M1 Pro）：
fatal error: concurrent map writes
程序不是打印了错误的数字，而是直接崩了。Go 的 map 内部有并发写检测，一旦发现两个 goroutine 同时写，就抛 fatal error——注意这是 fatal error 不是 panic，recover 抓不住，进程直接死。
为什么是崩溃而不是算错？因为 map 的写入涉及扩容、搬迁桶、修改哈希元数据，中途被另一个写打断会让内部结构彻底损坏。与其让你拿到一个损坏的 map，不如立刻停下。
切片、结构体字段、普通计数器则没这么幸运——它们不会崩，只会静默算错。比如 n++ 在汇编层面是「读、加、写」三步，两个 goroutine 可能都读到 5，各自加 1 写回 6，于是一次自增凭空消失。这类 bug 最难查，因为它只在特定的时序下出现。
记住两个判断条件，缺一不可：
- 两个或更多 goroutine 访问同一块内存；
- 其中至少一个是写操作，且没有同步手段保护。
11.1.2 用 sync.Mutex 保护 map
sync.Mutex 最核心的两个方法是 Lock() 和 Unlock()（Go 1.18 起还多了一个非阻塞尝试加锁的 TryLock()）。它保证同一时刻最多只有一个 goroutine 在临界区里。
把 map 和锁绑成一个结构体，是最常见的做法：
type MemStore struct {
mu sync.Mutex
tasks map[int64]Task
}
func (s *MemStore) Save(t Task) {
s.mu.Lock()
defer s.mu.Unlock()
s.tasks[t.ID] = t
}
两个细节：
- 锁和它保护的数据放在同一个结构体里，紧挨着。这样读代码的人一眼就知道「这个 map 由 mu 保护」，也避免了「两把锁保护一份数据」的混乱。
defer s.mu.Unlock()紧跟Lock()。中间的代码一旦 panic，defer保证锁一定被释放。如果忘了写 defer 又提前 return，锁就永远不释放，整个程序死锁。
sync.Mutex 的零值就是「未锁定」，可以直接用，不需要构造函数：
var mu sync.Mutex // 可用
mu.Lock()
// ...
mu.Unlock()
这也是为什么 MemStore 里的 mu 不需要在 NewMemStore 里初始化。
11.1.3 千万别复制 Mutex
sync.Mutex 是不可复制的。一旦把它按值传出去或按值接收，你复制的只是「锁的状态快照」，两个副本各自加锁、互不干扰，等于没有锁。
下面这个错误非常隐蔽：
type Counter struct {
mu sync.Mutex
n int
}
func (c Counter) Inc() { // 值接收者：复制了整个 Counter，包括 Mutex
c.mu.Lock()
defer c.mu.Unlock()
c.n++ // 改的是副本
}
go vet 会直接报出来，而实际跑起来更糟——n 永远是 0，因为每次 Inc 都在改一个用完即弃的副本：
main.go:13:9: Inc passes lock by value: main.Counter contains sync.Mutex
n = 0
修正方式：接收者改成指针 func (c *Counter) Inc()。规则可以简化成一句话：
只要结构体里含
Mutex、RWMutex、WaitGroup、Once，这个结构体的方法一律用指针接收者，并且永远不要按值传递它。
同样地，sync.Mutex 也不该放进 map 的 value 里（map[string]sync.Mutex），因为从 map 里取 value 就是一次复制。要放就放指针。
11.1.4 sync.RWMutex：读多写少时更划算
TaskAPI 的访问模式很典型：Get、List、Count 这些读操作远多于 Save、Delete 这些写操作。而 sync.Mutex 太保守了——读操作之间根本不冲突，却也被迫排队。
sync.RWMutex 提供了两套锁：
| 方法 | 语义 |
|---|---|
Lock() / Unlock() | 写锁，独占，排斥所有读写 |
RLock() / RUnlock() | 读锁，可多个读锁同时持有，但与写锁互斥 |
改造成本极低，只需把读方法换成 RLock：
type MemStore struct {
mu sync.RWMutex
tasks map[int64]Task
}
func (s *MemStore) Save(t Task) {
s.mu.Lock() // 写：独占
defer s.mu.Unlock()
s.tasks[t.ID] = t
}
func (s *MemStore) Get(id int64) (Task, bool) {
s.mu.RLock() // 读：共享
defer s.mu.RUnlock()
t, ok := s.tasks[id]
return t, ok
}
Count() 这类只读方法同样用 RLock。用 go run -race 跑一段并发读写验证：100 个 goroutine 各自 Save 再 Get，最后 Count() 输出 100，且 -race 无任何竞态报告。
有三点必须注意：
- RLock 不能嵌套升级为 Lock。持着读锁再去拿写锁会死锁。官方文档明确写着
RWMutex.RLock cannot be upgraded into a RWMutex.Lock。需要「先读再改」时，直接用写锁把整段包起来。 - 写锁有优先级。一旦有 goroutine 在等
Lock()，后续的RLock()会排队等它，而不是插队。这是为了防止写操作被源源不断的读饿死。 - RLock 不是免费的。它要维护读者计数，在多核下这个计数是热点数据。所以「读多写少就用 RWMutex」不是无条件的——11.2 节会用实测数据说明什么时候反而不如
sync.Map。
11.1.5 锁的粒度
「粒度」指临界区的大小。粒度太粗，并发退化成串行；太细，又容易漏保护。
// 粒度合理：只把「读改写」这一步包进锁
func (s *MemStore) CloseOne(id int64) error {
s.mu.Lock()
defer s.mu.Unlock()
t, ok := s.tasks[id]
if !ok {
return ErrNotFound
}
t.Done = true
s.tasks[id] = t
return nil
}
反例是把批量操作整个塞进锁里，中间还夹着 time.Sleep 或数据库调用——那等于把所有并发都变回了串行，worker pool 白建了。原则是：临界区里只放「必须原子完成」的那几行，绝不放 IO、网络调用、time.Sleep。
一个实用的排查手法：如果加了锁之后性能反而没比单线程好多少，先去看临界区里是不是混进了耗时操作。
11.1.6 sync/atomic：单变量的无锁操作
对于单个整数或布尔值，用锁显得太重。sync/atomic 提供了原子操作，Go 1.19 之后推荐用类型化的封装：
type Stats struct {
processed atomic.Int64
failed atomic.Int64
}
常用方法：
| 方法 | 用途 |
|---|---|
Add(n) | 原子加，返回新值 |
Load() / Store(v) | 原子读 / 写 |
CompareAndSwap(old, new) | 仅当当前值等于 old 时才写入 new |
Swap(v) | 写入新值并返回旧值 |
用它替换普通计数器：
var st Stats
var wg sync.WaitGroup
for i := 0; i < 100; i++ {
wg.Go(func() { st.processed.Add(1) })
}
wg.Wait()
fmt.Println("processed =", st.processed.Load())
实测输出：
processed = 100
failed = 0
对比之下，把 Add 换成普通的 naive++，-race 立刻报错：
==================
WARNING: DATA RACE
Read at 0x00c000116078 by goroutine 108:
main.main.func2()
main.go:30 +0x2c
Previous write at 0x00c000116078 by goroutine 109:
main.main.func2()
main.go:30 +0x3c
==================
注意一个反直觉的事实：这次 naive 的最终值碰巧是 100，看起来「没问题」。竞态的危害正在于此——它可能几万次运行都正常，然后在某个高负载的夜里丢一次更新。「跑起来是对的」不能证明没有竞态，只有 -race 能。
CompareAndSwap（CAS）是原子包里最有表达力的操作，常用于「只在没人改过时才更新」。实测：n 存 5 时 CompareAndSwap(5, 10) 返回 true 且值变成 10；紧接着 CompareAndSwap(5, 20) 返回 false，值仍是 10——因为当前值已经不是 5 了。atomic.Bool 则适合做「停止标志」，s.stopped.Load() 判断、Store(true) 置位，都是原子的。
原子操作的边界很清楚：它只能保护单个变量，无法保证多个变量之间的一致性。如果「更新 A 的同时必须更新 B」是一个不变量，那就必须用锁，因为两次原子操作之间仍然可以被其他 goroutine 插入。
11.1.7 四种同步手段怎么选
| 手段 | 适用场景 | 不适用 |
|---|---|---|
sync.Mutex | 保护一段逻辑或一组数据；读写都频繁 | 纯读多写少（浪费并发） |
sync.RWMutex | 读远多于写，且临界区短 | 写很频繁（写锁会饿死读者） |
sync/atomic | 单个计数、标志位、指针 | 需要多变量一致性的场景 |
| channel | 传递数据所有权、控制流程、做队列 | 单纯保护一个字段（杀鸡用牛刀） |
一句话总结 Go 社区的口诀：不要通过共享内存来通信，而要通过通信来共享内存。但这不意味着「channel 永远优于锁」——保护一个结构体内部字段，锁更直接、更快、更好读。channel 的价值在于「传递」而不是「保护」。
11.1.8 TaskAPI 的并发安全 MemStore
把本节内容组装成 TaskAPI 的正式版本：
type MemStore struct {
mu sync.RWMutex
tasks map[int64]Task
stats Stats
}
type Stats struct {
reads atomic.Int64
writes atomic.Int64
}
func NewMemStore() *MemStore {
return &MemStore{tasks: make(map[int64]Task)}
}
func (s *MemStore) Save(t Task) {
s.mu.Lock()
defer s.mu.Unlock()
s.tasks[t.ID] = t
s.stats.writes.Add(1)
}
func (s *MemStore) Get(id int64) (Task, bool) {
s.mu.RLock()
defer s.mu.RUnlock()
t, ok := s.tasks[id]
s.stats.reads.Add(1)
return t, ok
}
func (s *MemStore) Count() int {
s.mu.RLock()
defer s.mu.RUnlock()
return len(s.tasks)
}
几点说明：
tasks由mu保护，stats由 atomic 保护，两类数据用两种手段，各司其职。NewMemStore必须初始化 map。make(map[int64]Task)这行不能省，否则第一次Save就是「向 nil map 写入」，直接 panic。- 所有导出方法都用指针接收者，避免复制 Mutex。
stats.reads.Add(1)放在 RLock 内部还是外部都行（atomic 自身是安全的），放在内部语义更清楚。
这份 store 现在可以被任意多个 goroutine 并发使用，这是第 13 章接 HTTP 处理器、第 14 章接数据库之前必须打好的一层地基。
11.1.9 小结与练习
- 数据竞态的两个条件：多 goroutine 访问同一内存 + 至少一个写 + 无同步。
sync.Mutex保护临界区，defer Unlock()是保命写法。- 含锁的结构体只能用指针接收者，
go vet会帮你抓值接收者的错误。 sync.RWMutex在读多写少时有优势，但读锁有计数开销，写锁有优先级。sync/atomic只保护单个变量；多变量不变量必须用锁。- 临界区里不要放 IO 和 sleep。
练习：
- 把 11.1.1 的并发 map 例子改成用
sync.Mutex保护，用-race验证通过。 - 给
MemStore加一个List() []Task方法，用 RLock 实现，并思考「返回切片是否安全」（提示：切片底层数组可能被后续写操作影响）。 - 故意把
Inc的接收者写成值类型，运行go vet ./...看提示信息，然后改回来。 - 用
atomic.Int64给MemStore加一个「当前并发调用数」的瞬时指标，观察它在压测时的变化。
下一节补齐另外三个常用同步原语：WaitGroup 收口 goroutine、Once 做一次性初始化、sync.Map 做只增不减的缓存。
阅读导航：上一节：10.3 worker pool 与 pipeline · 下一节：11.2 WaitGroup/Once/sync.Map 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。