一、Mutex 锁

1、mutex 数据机构

type Mutex struct {    state int32 //高29位，标识当前挂起协程数（2^29-1个），后三位分别表示是否为饥饿模式、是否有协程抢锁和是否已上锁    sema  uint32 //协程挂起/唤醒信号量}

2、上锁原理

通过 cas 原子操作，改变 state 上锁标记位，成功把 state 上锁标记位从 0 改为 1，则上锁成功，否则上锁失败，陷入自旋或者阻塞

3、锁升级

两种锁竞争方案对比

• 自旋：主动竞争锁，类似轮询方案，操作比较轻，无须上下文切换，但可能长时间轮询不得锁，浪费 cpu 时间片，适合锁竞争不激烈的场景

• 阻塞等待：精准唤醒，类似事件驱动，需挂起协程，有上下文切换，操作相对比较重，适合竞争比较激烈的场景

锁竞争升级过程

当 goroutine 首次抢锁不得时，会进行 4 次自旋（每次自旋都是尝试抢锁的过程），经历自旋仍不得锁后，会被阻塞挂起。（放进阻塞队列队尾，等待其他 goroutine 解锁唤醒）

4、正常模式 VS 饥饿模式

正常模式：各等待锁的 goroutine 正常抢锁，一般正在占有时间片（处于自旋的 goroutine 容易取得锁）

饥饿模式：阻塞队列中等待时间长的 goroutine（等待大于 1ms）优先取得锁，依次从队头遍历到队尾

正常模式 -> 饥饿模式的切换条件：阻塞队列中存在等锁大于 1ms 的协程

饥饿模式 -> 正常模式的切换条件：阻塞队列为空或者当前阻塞队列等锁时间少于 1ms

二、RwMutex 锁

1、实现原理：

• 读和写的互斥操作，通过 cas 原子操作 readerCount - rwmutexMaxReaders(一个常量：2^29，代表共享读锁的最大值)实现

• 写和写的互斥，是通过 sync.mutex 实现的

• 获取读锁成功时，readerCount 会+1，获取写锁时 readerCount 会一次性减掉 rwmutexMaxReaders，并将此时的 readerCount 加到 readerWait 中，如果 readerWait>0，证明前面还有读锁没释放，则此时

2、RLock 流程

func (rw *RWMutex) RLock() {    if atomic.AddInt32(&rw.readerCount, 1) < 0 {        runtime_SemacquireMutex(&rw.readerSem, false, 0) //读阻塞队列    }}

• 将 readerCount+1（原子操作）

• 若 readerCount > 0 证明此前无写锁接入，获取读锁成功

• 若 readerCount < 0 证明此前有写锁接入（加写锁会一次减掉 rwmutexMaxReaders），获取读锁失败，加到读锁等待队列

3、RUnlock 流程

func (rw *RWMutex) RUnlock() {    if r := atomic.AddInt32(&rw.readerCount, -1); r < 0 {        rw.rUnlockSlow(r)    }}
func (rw *RWMutex) rUnlockSlow(r int32) {    if r+1 == 0 || r+1 == -rwmutexMaxReaders {        fatal("sync: RUnlock of unlocked RWMutex")    }    if atomic.AddInt32(&rw.readerWait, -1) == 0 {        runtime_Semrelease(&rw.writerSem, false, 1) //唤醒写协程    }}

• 将 readerCount-1（原子操作）

• 若 readerCount > 0 证明还有其他读锁没释放，直接返回就行

• 若 readerCount < 0 进入 rUnlockSlow 流程

• rUnlockSlow 流程有以下情况：

1. readerCount - 1 再 + 1 == 0 或者 readerCount - 1 再 + 1 == -rwmutexMaxReaders，证明解锁前 readerCount 都是=0 的，readerCount=0 意味根本没加读锁，何来释放？所以直接 fatal error

2. 既然 readerCount 为负数，证明有写锁进入了阻塞，再判断 rw.readerWait - 1 是否为 0，为 0 则表示当前为持有读锁的最后一个 goroutine，有义务把写阻塞队列的协程唤醒

4、Lock 流程（写锁）

func (rw *RWMutex) Lock() {    rw.w.Lock()    r := atomic.AddInt32(&rw.readerCount, -rwmutexMaxReaders) + rwmutexMaxReaders    if r != 0 && atomic.AddInt32(&rw.readerWait, r) != 0 {        runtime_SemacquireMutex(&rw.writerSem, false, 0)    }}

• 先通过内置的互斥锁 mutex 去进行多个写操作的竞争互斥

• 获取到 mutex 锁的 goroutine 对 readerCount 进行一次性扣减 rwmutexMaxReaders（&地址方式操作，readerCount 的值直接变成负数）

• 如果当前 r 为 0，则证明前面无读锁，写操作获得锁

• 如果 r 不为 0，证明前面有 r 个协程持有读锁，并将 r 加到 readerWait 变量中，然后把自己放到写阻塞队列

5、UnLock 流程（写锁）

func (rw *RWMutex) Unlock() {    r := atomic.AddInt32(&rw.readerCount, rwmutexMaxReaders)    if r >= rwmutexMaxReaders {        fatal("sync: Unlock of unlocked RWMutex")    }    for i := 0; i < int(r); i++ {        runtime_Semrelease(&rw.readerSem, false, 0)    }    rw.w.Unlock()}

• 将 readerCount + rwmutexMaxReaders

• 如果结果 r 比 rwmutexMaxReaders 还大，则说明要么没加过写锁（即没进行过-rwmutexMaxReaders 操作）或者超出读锁最大上限，直接 fatal error

• 如果结果 r > 0，即代表有 r 个 goroutine 在读阻塞队列，需要当前协程唤醒，唤醒后再释放 mutex 锁

• 如果 r == 0， 证明写锁前，没读书被阻塞，直接释放 mutex 锁

6、读写锁流程串联

• T1 时刻，有 3 个 goroutine 获得读锁（调用 RLock 方法）

• T2 时刻，goroutine4 尝试获取写锁（调用 Lock 方法），因为读锁没释放，goroutine4 进入阻塞队列

• T3 时刻，goroutine5 尝试获取读锁（调用 RLock 方法），因为前面有写锁尝试获取，goroutine5 进入了读阻塞队列；goroutine6 尝试获取写锁（调用 Lock 方法），因为前面有写锁加了 sync.mutex.lock()，goroutine6 进入了互斥锁的等待队列

• T4 时刻，goroutine1、goroutine2、goroutine3 释放了 RW 锁，goroutine3 唤醒 goroutine4，此时 goroutine4 获得了 RW 锁

• T5 时刻，goroutine4 释放了锁，并唤醒 goroutine5，此时 goroutine5 获得了读锁。接着 goroutine4 释放 sync.mutex.unlock()，唤醒了 goroutine6，因为 goroutine5 取得了读锁，goroutine6 再次进入了 RW 锁的阻塞队列

7、一些结论

• 读锁更容易优先获得，一个是因为写锁要经历 mutex 和 RW 锁的竞争，另外写锁释放是先唤醒读锁的协程后再释放互斥锁 mutex，此时唤醒读协程肯定比后释放的写协程更快获得 RW 锁

• 在写协程尝试获取 RW 锁时，发现前面有读协程占有了读锁，即使写协程没加锁成功进入了阻塞，但是后面再到来的读锁也会进入阻塞，因为只要写锁操作过-rwmutexMaxReaders，readercount 就是一个负数，即读操作无法获得读锁