前言大家好，我是asong。说到并发编程和多线程编程，大家第一时间会想到锁。锁是并发编程中的同步原语。它们可以保证多个线程在访问同一块内存时不会发生竞争，保证并发安全；在Go语言中，更推荐channel通过通信实现共享内存。这个设计点与很多主流编程语言不一致。不过Go语言在sync包中也提供了互斥锁和读写锁。渠道毕竟不能满足所有场景，互斥锁和读写锁的使用离不开我们，所以接下来我会分两篇分享互斥锁和读写锁的实现方式。本篇我们先来了解一下互斥锁。完成。本文基于Golang版本：1.18Go语言mutex设计与实现mutex介绍sync包下的mutex是互斥量，提供了三个公共方法：调用Lock()获取锁，调用Unlock()释放锁，在Go1中。18新增非阻塞取锁操作的TryLock()方法：Lock()：调用Lock方法进行加锁操作。使用时需要注意的是，在同一个goroutine中，只有在释放锁的时候才能再次加锁，否则会导致程序panic。Unlock()：调用UnLock方法进行解锁操作。在使用的时候要注意，在没有加锁的情况下释放锁会导致程序panic。加锁的Mutex不与具体的goroutine相关联，所以你可以使用一个goroutine为它加锁，然后使用其他goroutine解锁。tryLock()：调用TryLock方法尝试获取锁。当锁被其他goroutine占用，或者当前锁处于饥饿模式时，会立即返回false。当锁可用时，尝试获取锁。如果获取失败，则不会自旋/阻塞。它也会立即返回false；mutex的结构比较简单，只有两个字段：typeMutexstruct{stateint32semauint32}state：表示当前mutex的状态，复合字段；sema：信号量变量，用于控制等待goroutine的阻塞睡眠和唤醒乍一看可能有点迷惑。Mutex应该是个复杂的东西。怎么可能只有两个字段就实现了呢？那是因为设计使用位作为标志。状态的不同位代表不同的状态，用最小的内存来代表更多的意义。低三位从低到高分别代表mutexed、mutexWoken、mutexStarving。,剩下的位用来表示当前有多少个goroutines在等待锁：const(mutexLocked=1<>mutexWaiterShift!=0表示等待队列中有等待goroutines//atomic.CompareAndSwapInt32(&m.state,old,old|mutexWoken)尝试将当前锁的低2位的Woken状态位设置为1，表示已经被唤醒。这是为了通知其他服务员不应该在Unlock()中被唤醒if!awoke&&old&mutexWoken==0&&old>>mutexWaiterShift！=0&&atomic.CompareAndSwapInt32(&m.state,old,old|mutexWoken){//设置当前goroutine成功唤醒awoke=true}//自旋runtime_doSpin()//自旋次数iter++//记录当前的状态thelockold=m.statecontinue}}Spin这里的条件还是很复杂的。我们之所以要让当前goroutine进入自旋，是因为我们看好当前持有锁的goroutine可以在短时间内归还锁。所以我们需要一些条件来判断。文中描述一下mutex的判断条件：old&(mutexLocked|mutexStarving)==mutexLocked是用来判断锁是否处于普通模式并被锁定的，为什么要这样判断呢？mutexLocked的二进制表示是0001。mutexStarving的二进制表示是0100。mutexStarving的二进制表示是0101。用0101做当前状态下的&操作。如果当前处于饥饿模式，则低三位必须为1。如果当前处于锁模式，则低1位必须为1。为1，因此可以通过该方法判断当前锁是否处于正常模式并锁定；runtime_canSpin()方法用于判断自旋条件是否满足：///go/go1.18/src/runtime/proc.goconstactive_spin=4funcsync_runtime_canSpin(iint)bool{ifi>=active_spin||中央处理器<=1||gomaxprocs<=int32(sched.npidle+sched.nmspinning)+1{返回false}如果p:=getg()。m.p.ptr();!runqempty(p){returnfalse}returntrue}旋转条件如下：旋转次数必须在4次以内。CPU必须是多核GOMAXPROCS>1。当前机器上至少有一个正在运行的处理器P。且处理运行队列为空；判断当前goroutine可以自旋后，调用runtime_doSpin方法自旋：constactive_spin_cnt=30funcsync_runtime_doSpin(){procyield(active_spin_cnt)}//asm_amd64.sTEXTruntimeprocyield(SB),NOSPLIT,$0-0MOVLcycles+0(FP),AXagain:PAUSESUBL$1,AXJNZagainRET循环次数设置为30次，自旋操作就是执行30次PAUSE指令，占用CPU，消耗CPU时间。执行忙等待；这是整个自旋操作的逻辑，这是优化等待阻塞->唤醒->参与抢占锁过程效率不高，所以使用自旋进行优化。预计在这个过程中会释放锁，抢锁就绪，为想要的状态做准备。自旋逻辑处理完后，根据上下文计算出当前互斥量的最新状态，根据不同的条件来确定。计算mutexLocked、mutexStarving、mutexWoken和mutexWaiterShift：首先计算mutexLocked的值：//根据旧状态声明一个新状态new:=old//只有old&mutexStarving==0才能锁定新状态{new|=mutexLocked}计算mutexWaiterShift的值：//如果old已经被锁定或饥饿，则等待者按FIFO顺序排队ifold&(mutexLocked|mutexStarving)!=0{new+=1<starvationThresholdNs//再次获取锁的当前状态old=m.state//如果当前处于饥饿模式，ifold&mutexStarving!=0{//如果当前锁既没有被获取也没有被唤醒，或者等待队列为空这意味着锁状态存在不一致的问题ifold&(mutexLocked|mutexWoken)!=0||old>>mutexWaiterShift==0{throw("sync:inconsistentmutexstate")}//当前goroutine已经获取到锁，正在等待Queue-1delta:=int32(mutexLocked-1<>mutexWaiterShift==1{delta-=mutexStarving}//更新状态value并中止for循环，获取锁并退出atomic.AddInt32(&m.state,delta)break}//设置当前goroutine为唤醒状态，并重置次数awoke=trueiter=0}else{//锁被其他goroutine占用，恢复状态继续for循环old=m.state}这块逻辑很复杂，判断是否通过CAS获取锁，锁是不是通过CAS获取的，会调用runtime.sync_runtime_SemacquireMutex保证资源不会被两个goroutin获取es通过信号量。runtime.sync_runtime_SemacquireMutex会在方法中不断尝试获取锁，然后陷入睡眠等待信号量的释放。当前goroutine一旦能够获取到信号量，就会立即Return，如果是新的goroutine，则需要放在队列的尾部；如果是等待锁被唤醒的goroutine，应该放在队头，整个过程需要写代码加深理解与加锁操作相比，解锁逻辑没有那么复杂。接下来我们看一下UnLock的逻辑：func(m*Mutex)Unlock(){//快速路径：丢弃锁位。new:=atomic.AddInt32(&m.state,-mutexLocked)ifnew!=0{//概述慢速路径以允许内联快速路径。//为了在跟踪过程中隐藏unlockSlow，我们在跟踪GoUnblock时跳过一个额外的帧。m.unlockSlow(new)}}使用AddInt32方法快速解锁，将m.state的低1位置0，再判断新的m.state值。如果值为0，表示当前锁完全释放，解锁结束。不等于0，当前锁没有被占用，会有未被唤醒的等待goroutine，需要一系列的唤醒操作。这部分逻辑在unlockSlow方法中：func(m*Mutex)unlockSlow(newint32){//这里表示unlocked如果一个锁没有上锁，直接会panicif(new+mutexLocked)&mutexLocked==0{throw("sync:unlockofunlockedmutex")}//正常模式锁释放逻辑ifnew&mutexStarving==0{old:=newfor{//如果没有waiter，直接返回//如果有锁locked,表示一个goroutine已经获取到锁，可以return//如果锁被唤醒，表示有等待goroutines唤醒，不需要尝试获取其他goroutines//如果锁处于饥饿模式，加锁后会直接交给等待队列头的goroutine。如果旧>>mutexWaiterShift==0||old&(mutexLocked|mutexWoken|mutexStarving)!=0{返回rn}//抢占唤醒标志，这里是设置锁被唤醒的状态，然后waiterqueue-1new=(old-1<
