【Go基础篇】 彻底搞懂 Mutex 实现原理

文章有点长，但是我希望你能看到最后，面试必考题！

上一讲我们一起体验了 Mutex 的使用，竟是那么简单，只有简简单单两个方法，Lock 和 Unlock，进入临界区之前调用 Lock 方法，退出临界区的时候调用 Unlock 方法。这个时候，你一定会有一丝好奇：“它的实现是不是也很简单呢？”

上篇文章我们讲述了 Mutex 是怎么使用的，回顾一下其实很简单，只有 Lock 和 Unlock，在进入临界区调用 Lock 方法，退出使用 unlock 方法。有人会问：想看下底层源码实现是不是很简单的？今天带大家看看源码实现，互相学习。同时也会涉及互斥锁、信号量等并发编程中的常见概念。

1、实现原理

其实 Mutex 到现在已经优化了很多版本了，总结了一下演进的过程可分为四个阶段，分别是：

• 初版 Mutex：使用一个 flag 变量表示锁是否被持有；
• 给新人机会：照顾新来的 goroutine 先获取到锁；
• 多给些机会：照顾新来的和被唤醒的 goroutine 获取到锁；
• 解决饥饿：存在竞争关系，有饥饿情况发生，需要解决。

1.1 初版 Mutex

设置一个 flag 变量，当锁被某个 goroutine 持有，则标记，当 flag 为1时，就认为锁已被 goroutine 持有，其他的竞争的 goroutine 只能等待。到 flag 的值是0，通过 CAS 将这个 flag 设置为 1，此时就被当前这个 goroutine 获取到锁了。

这是 2008 年 Russ Cox 提交的第一版 Mutex 源码，很简洁！

简单说下 CAS，全称（compare-and-swap，或者 compare-and-set）。CAS是指将给定的值和内存地址中的值比较，如果相同就用新值替换内存地址中的值，这个过程是原子性的。所谓原子性，就是操作的值总是最新的，不会出现其他线程同时修改了这个值，若有，CAS 会返回失败。这个在操作系统中也有实现，有兴趣的可以搜索相关资料学习一下。

上述代码中最核心的 Mutex 结构体中包含着两个字段：

• key：用于标识是否被某个 goroutine 占用，若果大于1，则表示已被占用；
• sema：一个信号的变量，用于控制等待 goroutine 的阻塞与控制 goroutine。

用一张图表达一下文字，看看能不能加深大家的印象：

在使用 Mutex 的时候，需要严格遵循 “谁申请，谁释放” 原则。

在代码第24行，调用 Lock 请求获取锁时，我们通过 xadd 进行 CAS 操作，通过循环执行 CAS 一直到成功，保证 key + 1 操作完成，当锁没有被任何 goroutine 占用，Lock 方法就会被改变为1 ，此时这个 goroutine 就拥有了这个锁；若锁已经被其他的 goroutine 占用了，那么当前这个 goroutine 会把 key + 1，并且会调用 semacquire 方法（line：27行），然后使用信号量设置自己休眠，等待锁释放的时候，信号量就会将它再次唤醒。

当前拿到锁的 goroutine 在进行 Unlock 释放锁操作时，会把 Key - 1（line：31）。若此时没有其他的 goroutine 在等待锁，直接返回了。如果还有其他的 goroutine 在等待锁，那么会调用 semrelease 方法（line：34），然后用信号量唤醒其他等待着的 goroutine。

所以我们总计一下，在初版的 Mutex 是利用 CAS 进行原子操作设置 key 的。key 标记锁被 goroutine 占用的同时也记录了当前获取锁和等待锁的 goroutine 数量，设计非常简洁。

我们会发现，Unlock 方法可以被任意的 goroutine 调用释放锁，即使是没持有这个互斥锁的 goroutine，也可以进行这个操作。这是因为，Mutex 本身并没有包含持有这把锁的 goroutine 的信息，所以，Unlock 也不会对此进行检查。Mutex 的这个设计在现在也一直在用。所以一定记住：

在使用 Mutex 的时候，需要严格遵循 “谁申请，谁释放” 原则。

1.2 给新人机会

在 2011 年 6 月 30 日的 commit 中对 Mutex 进行了一次大的调整，如下：

对比一下上面代码，会发现 key 改成了 state，自然代表的意思也不一样了。

state 一个字段多个意义，方便我们利用较少的内存实现互斥锁。state 由三个部分组成，分别是：

• 第一位：锁是否被占有；
• 第二位：是否有被唤醒的 goroutine；
• 其他位：是等待此锁的 goroutine 数量。

获取锁的方法 Lock 也得复杂了，我们一起看下：

首先看下第 3 行 Fast path 的解释，当没有 goroutine 占有锁且没有其他的 goroutine，直接获得锁。

相反，state 不为 0，进行循环检查，若当前的 goroutine 没有获取到锁，就会进行休眠等待，当锁释放被唤醒后，需要和正在请求锁的 goroutine 进行竞争获得锁。代码中的第 10 行将当前的 flag 设置为加锁状态，如果能成功地通过 CAS 把这个新值赋予 state（line: 19 和 20 ），就代表抢夺锁的操作成功了。

此时需要注意，若成功的改变了 state 值，但是之前的 state 是有锁的状态，那么 state 只是清除 mutexWoken 标志或者增加一个 waiter 而已。

请求锁的 goroutine 有两类，

• 一类是新来请求锁的 goroutine；
• 另一类是被唤醒的等待请求锁的 goroutine。

锁的状态也有两种：加锁和未加锁。我用一张表格，来说明一下 goroutine 不同来源不同状态下的处理逻辑。

接下来我们看下释放锁 Unlock。

其中，第 3 行是通过减 1 将占有锁的标识状态设置成未加锁，第 4 到 6 行会检测原来锁的状态是否是未加锁，如果是直接回抛 pannic。

其中，第 3 行是通过减 1 将占有锁的标识状态设置成未加锁，第 4 到 6 行会检测原来锁的状态是否是未加锁，如果是直接回抛 pannic。

将加锁的释放，还需要一些额外的才做，因为可能还会存在一些等待这个锁的 goroutine（也称为 waiter）需要通过信号量的方式唤醒其中的一个，会出现一下两种情况

• 当没有其他 goroutine （waiter），此时对锁竞争的 goroutine 只有一个，Unlock 后就可以直接返回了，当有唤醒的 goroutine，或者被别人加了锁，此时当前的 goroutine Unlock 后也可以直接返回
• 若有等待的 goroutine 并且没有唤醒的 waiter，那么就需要唤醒一个等待的 waiter。在这之前，我们需要将 waiter 的数量减1，并且标记 mutexWoken，这样 Unlock 就可以返回了。

通过这样复杂的检查、判断、设置，我们就可以安全地将互斥锁 Unlock 了。

这一次的改动总结一下，就是新来的 goroutine 也有机会获取到锁，甚至一个 goroutine 会连续获得，和之前的设计不太一样，从代码复杂度就能看的出来。这一版需要竞争获得 Mutex。

没有空闲的锁或者竞争失败才加入到等待队列中。但是其实还可以进一步优化。我们接着往下看。

1.3 多给些机会

在 2015 年 2 月版本提交的改动中，如果新来的 goroutine 或者被唤醒的 goroutine 获取不到锁，那么会通过自旋等待的方式，超过一定的自旋次数后，再执行原来的逻辑。

这次的优化，增加了第 13 行到 21 行、第 25 行到第 27 行以及第 36 行。其中第 13 行到 21 行。

如果可以自旋的话，在（line:9）的 for 循环会重新检查锁是否 Unlock。对于临界区代码执行非常短的场景来说，这是一个非常高效的一次优化。因为临界区的代码耗时很短，锁很快就能释放，而抢夺锁的 goroutine 不用通过休眠唤醒方式等待调度，只需要自旋转几次，就能获得锁。

1.4 解决饥饿

通过上面几次的优化迭代，Mutex 的底层源码越来越复杂，在高并发场景下获取锁也更加的公平竞争。但是细细回味，会发现新来的 goroutine 也参与到了竞争，会导致每次新来的 goroutine 能够拿到，在极端的情况，会出现等待中的 goroutine 一直在等待，拿不到锁。

之前的版本中也有这样的困境，goroutine 总是拿不到。

当然 Go 聪明的开发者不能允许这种事情发生，2016 年 Go 1.9 中 Mutex 增加了饥饿模式，让竞争锁变得更加公平，等待时间限制在 1ms，也修复了一个历史大 Bug。

总是把唤醒的 goroutine 放到等待队列的末尾，会导致增加不公平的等待时间。

之后，2018 年，Go 开发者将 fast path 和 slow path 拆成独立的方法，以便内联，提高性能。2019 年也有一个 Mutex 的优化，虽然没有对 Mutex 做修改，但是，对于 Mutex 唤醒后持有锁的那个 waiter，调度器可以有更高的优先级去执行，这已经是很细致的性能优化了。

图来自（http://draveness.me）

在默认情况下，互斥锁的所有状态位都是 0，int32 中的不同位分别表示了不同的状态：

• mutexLocked — 表示互斥锁的锁定状态；
• mutexWoken — 表示从正常模式被从唤醒；
• mutexStarving — 当前的互斥锁进入饥饿状态；
• waitersCount — 当前互斥锁上等待的 Goroutine 个数；

当然，你也可以暂时跳过这一段，源码有点多，以后喝茶慢慢品，但是要记住:

Mutex 绝不容忍一个 goroutine 被落下，永远没有机会获取锁。不抛弃不放弃是它的宗旨，而且它也尽可能地让等待较长的 goroutine 更有机会获取到锁。

**点击时展开查看源码，真的有点长呀！**

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 || old>>mutexWaiterShift == 0 {
                    throw("sync: inconsistent mutex state")
                }
                // 有点绕，加锁并且将waiter数减1
                delta := int32(mutexLocked - 1<<mutexWaiterShift)
                if !starving || old>>mutexWaiterShift == 1 {
                    delta -= mutexStarving // 最后一个waiter或者已经不饥饿了，清除饥饿标记
                }
                atomic.AddInt32(&m.state, delta)
                break
            }
            awoke = true
            iter = 0
        } else {
            old = m.state
        }
    }
}

func (m *Mutex) Unlock() {
    // Fast path: drop lock bit.
    new := atomic.AddInt32(&m.state, -mutexLocked)
    if new != 0 {
        m.unlockSlow(new)
    }
}

func (m *Mutex) unlockSlow(new int32) {
    if (new+mutexLocked)&mutexLocked == 0 {
        throw("sync: unlock of unlocked mutex")
    }
    if new&mutexStarving == 0 {
        old := new
        for {
            if old>>mutexWaiterShift == 0 || old&(mutexLocked|mutexWoken|mutexStarving) != 0 {
                return
            }
            new = (old - 1<<mutexWaiterShift) | mutexWoken
            if atomic.CompareAndSwapInt32(&m.state, old, new) {
                runtime_Semrelease(&m.sema, false, 1)
                return
            }
            old = m.state
        }
    } else {
        runtime_Semrelease(&m.sema, true, 1)
    }
}

1.5 饥饿模式和正常模式

Mutex 有两种操作模式

• 正常模式
• 饥饿模式

我们需要在这里先了解正常模式和饥饿模式都是什么以及它们有什么样的关系。接下来我们分析一下 Mutex 对这两种模式的处理。

在调用 Lock 方法是，当前没有竞争直接获取锁返回。否则进入了 lockSlow 方法，

正常模式下，goroutine 都是进入先入先出到等待队列中，被唤醒的 goroutine 不会直接拿到锁，而是和新来的 goroutine 就行竞争。但是新来的 goroutine 有先天的优势拿到锁，因为他们正在 CPU中运行。所以高并发下，被唤醒的 goroutine 可能拿不到锁，这是他就会被插入到队列的前面，此时如果 goroutine 获取不到锁的时间超过了设定的阈值 1 ms，那么此时 Mutex 就会进入到饥饿模式。

饥饿模式下， Mutex 的拥有者将直接把锁交给队列最前面的 goroutine，新来的 goroutine 会加入到等待队列的尾部，如果拥有锁的 goroutine 发现一下两种情况，会把 Mutex 转换成正常模式：

• 此时新的 goroutine 的等待时间小于 1ms
• 此 goroutine 已经是队列中的最后一个了，没有其它的等待锁的 goroutine 了

对比一下，正常模式拥有更好的性能，因为即使有等待抢锁的 goroutine， 也可以连续多次获取到锁。

饥饿模式是一种平衡，他让一些等待很长的 goroutine，能够优先拿到锁。

我们已经从多个方面和历史版本分析了 Mutex 的实现原理，这里我们从 Lock 和 Unlock 个方面总结注意事项。

Mutex 的 Lock 过程比较复杂，目前使用的新版本中，它涉及自旋、信号量以及调度等概念：

• 如果 Mutex 处于初始化状态，会通过置位 mutexLocked 加锁；
• 如果 Mutex 处于 mutexLocked 状态并且在正常模式下工作，会进入自旋；
• 如果当前 goroutine 等待锁的时间超过了 1ms，Mutex 就会切换到饥饿模式；
• Mutex 在正常情况下会将尝试获取锁的 goroutine 切换至休眠状态，等待锁的持有者唤醒；
• 如果当前 goroutine 是 Mutex 上的最后一个等待的协程或者等待的时间小于 1ms，那么它会将 Mutex 切换回正常模式；

Mutex 的 Unlock 过程与之相比就比较简单，其代码行数不多、逻辑清晰，也比较容易理解：

• 当 Mutex 已经被解锁时，调用 Unlock 会直接抛出异常；
• 当 Mutex 处于饥饿模式时，将锁的所有权交给队列中的下一个等待者，等待者会负责设置 mutexLocked 标志位；
• 当 Mutex 处于正常模式时，如果没有 goroutine 等待锁的释放或者已经有被唤醒的 goroutine 获得了锁，会直接返回

2、总结

我们已经从多个方面和历史版本分析了 Mutex 的实现原理，这里我们从 Lock 和 Unlock 个方面总结注意事项。

Mutex 的 Lock 过程比较复杂，目前使用的新版本中，它涉及自旋、信号量以及调度等概念：

• 如果 Mutex 处于初始化状态，会通过置位 mutexLocked 加锁；
• 如果 Mutex 处于 mutexLocked 状态并且在正常模式下工作，会进入自旋；
• 如果当前 goroutine 等待锁的时间超过了 1ms，Mutex 就会切换到饥饿模式；
• Mutex 在正常情况下会将尝试获取锁的 goroutine 切换至休眠状态，等待锁的持有者唤醒；
• 如果当前 goroutine 是 Mutex 上的最后一个等待的协程或者等待的时间小于 1ms，那么它会将 Mutex 切换回正常模式；

Mutex 的 Unlock 过程与之相比就比较简单，其代码行数不多、逻辑清晰，也比较容易理解：

• 当 Mutex 已经被解锁时，调用 Unlock 会直接抛出异常；
• 当 Mutex 处于饥饿模式时，将锁的所有权交给队列中的下一个等待者，等待者会负责设置 mutexLocked 标志位；
• 当 Mutex 处于正常模式时，如果没有 goroutine 等待锁的释放或者已经有被唤醒的 goroutine 获得了锁，会直接返回

3、思考问题

Q：目前 Mutex 的 state 字段有几个意义，这几个意义分别是由哪些字段表示的？

A：state 字段一共有四个子字段，前三个 bit 是 mutexLocked（锁标记）、mutexWoken（唤醒标记）、mutexStarving（饥饿标记），剩余 bit 标示 mutexWaiter（等待数量）。