面试官：哥们Go语言的互斥锁了解多少？

前言

asongGochannelGosyncchannel

Golang

Go语言互斥锁设计实现

mutex介绍

syncmutexLock()Unlock()Go1.18TryLock()

Lock()LockgoroutinepanicUnlock()UnLockpanictryLock()TryLock

mutex

type Mutex struct {
 state int32
 sema  uint32
}

statesemagoroutine

statemutexedmutexWokenmutexStarvinggoroutine

const (
   mutexLocked = 1 << iota // 表示互斥锁的锁定状态
   mutexWoken // 表示从正常模式被从唤醒
   mutexStarving // 当前的互斥锁进入饥饿状态
   mutexWaiterShift = iota // 当前互斥锁上等待者的数量
)

mutexgouroutinegoroutinegoroutineGo1.9goroutine1msgoroutine

mutexmutex

Lock加锁

Lock

func (m *Mutex) Lock() {
 // 判断当前锁的状态，如果锁是完全空闲的，即m.state为0，则对其加锁，将m.state的值赋为1
 if atomic.CompareAndSwapInt32(&m.state, 0, mutexLocked) {
  if race.Enabled {
   race.Acquire(unsafe.Pointer(m))
  }
  return
 }
 // Slow path (outlined so that the fast path can be inlined)
 m.lockSlow()
}

上面的代码主要两部分逻辑：

CASstategoroutinelockSlowgoroutinelockSlow

lockSlowfor

CAS

初始化状态

locakSlow

func (m *Mutex) lockSlow() {
 var waitStartTime int64 
 starving := false
 awoke := false
 iter := 0
 old := m.state
 ........
}

waitStartTimewaiterstarvingawokegoroutineiterold

自旋

自旋的判断条件非常苛刻：

for {
    // 判断是否允许进入自旋 两个条件，条件1是当前锁不能处于饥饿状态
    // 条件2是在runtime_canSpin内实现，其逻辑是在多核CPU运行，自旋的次数小于4
  if old&(mutexLocked|mutexStarving) == mutexLocked && runtime_canSpin(iter) {
      // !awoke 判断当前goroutine不是在唤醒状态
      // old&mutexWoken == 0 表示没有其他正在唤醒的goroutine
      // old>>mutexWaiterShift != 0 表示等待队列中有正在等待的goroutine
      // atomic.CompareAndSwapInt32(&m.state, old, old|mutexWoken) 尝试将当前锁的低2位的Woken状态位设置为1，表示已被唤醒, 这是为了通知在解锁Unlock()中不要再唤醒其他的waiter了
   if !awoke && old&mutexWoken == 0 && old>>mutexWaiterShift != 0 &&
    atomic.CompareAndSwapInt32(&m.state, old, old|mutexWoken) {
     // 设置当前goroutine唤醒成功
          awoke = true
   }
      // 进行自旋
   runtime_doSpin()
      // 自旋次数
   iter++
      // 记录当前锁的状态
   old = m.state
   continue
  }
}

goroutinemutex

old&(mutexLocked|mutexStarving) == mutexLocked

mutexLocked

mutexStarving

mutexLocked|mutexStarving&

runtime_canSpin()

// / go/go1.18/src/runtime/proc.go
const active_spin     = 4
func sync_runtime_canSpin(i int) bool {
 if i >= active_spin || ncpu <= 1 || gomaxprocs <= int32(sched.npidle+sched.nmspinning)+1 {
  return false
 }
 if p := getg().m.p.ptr(); !runqempty(p) {
  return false
 }
 return true
}

自旋条件如下：

CPUGOMAXPROCS>1

goroutineruntime_doSpin

const active_spin_cnt = 30
func sync_runtime_doSpin() {
 procyield(active_spin_cnt)
}
// asm_amd64.s
TEXT runtime·procyield(SB),NOSPLIT,$0-0
 MOVL cycles+0(FP), AX
again:
 PAUSE
 SUBL $1, AX
 JNZ again
 RET

30PAUSECPUCPU

这就是整个自旋操作的逻辑，这个就是为了优化 等待阻塞->唤醒->参与抢占锁这个过程不高效，所以使用自旋进行优化，在期望在这个过程中锁被释放。

抢锁准备期望状态

mutexLockedmutexStarvingmutexWokenmutexWaiterShift

mutexLocked

// 基于old状态声明到一个新状态
  new := old
  // 新状态处于非饥饿的条件下才可以加锁
  if old&mutexStarving == 0 {
   new |= mutexLocked
  }

mutexWaiterShift

//如果old已经处于加锁或者饥饿状态，则等待者按照FIFO的顺序排队
if old&(mutexLocked|mutexStarving) != 0 {
   new += 1 << mutexWaiterShift
  }

mutexStarving

// 如果当前锁处于饥饿模式，并且已被加锁，则将低3位的Starving状态位设置为1，表示饥饿
if starving && old&mutexLocked != 0 {
   new |= mutexStarving
  }

mutexWoken

// 当前goroutine的waiter被唤醒,则重置flag
if awoke {
   // 唤醒状态不一致，直接抛出异常
   if new&mutexWoken == 0 {
    throw("sync: inconsistent mutex state")
   }
     // 新状态清除唤醒标记，因为后面的goroutine只会阻塞或者抢锁成功
     // 如果是挂起状态，那就需要等待其他释放锁的goroutine来唤醒。
     // 假如其他goroutine在unlock的时候发现Woken的位置不是0，则就不会去唤醒，那该goroutine就无法在被唤醒后加锁
   new &^= mutexWoken
}

CAS

CAS

// 这里尝试将锁的状态更新为期望状态
if atomic.CompareAndSwapInt32(&m.state, old, new) {
  // 如果原来锁的状态是没有加锁的并且不处于饥饿状态，则表示当前goroutine已经获取到锁了，直接推出即可
   if old&(mutexLocked|mutexStarving) == 0 {
    break // locked the mutex with CAS
   }
   // 到这里就表示goroutine还没有获取到锁，waitStartTime是goroutine开始等待的时间，waitStartTime != 0就表示当前goroutine已经等待过了，则需要将其放置在等待队列队头，否则就排到队列队尾
   queueLifo := waitStartTime != 0
   if waitStartTime == 0 {
    waitStartTime = runtime_nanotime()
   }
      // 阻塞等待
   runtime_SemacquireMutex(&m.sema, queueLifo, 1)
      // 被信号量唤醒后检查当前goroutine是否应该表示为饥饿
     // 1. 当前goroutine已经饥饿
     // 2. goroutine已经等待了1ms以上
   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")
    }
        // 当前goroutine已经获取了锁，等待队列-1
    delta := int32(mutexLocked - 1<<mutexWaiterShift
         // 当前goroutine非饥饿状态 或者 等待队列只剩下一个waiter，则退出饥饿模式(清除饥饿标识位)              
    if !starving || old>>mutexWaiterShift == 1 {
     delta -= mutexStarving
    }
        // 更新状态值并中止for循环，拿到锁退出
    atomic.AddInt32(&m.state, delta)
    break
   }
      // 设置当前goroutine为唤醒状态，且重置自璇次数
   awoke = true
   iter = 0
  } else {
      // 锁被其他goroutine占用了，还原状态继续for循环
   old = m.state
  }

CASruntime.sync_runtime_SemacquireMutexgoroutineruntime.sync_runtime_SemacquireMutexgoroutinegoroutinegoroutine

解锁

UnLock

func (m *Mutex) Unlock() {
 // Fast path: drop lock bit.
 new := atomic.AddInt32(&m.state, -mutexLocked)
 if new != 0 {
  // Outlined slow path to allow inlining the fast path.
  // To hide unlockSlow during tracing we skip one extra frame when tracing GoUnblock.
  m.unlockSlow(new)
 }
}

AddInt320goroutineunlockSlow

func (m *Mutex) unlockSlow(new int32) {
  // 这里表示解锁了一个没有上锁的锁，则直接发生panic
 if (new+mutexLocked)&mutexLocked == 0 {
  throw("sync: unlock of unlocked mutex")
 }
  // 正常模式的释放锁逻辑
 if new&mutexStarving == 0 {
  old := new
  for {
      // 如果没有等待者则直接返回即可
      // 如果锁处于加锁的状态，表示已经有goroutine获取到了锁，可以返回
      // 如果锁处于唤醒状态，这表明有等待的goroutine被唤醒了，不用尝试获取其他goroutine了
      // 如果锁处于饥饿模式，锁之后会直接给等待队头goroutine
   if old>>mutexWaiterShift == 0 || old&(mutexLocked|mutexWoken|mutexStarving) != 0 {
    return
   }
   // 抢占唤醒标志位，这里是想要把锁的状态设置为被唤醒，然后waiter队列-1
   new = (old - 1<<mutexWaiterShift) | mutexWoken
   if atomic.CompareAndSwapInt32(&m.state, old, new) {
        // 抢占成功唤醒一个goroutine
    runtime_Semrelease(&m.sema, false, 1)
    return
   }
      // 执行抢占不成功时重新更新一下状态信息，下次for循环继续处理
   old = m.state
  }
 } else {
    // 饥饿模式释放锁逻辑，直接唤醒等待队列goroutine
  runtime_Semrelease(&m.sema, true, 1)
 }
}

goroutinefunc runtime_Semrelease(s *uint32, handoff bool, skipframes int)handoff is true, pass count directly to the first waiter.

非阻塞加锁

Go1.18TryLock()

func (m *Mutex) TryLock() bool {
  // 记录当前状态
 old := m.state
  //  处于加锁状态/饥饿状态直接获取锁失败
 if old&(mutexLocked|mutexStarving) != 0 {
  return false
 }
 // 尝试获取锁，获取失败直接获取失败
 if !atomic.CompareAndSwapInt32(&m.state, old, old|mutexLocked) {
  return false
 }


 return true
}

TryLock

• 判断当前锁的状态，如果锁处于加锁状态或饥饿状态直接获取锁失败
• 尝试获取锁，获取失败直接获取锁失败

TryLock

总结

通读源码后你会发现互斥锁的逻辑真的十分复杂，代码量虽然不多，但是很难以理解，一些细节点还需要大家多看看几遍才能理解其为什么这样做，文末我们再总结一下互斥锁的知识点：

goroutinegoroutineGo1.9goroutine1msMutexgoroutine1msMutexgoroutinegoroutineunlockedslowmutexLockedgoroutine

MutexMutexLock

本文之后你对互斥锁有什么不理解的吗？欢迎评论区批评指正～；

好啦，本文到这里就结束了，我是asong，我们下期见。

欢迎关注公众号：Golang梦工厂