go语言的锁

本篇文章主要讲锁，主要会涉及go的sync.Mutex和sync.RWMutex。

一.锁的概念和发展

1.1 锁的概念

所谓的加锁和解锁其实就是指一个数据是否被占用了，通过Mutex内的一个状态来表示。

例如，取 0 表示未加锁，1 表示已加锁；

上锁：把 0 改为 1；
解锁：把 1 置为 0.
上锁时，假若已经是 1，则上锁失败，需要等锁的主人解锁，将状态改为 0，才可以被其他锁锁上。

这就是一个锁的基本骨架，锁主要就是加锁和解锁两个状态。

并且这里要注意一个点，就是这两个操作具有原子性，不可以被拆解。

1.2 由自旋等阻塞的升级过程

一个优先的工具需要具备探测并适应环境，从而采取不同对策因地制宜的能力.

针对 goroutine 加锁时发现锁已被抢占的这种情形，此时摆在面前的策略有如下两种：

阻塞/唤醒：将当前 goroutine 阻塞挂起，直到锁被释放后，以回调的方式将阻塞 goroutine 重新唤醒，进行锁争夺；
自旋 + CAS：基于自旋结合 CAS 的方式，重复校验锁的状态并尝试获取锁，始终把主动权握在手中.

阻塞和唤醒大家肯定都知道，这里主要说一下自旋和CAS

如果看过gmp的同学对自旋肯定不陌生，所谓的自旋其实就是轮询，什么意思？这里举一个例子：

比如所谓的主动轮询，其实就是指如果加锁失败之后，它会停歇一会，然后再次询问，我可以加锁嘛？如果可以，就取到这个锁，要是不可以，就进行下一次的轮询。

和阻塞/唤醒不同，他是需要等待通知，说这把锁释放了，然后后续的goroutine才可以拿到这把锁。

CAS是什么？

CAS全称为Compare-And-Swap，是一种原子操作，用于多线程编程中实现无锁同步。

上述的方案各有各的优缺点，都有其对应的适用场景，接下来来看一下

锁竞争方案	优势	劣势	适用场景
阻塞/唤醒	精准打击，不浪费 CPU 时间片	需要挂起协程，进行上下文切换，操作较重	并发竞争激烈的场景
自旋+CAS	无需阻塞协程，短期来看操作较轻	长时间争而不得，会浪费 CPU 时间片	并发竞争强度低的场景

这里对这两种锁思想做一个介绍吧：

阻塞/唤醒：这种形式被称为是悲观锁，当G获取锁失败而阻塞时，会被挂起，标记为waiting的状态，主动让出Processor，直接让M和G结合，而P去执行其他的G（保证不会浪费这个P）锁被释放之后，才会唤醒G

自旋+CAS：这种形式被称为是乐观锁，主动权掌握在自己的手中（也就是不释放processor），会不断主动轮询尝试获取这个锁

而sync.Mutex结合了上述的两种方案，指定了一个锁升级的过程，让我们来看看吧

进行了一个怎么样的锁升级？

其实就是设计了一个状态的转化，由乐观转换为悲观，为什么要这样设计呢？

先来说说具体的方法：

首先保持乐观，goroutine 采用自旋 + CAS 的策略争夺锁；
尝试持续受挫达到一定条件后，判定当前过于激烈，则由自旋转为阻塞/挂起模式.

这样做的原因是可以具备探测和适应环境，因地制宜采取不同的策略，首先采用乐观的状态，如果几次自旋无果，就认为现在是并发激烈的情况，就会转化为悲观的状态。

1.3 饥饿模式

上一小节的升级策略主要是面向性能，而本小节引入的饥饿模式，则是对公平性问题的探讨。

下面首先拎清两个概念：

饥饿：顾名思义，是因为非公平机制的原因，导致 Mutex 阻塞队列中存在 goroutine 长时间取不到锁，从而陷入饥荒状态；
饥饿模式：当 Mutex 阻塞队列中存在处于饥饿态的 goroutine 时，会进入模式，将抢锁流程由非公平机制转为公平机制.

Mutex运作下的两种模式

正常模式/非饥饿模式：这是 sync.Mutex 默认采用的模式. 当有 goroutine 从阻塞队列被唤醒时，会和此时先进入抢锁流程的 goroutine 进行锁资源的争夺，假如抢锁失败，会重新回到阻塞队列头部.

这里虽然有一个阻塞队列，当锁资源被释放，按理说阻塞队列的队首的G或获取这个锁资源，这其实是很公平了，但是实际上他只是看似公平，因为还有没进阻塞队列的G，还记得什么时候进阻塞队列嘛？对，就是当自旋结束才会进，这样一来就很清晰了，队首的G会和自旋的G抢占这个锁，如果说队首的G排了半天队，结果被这个初出茅庐的自旋G抢了锁资源，这还叫公平嘛？结果显而易见，肯定是不公平的，于是为了解决这个问题，就有了饥饿模式。

（值得一提的是，此时被唤醒的老 goroutine 相比新 goroutine 是处于劣势地位，因为新 goroutine 已经在占用 CPU 时间片，且新 goroutine 可能存在多个，从而形成多对一的人数优势，因此形势对老 goroutine 不利.）

饥饿模式：这是 sync.Mutex 为拯救陷入饥荒的老 goroutine 而启用的特殊机制，饥饿模式下，锁的所有权按照阻塞队列的顺序进行依次传递. 新 goroutine 进行流程时不得抢锁，而是进入队列尾部排队.

这样就可以避免自旋的锁抢占锁资源了

两种模式的转化

默认为正常模式；
正常模式 -> 饥饿模式：当阻塞队列存在 goroutine 等锁超过 1ms 而不得，则进入饥饿模式；
饥饿模式 -> 正常模式：当阻塞队列已清空，或取得锁的 goroutine 等锁时间已低于 1ms 时，则回到正常模式.

小结：正常模式灵活机动，性能较好；饥饿模式严格死板，但能捍卫公平的底线. 因此，两种模式的切换体现了 sync.Mutex 为适应环境变化，在公平与性能之间做出的调整与权衡. 回头观望，这一项因地制宜、随机应变的能力正是许多优秀工具所共有的特质.

二.sync.Mutex

在这之前呢，做一个简单的补充，在sync下，提供了一个接口，提供了一个实现属于自己的锁的方法哦

type Locker interface {
	Lock()
	Unlock()
}

2.1 核心数据结构

type Mutex struct {
	state int32
	sema  uint32
}

state：锁中最核心的状态字段，不同 bit 位分别存储了 mutexLocked(是否上锁)、mutexWoken（是否有 goroutine 从阻塞队列中被唤醒）、mutexStarving（是否处于饥饿模式）的信息，具体在 2.2 节详细展开；
sema：用于阻塞和唤醒 goroutine 的信号量.

const (
    mutexLocked = 1 << iota // mutex is locked
    mutexWoken
    mutexStarving
    mutexWaiterShift = iota

    starvationThresholdNs = 1e6
)

mutexLocked = 1：state 最右侧的一个 bit 位标志是否上锁，0-未上锁，1-已上锁；
mutexWoken = 2：state 右数第二个 bit 位标志是否有 goroutine 从阻塞中被唤醒，0-没有，1-有；
mutexStarving = 4：state 右数第三个 bit 位标志 Mutex 是否处于饥饿模式，0-非饥饿，1-饥饿；
mutexWaiterShift = 3：右侧存在 3 个 bit 位标识特殊信息，分别为上述的 mutexLocked、mutexWoken、mutexStarving；
starvationThresholdNs = 1 ms：sync.Mutex 进入饥饿模式的等待时间阈值.

2.2 state字段

低 3 位分别标识 mutexLocked（是否上锁）、mutexWoken（是否有协程在抢锁）、mutexStarving（是否处于饥饿模式），高 29 位的值聚合为一个范围为 0~2^29-1 的整数，表示在阻塞队列中等待的协程个数.

2.3 加锁Mutex.Lock() （了解即可）

在之前说过一个锁要实现加锁和解锁的操作，接下来就来看看加锁的操作

func (m *Mutex) Lock() {
	// Fast path: 尝试直接通过 CAS 抢占锁
	if atomic.CompareAndSwapInt32(&m.state, 0, mutexLocked) {
		if race.Enabled {
			race.Acquire(unsafe.Pointer(m))
		}
		return
	}
	// Slow path: 处理锁竞争或锁已被持有的情况
	m.lockSlow()
}

使用 原子操作 CompareAndSwapInt32 检查锁状态：如果锁的 state 为 0（未锁定），则将其设为 mutexLocked（1），表示锁被当前 Goroutine 持有。

否则就进入lockslow

来看下这个lockslow

func (m *Mutex) lockSlow() {
	var waitStartTime int64
	starving := false
	awoke := false
	iter := 0
	old := m.state
• waitStartTime：标识当前 goroutine 在抢锁过程中的等待时长，单位：ns；
• starving：标识当前是否处于饥饿模式；
• awoke：标识当前是否已有协程在等锁；
• iter：标识当前 goroutine 参与自旋的次数；
• old：临时存储锁的 state 值.
    
	for {
		// 进入该 if 分支，说明抢锁失败，处于饥饿模式，但仍满足自旋条件
		if old&(mutexLocked|mutexStarving) == mutexLocked && runtime_canSpin(iter) {
            // 进入该 if 分支，说明当前锁阻塞队列有协程，但还未被唤醒，因此需要将      
            // mutexWoken 标识置为 1，避免再有其他协程被唤醒和自己抢锁
			if !awoke && old&mutexWoken == 0 && old>>mutexWaiterShift != 0 &&
				atomic.CompareAndSwapInt32(&m.state, old, old|mutexWoken) {
				awoke = true
			}
			runtime_doSpin()
			iter++
			old = m.state
			continue
		}

        // ......
	}
    
	if race.Enabled {
		race.Acquire(unsafe.Pointer(m))
	}
}

走进 for 循环；
假如满足三个条件：I 锁已被占用、 II 锁为正常模式、III 满足自旋条件（runtime_canSpin 方法），则进入自旋后处理环节；
在自旋后处理中，假如当前锁有尚未唤醒的阻塞协程，则通过 CAS 操作将 state 的 mutexWoken 标识置为 1，将局部变量 awoke 置为 true；
调用 runtime_doSpin 告知调度器 P 当前处于自旋模式；
更新自旋次数 iter 和锁状态值 old；
通过 continue 语句进入下一轮尝试.

上面的部分可以自旋的情况，当一定次数的自旋之后，会改变状态，调整字段，然后进入悲观状态，我们来看看，简单过一遍吧，结合ai的解读

func (m *Mutex) lockSlow() {
    // ......
	for {
        // ......
		new := old
		// 若非饥饿模式，尝试直接获取锁
		if old&mutexStarving == 0 {
			new |= mutexLocked
		}
        // 若锁已被持有或处于饥饿模式，增加等待者数量
		if old&(mutexLocked|mutexStarving) != 0 {
			new += 1 << mutexWaiterShift
		}
        // 若锁已被持有或处于饥饿模式，增加等待者数量
		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 
			}
			 // 加入等待队列（LIFO 或 FIFO，取决于是否已等待过）
			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")
				}
				delta := int32(mutexLocked - 1<<mutexWaiterShift)
				if !starving || old>>mutexWaiterShift == 1 {
                    // 退出饥饿模式
					delta -= mutexStarving
				}
				atomic.AddInt32(&m.state, delta)
				break
			}
			awoke = true
			iter = 0
		} else {
			old = m.state// CAS 失败，重新加载状态
		}
	}

	if race.Enabled {
		race.Acquire(unsafe.Pointer(m))
	}
}

2.4 Unlock （了解即可）

func (m *Mutex) Unlock() {
	if race.Enabled {
		_ = m.state
		race.Release(unsafe.Pointer(m))
	}
	new := atomic.AddInt32(&m.state, -mutexLocked)
	if new != 0 {
		
		m.unlockSlow(new)
	}
}

func (m *Mutex) unlockSlow(new int32) {
	if (new+mutexLocked)&mutexLocked == 0 {
		fatal("sync: unlock of unlocked mutex")
	}
	if new&mutexStarving == 0 {
		old := new
		for {
			// If there are no waiters or a goroutine has already
			// been woken or grabbed the lock, no need to wake anyone.
			// In starvation mode ownership is directly handed off from unlocking
			// goroutine to the next waiter. We are not part of this chain,
			// since we did not observe mutexStarving when we unlocked the mutex above.
			// So get off the way.
			if old>>mutexWaiterShift == 0 || old&(mutexLocked|mutexWoken|mutexStarving) != 0 {
				return
			}
			// Grab the right to wake someone.
			new = (old - 1<<mutexWaiterShift) | mutexWoken
			if atomic.CompareAndSwapInt32(&m.state, old, new) {
				runtime_Semrelease(&m.sema, false, 1)
				return
			}
			old = m.state
		}
	} else {
		// Starving mode: handoff mutex ownership to the next waiter, and yield
		// our time slice so that the next waiter can start to run immediately.
		// Note: mutexLocked is not set, the waiter will set it after wakeup.
		// But mutex is still considered locked if mutexStarving is set,
		// so new coming goroutines won't acquire it.
		runtime_Semrelease(&m.sema, true, 1)
	}
}

这里就不再多介绍了，可以下去自己看一下源码，借助ai，其实只要知道大致的思想，在去编写的时候，在看即可了。

三.sync.RWMutex

从逻辑上，可以把 RWMutex 理解为一把读锁加一把写锁；
写锁具有严格的排他性，当其被占用，其他试图取写锁或者读锁的 goroutine 均阻塞；
读锁具有有限的共享性，当其被占用，试图取写锁的 goroutine 会阻塞，试图取读锁的 goroutine 可与当前 goroutine 共享读锁；
综上可见，RWMutex 适用于读多写少的场景，最理想化的情况，当所有操作均使用读锁，则可实现去无化；最悲观的情况，倘若所有操作均使用写锁，则 RWMutex 退化为普通的 Mutex

3.1 核心数据结构

 type RWMutex struct {
	w           Mutex        // held if there are pending writers
	writerSem   uint32       // semaphore for writers to wait for completing readers
	readerSem   uint32       // semaphore for readers to wait for completing writers
	readerCount atomic.Int32 // number of pending readers
	readerWait  atomic.Int32 // number of departing readers
}

rwmutexMaxReaders：共享读锁的 goroutine 数量上限，值为 2^29；
w：RWMutex 内置的一把普通互斥锁 sync.Mutex；
writerSem：关联写锁阻塞队列的信号量；
readerSem：关联读锁阻塞队列的信号量；
readerCount：正常情况下等于介入读锁流程的 goroutine 数量；当 goroutine 接入写锁流程时，该值为实际介入读锁流程的 goroutine 数量减 rwmutexMaxReaders.
readerWait：记录在当前 goroutine 获取写锁前，还需要等待多少个 goroutine 释放读锁.

源码的走读就不再写了，后续在学分布式锁的时候在完善。