Go 并发编程之 Mutex

我们比较常见的大型项目的设计中都会出现并发访问问题，并发就是为了解决数据的准确性，保证同一个临界区的数据只能被一个线程进行操作，日常中使用到的并发场景也是很多的： 计数器：计数器结果不准确； 秒杀系统：由于同一时间访问量比较大，导致的超卖； 用户账户异常：同一时间支付导致的账户透支； buffer 数据异常：更新 buffer 导致的数据混乱。 上面都是并发带来的数据准确性的问题，决绝方案就是使用互斥锁，也就是今天并发编程中的所要描述的 Mutex 并发原语。 实现机制 互斥锁 Mutex 就是为了避免并发竞争建立的并发控制机制，其中有个“临界区”的概念。 在并发编程过程中，如果程序中一部分资源或者变量会被并发访问或者修改，为了避免并发访问导致数据的不准确，这部分程序需要率先被保护起来，之后操作，操作结束后去除保护，这部分被保护的程序就叫做临界区。 使用互斥锁，限定临界区只能同时由一个线程持有，若是临界区此时被一个线程持有，那么其他线程想进入到这个临界区的时候，就会失败或者等待释放锁，持有此临界区的线程退出，其他线程才有机会获得这个临界区。 Mutex 是 Go 语言中使用最广泛的同步原语，也称为并发原语，解决的是并发读写共享资源，避免出现数据竞争 data race 问题。 基本使用 互斥锁 Mutex 提供了两个方法 Lock 和 Unlock：进入到临界区使用 Lock 方法加锁，退出临界区使用 Unlock 方法释放锁。 1 2 3 4 5 6 7 type Locker interface { Lock() Unlock() } func(m *Mutex)Lock() func(m *Mutex)Unlock() 当一个 goroutine 调用 Lock 方法获取到锁后，其他 goroutine 会阻塞在 Lock 的调用上，直到当前获取到锁的 goroutine 释放锁。 接下来是一个计数器的例子，是由 100 个 goroutine 对计数器进行累加操作，最后输出结果： 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 package main import ( "fmt" "sync" ) func main() { var mu sync.Mutex countNum := 0 // 确认辅助变量是否都执行完成 var wg sync.WaitGroup // wg 添加数目要和 创建的协程数量保持一致 wg.Add(100) for i := 0; i < 100; i++ { go func() { defer wg.Done() for j := 0; j < 1000; j++ { mu.Lock() countNum++ mu.Unlock() } }() } wg.Wait() fmt.Printf("countNum: %d", countNum) } 实际使用 很多时候 Mutex 并不是单独使用的，而是嵌套在 Struct 中使用，作为结构体的一部分，如果嵌入的 struct 有多个字段，我们一般会把 Mutex 放在要控制的字段上面，然后使用空格把字段分隔开来。 甚至可以把获取锁、释放锁、计数加一的逻辑封装成一个方法。 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 package main import ( "fmt" "sync" ) // 线程安全的计数器 type Counter struct { CounterType int Name string mu sync.Mutex count uint64 } // 加一方法 func (c *Counter) Incr() { c.mu.Lock() defer c.mu.Unlock() c.count++ } // 取数值方法 线程也需要受保护 func (c *Counter) Count() uint64 { c.mu.Lock() defer c.mu.Unlock() return c.count } func main() { // 定义一个计数器 var counter Counter var wg sync.WaitGroup wg.Add(100) for i := 0; i < 100; i++ { go func() { defer wg.Done() for j := 0; j < 1000; j++ { counter.Incr() } }() } wg.Wait() fmt.Printf("%d\n", counter.Count()) } 思考问题 Q：你已经知道，如果 Mutex 已经被一个 goroutine 获取了锁，其它等待中的 goroutine 们只能一直等待。那么，等这个锁释放后，等待中的 goroutine 中哪一个会优先获取 Mutex 呢？ A：FIFO，先来先服务的策略，Go 的 goroutine 调度中，会维护一个保障 goroutine 运行的队列，当获取到锁的 goroutine 执行完临界区的操作的时候，就会释放锁，在队列中排在第一位置的 goroutine 会拿到锁进行临界区的操作。 实现原理 Mutex 的架构演进目前分为四个阶段： 初版 Mutex：使用一个 flag 变量表示锁?是否被持有； 给新人机会：照顾新来的 goroutine 先获取到锁； 多给些机会：照顾新来的和被唤醒的 goroutine 获取到锁； 解决饥饿：存在竞争关系，有饥饿情况发生，需要解决。 初版 Mutex 1 2 3 4 5 // 互斥锁的结构，包含两个字段 type Mutex struct { key int32 // 锁是否被持有的标识 sema int32 // 信号量专用，用以阻塞/唤醒goroutine } Unlock 方法可以被任意的 goroutine 调用释放锁，即使是没持有这个互斥锁的 goroutine，也可以进行这个操作。这是因为，Mutex 本身并没有包含持有这把锁的 goroutine 的信息，所以，Unlock 也不会对此进行检查。Mutex 的这个设计一直保持至今。 在使用 Mutex 的时候，需要严格遵循 “谁申请，谁释放” 原则。 解决饥饿 由于使用了给新人机会，又肯呢个会出现每次都会被新来的 goroutine 获取到锁，导致等待的 goroutine 一直获取不到锁，造成饥饿问题。 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 type Mutex struct { state int32 sema uint32 } const ( mutexLocked = 1 << iota // mutex is locked mutexWoken mutexStarving // 从state字段中分出一个饥饿标记 mutexWaiterShift = iota starvationThresholdNs = 1e6 ) func (m *Mutex) Lock() { // Fast path: 幸运之路，一下就获取到了锁 if atomic.CompareAndSwapInt32(&m.state, 0, mutexLocked) { return } // Slow path：缓慢之路，尝试自旋竞争或饥饿状态下饥饿goroutine竞争 m.lockSlow() } func (m *Mutex) lockSlow() { var waitStartTime int64 starving := false // 此goroutine的饥饿标记 awoke := false // 唤醒标记 iter := 0 // 自旋次数 old := m.state // 当前的锁的状态 for { // 锁是非饥饿状态，锁还没被释放，尝试自旋 if old&(mutexLocked|mutexStarving) == mutexLocked && runtime_canSpin(iter) { if !awoke && old&mutexWoken == 0 && old>>mutexWaiterShift != 0 && atomic.CompareAndSwapInt32(&m.state, old, old|mutexWoken) { awoke = true } runtime_doSpin() iter++ old = m.state // 再次获取锁的状态，之后会检查是否锁被释放了 continue } new := old if old&mutexStarving == 0 { new |= mutexLocked // 非饥饿状态，加锁 } if old&(mutexLocked|mutexStarving) != 0 { new += 1 << mutexWaiterShift // waiter数量加1 } if starving && old&mutexLocked != 0 { new |= mutexStarving // 设置饥饿状态 } if awoke { if new&mutexWoken == 0 { throw("sync: inconsistent mutex state") } new &^= mutexWoken // 新状态清除唤醒标记 } // 成功设置新状态 if atomic.CompareAndSwapInt32(&m.state, old, new) { // 原来锁的状态已释放，并且不是饥饿状态，正常请求到了锁，返回 if old&(mutexLocked|mutexStarving) == 0 { break // locked the mutex with CAS } // 处理饥饿状态 // 如果以前就在队列里面，加入到队列头 queueLifo := waitStartTime != 0 if waitStartTime == 0 { waitStartTime = runtime_nanotime() } // 阻塞等待 runtime_SemacquireMutex(&m.sema, queueLifo, 1) // 唤醒之后检查锁是否应该处于饥饿状态 starving = starving || runtime_nanotime()-waitStartTime > starvationThresholdNs old = m.state // 如果锁已经处于饥饿状态，直接抢到锁，返回 if old&mutexStarving != 0 { if old&(mutexLocked|mutexWoken) != 0 || o