Go 锁与并发安全详解（源码级解读）

2022-02-11T10:00:00+08:00 | 7分钟阅读 | 更新于 2022-02-11T10:00:00+08:00

@ MoonlightMaze

学习目标

读完本文，你应该能够：

手写 Ticket Lock（票号锁） 理解公平锁原理
实现一个读写锁，理解读者优先与写者优先的取舍
使用 atomic 实现无锁计数器和无锁队列
深入理解 sync.Mutex 的饥饿模式切换机制
分析 sync.RWMutex 的读锁批量释放优化
实现 sync.Map 的简化版，理解双 map 设计
解决生产者-消费者、读者-写者、哲学家进餐等经典问题
理解 Singleflight 避免重复请求的原理

一、Ticket Lock（票号锁）

原理

票号锁是最公平的锁实现之一，通过维护两个计数器：

ticket: 下一个获得的票号
nowServing: 当前正在服务的票号

应用场景

在实际项目中，这个知识点经常应用于：

典型场景一：说明具体应用场景
典型场景二：说明另一个应用场景
面试要点：提醒面试中可能被问到的点

type TicketLock struct {
	ticket   uint32
	nowServing uint32
	mu       [2]uint64 // 对齐到不同的缓存行
}

func (t *TicketLock) Lock() {
	myTurn := atomic.AddUint32(&t.ticket, 1) - 1
	for atomic.LoadUint32(&t.nowServing) != myTurn {
		runtime.Gosched() // 让出 CPU
	}
}

func (t *TicketLock) Unlock() {
	atomic.AddUint32(&t.nowServing, 1)
}

复习提示：

与 Mutex 的区别：

Ticket Lock 是严格公平的，先到先得
Go 的 Mutex 在竞争低时是不公平的（快速路径）
当出现饥饿时，Mutex 会切换到公平模式

二、读写锁实现

2.1 读者优先版

type ReadWriteLock struct {
	mu        sync.Mutex
	readers   int
	waiting   int
	readsDone chan struct{}
}

func (rw *ReadWriteLock) RLock() {
	rw.mu.Lock()
	rw.readers++
	if rw.readers == 1 {
		rw.waiting++
	}
	rw.mu.Unlock()
	
	if rw.readers == 1 {
		<-rw.readsDone
	}
}

func (rw *ReadWriteLock) RUnlock() {
	rw.mu.Lock()
	rw.readers--
	if rw.readers == 0 {
		rw.waiting--
		close(rw.readsDone)
		rw.readsDone = make(chan struct{})
	}
	rw.mu.Unlock()
}

func (rw *ReadWriteLock) Lock() {
	rw.mu.Lock()
	rw.waiting++
	for rw.readers > 0 {
		rw.readsDone = make(chan struct{})
		rw.mu.Unlock()
		<-rw.readsDone
		rw.mu.Lock()
	}
	rw.mu.Unlock()
}

func (rw *ReadWriteLock) Unlock() {
	rw.mu.Lock()
	rw.waiting--
	close(rw.readsDone)
	rw.mu.Unlock()
}

2.2 写者优先版

写者优先的关键是在 RLock 时检查是否有等待的 writer：

func (rw *ReadWriteLock) RLock() {
	rw.mu.Lock()
	for 

### 应用场景

在实际项目中，这个知识点经常应用于：

1. **典型场景一**：说明具体应用场景
2. **典型场景二**：说明另一个应用场景
3. **面试要点**：提醒面试中可能被问到的点

rw.waiting > 0 {
		rw.mu.Unlock()
		runtime.Gosched()
		rw.mu.Lock()
	}
	rw.readers++
	rw.mu.Unlock()
}

三、atomic 无锁实现

3.1 无锁计数器

type AtomicCounter struct {
	value int64
}

func (ac *AtomicCounter) Increment() int64 {
	return atomic.AddInt64(&ac.value, 1)
}

func (ac *AtomicCounter) Decrement() int64 {
	return atomic.AddInt64(&ac.value, -1)
}

func (ac *AtomicCounter) Value() int64 {
	return atomic.LoadInt64(&ac.value)
}

3.2 CAS 循环实现



### 应用场景

在实际项目中，这个知识点经常应用于：

1. **典型场景一**：说明具体应用场景
2. **典型场景二**：说明另一个应用场景
3. **面试要点**：提醒面试中可能被问到的点

func (ac *AtomicCounter) Add(delta int64) int64 {
	for {
		old := atomic.LoadInt64(&ac.value)
		new := old + delta
		if atomic.CompareAndSwapInt64(&ac.value, old, new) {
			return new
		}
		// CAS 失败重试
	}
}

四、sync.Mutex 源码分析

4.1 结构体

type Mutex struct {
	state int32
	sema  uint32
}

// state 的低三位：
// bit 0: mutexLocked 锁定状态
// bit 1: mutexWoken 唤醒状态  
// bit 2: mutexStarving 饥饿状态
// 高29位：等待者数量

4.2 快速路径

func (m *Mutex) Lock() {
	// 快速路径：无竞争时几乎无开销
	if atomic.CompareAndSwapInt32(&m.state, 0, mutexLocked) {
		return
	}
	// 有竞争时走慢路径
	m.lockSlow()
}

4.3 饥饿模式切换

func (m *Mutex) lockSlow() {
	var waitStartTime int64
	stolen := false
	
	for {
		old := m.state
		new := old | mutexLocked
		
		if old&(mutexLocked|mutexStarving) == mutexLocked {
			// 锁定且非饥饿：唤醒等待者
			if old&mutexWoken == 0 {
				new |= mutexWoken
			}
			if old&mutexStarving == 0 {
				waitStartTime = monotime()
			}
			new++
		} else {
			// 未锁定或非饥饿：尝试获取
			if old&(mutexLocked|mutexStarving) == 0 {
				st

### 应用场景

在实际项目中，这个知识点经常应用于：

1. **典型场景一**：说明具体应用场景
2. **典型场景二**：说明另一个应用场景
3. **面试要点**：提醒面试中可能被问到的点

olen = true
			}
			new = mutexLocked
		}
		
		if atomic.CompareAndSwapInt32(&m.state, old, new) {
			if stolen {
				// 偷取锁成功，恢复正常模式
				runtime_Semrelease(&m.sema, false, 1)
				return
			}
			if old&mutexStarving == 0 {
				// 非饥饿模式：自旋
				runtime_SemacquireMutex(&m.sema, true, 8)
				return
			}
		}
	}
}

sync.Mutex：正常模式 ↔ 饥饿模式切换

<rect x="580" y="110" width="360" height="170" rx="12" stroke-dasharray="7 4"/>
<text x="760" y="145" text-anchor="middle" fill="#B03030" font-size="15" font-weight="700" stroke="none">饥饿模式 Starving</text>
<text x="760" y="178" text-anchor="middle" fill="#1a1a1a" font-size="12.5" stroke="none">锁直接交给队首等待者</text>
<text x="760" y="202" text-anchor="middle" fill="#1a1a1a" font-size="12.5" stroke="none">新来的 G 入队尾，不抢</text>
<text x="760" y="230" text-anchor="middle" fill="#B03030" font-size="12" stroke="none">严格 FIFO，保下限延迟</text>

等待 > 1ms 触发切换队列空或等待 < 1ms 切回高竞争：自旋抢锁FIFO：队首拿锁，新 G 排队

看图要点：面试必问「为什么 Go Mutex 要设计饥饿模式」。正常模式让新来的 G 自旋抢锁，吞吐高但等待者可能长期拿不到（长尾）；当某等待者等超过 1ms 就切饥饿模式，锁严格按 FIFO 交给队首，新 G 只能排队。饥饿模式退出条件：等待队列空，或队首等待者等待不足 1ms。双模式兼顾吞吐与公平。

面试官

什么时候 Mutex 会从正常模式切换到饥饿模式？

面试官

候选人

当一个 goroutine 等待锁的时间超过 1ms 时，Mutex 会切换到饥饿模式。在饥饿模式下：

新来的 goroutine 不会尝试获取锁，直接进入等待队列尾部
锁直接交给队列头部的等待者
一旦所有等待者都被服务，或者等待时间不足 1ms，会切换回正常模式

这种设计避免了长尾效应，保证最坏情况下的响应时间。

五、sync.RWMutex 源码

5.1 结构体

type RWMutex struct {
	w           Mutex          // 写锁
	writerSem   uint32         // 写者信号量
	readerSem   uint32         // 读者信号量
	readerCount int32          // 读者计数（含等待写入者）
	readerWait  int32          // 等待释放的读者数
}

5.2 读锁批量释放

func (rw *RWMutex) RUnlock() {
	if (rw.readerCount&readLockMask) == 0 {
		t

### 应用场景

在实际项目中，这个知识点经常应用于：

1. **典型场景一**：说明具体应用场景
2. **典型场景二**：说明另一个应用场景
3. **面试要点**：提醒面试中可能被问到的点

hrow("sync: RUnlock of unlocked RWMutex")
	}
	
	// 快速路径
	if atomic.AddInt32(&rw.readerCount, -readLockDelta) >= 0 {
		return
	}
	
	// 慢路径：可能有等待的写者
	if rw.readerCount&readLockMask == 0 {
		// 唤醒等待的写者
		runtime_Semrelease(&rw.writerSem, false, 1)
		return
	}
	
	// 还有读者在等待
	rw.lock.Lock()
	rw.readerCount -= readLockDelta
	if rw.readerCount < 0 {
		throw("sync: RUnlock of unlocked RWMutex")
	}
	rw.lock.Unlock()
}

六、经典并发问题

6.1 生产者-消费者

func producerConsumer

### 应用场景

在实际项目中，这个知识点经常应用于：

1. **典型场景一**：说明具体应用场景
2. **典型场景二**：说明另一个应用场景
3. **面试要点**：提醒面试中可能被问到的点

() {
	var buf [10]int
	var mu sync.Mutex
	cond := sync.NewCond(&mu)
	nEmpty, nFull := len(buf), 0
	
	var wg sync.WaitGroup
	wg.Add(2)
	
	go func() {
		defer wg.Done()
		for i := 0; i < 100; i++ {
			mu.Lock()
			for nFull == len(buf) {
				cond.Wait()
			}
			buf[nFull] = i
			nFull++
			mu.Unlock()
			cond.Broadcast()
		}
	}()
	
	go func() {
		defer wg.Done()
		for i := 0; i < 100; i++ {
			mu.Lock()
			for nFull == 0 {
				cond.Wait()
			}
			nFull--
			mu.Unlock()
			cond.Broadcast()
		}
	}()
	
	wg.Wait()
}

6.2 哲学家进餐

const N = 5

type Philosopher struct {
	id     int
	left   *Mutex
	right  *Mutex
	eating bool
}

func (p *Philosopher) dine() {
	for {
		// thinking...
		
		//  Pick up forks (prevent deadlock with ordering)
		first, second := p.left, p.right
		if p.id%2 == 0 {
			first, second = second, first
		}
		
		first.Lock()
		second.Lock()
		
		// eating...
		
		second.Unlock()
		first.Unlock()
	}
}

复习提示：

死锁预防：让哲学家按固定顺序拿筷子（奇数先拿左边，偶数先

应用场景

在实际项目中，这个知识点经常应用于：

典型场景一：说明具体应用场景
典型场景二：说明另一个应用场景
面试要点：提醒面试中可能被问到的点

拿右边），打破循环等待条件。

七、Singleflight 实现

Singleflight 避免相同请求的重复执行：

type call struct {
	wg  sync.WaitGroup
	val interface{}
	err error
}

type Group struct {
	mu sync.Mutex
	m  map[string]*call
}

func (g *Group) Do(key string, fn func() (interface{}, error)) (interface{}, error) {
	g.mu.Lock()
	if g.m == nil {
		g.m = make(map[string]*call)
	}
	if c, ok := g.m[key]; ok {
		g.mu.Unlock()
		c.wg.Wait()
		return c.val, c.err
	}
	c := new(call)
	c.wg.Add(1)
	g.m[key] = c
	g.mu.Unlock()
	
	c.val, c.err = fn()
	c.wg.Done()
	
	g.mu.Lock()
	delete(g.m, key)
	g.mu.Unlock()
	
	return c.val, c.err
}

八、Promise/Future 实现

type Promise struct {
	mu    sync.Mutex
	done  chan struct{}
	value interface{}
	err   error
}

func NewPromise() *Promise {
	return &Promise{
		done: make(chan struct{}),
	}
}

func (p *Promise) Resolve(value interface{}) {
	p.mu.Lock()
	p.value = value
	p.mu.Unlock()
	close(p.done)
}

func (p *Promise) Await() (interface{}, error) {
	<-p.done
	p.mu.Lock()
	defer p.mu.Unlock()
	return p.value, p.err
}

func (p *Promise) Then(fn func(interface{}) interface{}) *Promise {
	go func() {
		v, _ := p.Await()
		result := fn(v)
		// 可以继续链式调用
	}()
	return p
}

九、sync.Map 源码分析

9.1 双 Map 设计

type Map struct {
	mu Mute

### 应用场景

在实际项目中，这个知识点经常应用于：

1. **典型场景一**：说明具体应用场景
2. **典型场景二**：说明另一个应用场景
3. **面试要点**：提醒面试中可能被问到的点

x
	read atomic.Value  // readOnly
	dirty map[interface{}]*entry
	misses int
}

type readOnly struct {
	m       map[interface{}]*entry
	dirty   bool
}

9.2 懒迁移策略

read map：存放最近访问的元素
dirty map：存放所有元素（包括 expired）
当 read map 命中率为低时（misses >= threshold），将 dirty 晋升为 read，清空旧的 read

func (m *Map) Load(key interface{}) (value interface{}, ok bool) {
	read, _ := m.read.Load().(readOnly)
	e, ok := read.m[key]
	if !ok && m.dirty != nil {
		m.mu.Lock()
		// Double-check
		read, _ = m.read.Load().(readOnly)
		e, ok = read.m[key]
		if !ok && m.dirty != nil {
			e, ok = m.dirty[key]
			m.missLocked()
		}
		m.mu.Unlock()
	}
	if !ok {
		return nil, false
	}
	return e.load()
}

十、自测题

面试官

什么时候应该用 Mutex，什么时候用 atomic？

面试官

候选人

选择依据：

简单计数器/标志位：用 atomic，开销更小
复合操作（check-then-act）：用 Mutex，保证原子性
高竞争场景：atomic + CAS 通常比 Mutex 快
需要批量操作：用 Mutex 更清晰

经验法则：单个变量的读-改-写用 atomic，涉及多个变量的操作用 Mutex。

面试官

sync.Pool 适用什么场景？有什么坑？

面试官

候选人

适用场景：

高频创建销毁的临时对象（如 bufio.Scanner）
减少 GC 压力（对象复用）
多线程共享的对象池

陷阱：

对象可能被其他 goroutine 清除
不适用于有状态的对象
GC 可能会回收 pool 中的对象

十一、本章小结

复习提示：

Mutex 的饥饿模式是性能优化的关键，长时间等待会触发公平调度
RWMutex 的批量释放在读多写少场景下能显著提升吞吐
atomic 适用于简单场景，复杂操作仍需 Mutex
Singleflight 是去重利器，特别适合 cache 击穿防护
下一篇将深入讲解 数据结构实现，包括 LRU、跳表、Trie 等面试高频题

参考

《Go 并发编程实战》第 4 章：锁
Go 源码：src/sync/mutex.go, src/sync/rwmutex.go
Singleflight 实现：https://pkg.go.dev/golang.org/x/sync/singleflight

上一页 Go Goroutine 管理与调度详解（GMP 模型）

下一页 Go 数据结构手撕题详解（LRU/LFU/跳表/Trie）