Go锁，我终于搞懂了

编程入门行业动态更新时间:2024-10-27 03:28:08

Go锁，<a href=https://www.elefans.com/category/jswz/34/1737615.html style= 我终于搞懂了"/>

Go锁，我终于搞懂了

自从前两天写了Go互斥锁实现原理后，我感觉我错了，我不应该看源码，看了也没啥用，会了也不敢用。按照源码中的写法，没人想做代码review，也不好做单元测试。

我就是想知道锁是怎么实现的，结果整一堆优化逻辑，感觉懂了又感觉啥都不懂。所以我痛定思痛，我看早期版本行吧。来让我们看2014年的版本

互斥锁：.go

读写互斥锁：.go

互斥锁

这个版本的互斥锁没有饥饿模式、没有自旋、没有各种小的性能优化点。但这段代码把最核心的逻辑展示的十分清楚。

说明

type Mutex struct {state int32sema  uint32
}

state表示互斥锁的状态，比如是否被锁定等。
sema表示信号量，协程阻塞等待该信号量，解锁的协程释放信号量从而唤醒等待信号量的协程。

state的组成如下图所示：

Locked: 表示该Mutex是否已被锁定，0：没有锁定 1：已被锁定。

Woken: 表示是否有协程已被唤醒，0：没有协程唤醒 1：已有协程唤醒，正在加锁过程中。释放锁时，如果正常模式下，不会再唤醒其它协程。

Waiter: 表示阻塞等待锁的协程个数，协程解锁时根据此值来判断是否需要释放信号量

因为在Go互斥锁实现原理已经写过很多基础知识了，这次只把核心逻辑、核心点写一下，如果对其中部分知识不太理解，可以看一下上一篇文章。

源码

// Copyright 2009 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.// Package sync provides basic synchronization primitives such as mutual
// exclusion locks.  Other than the Once and WaitGroup types, most are intended
// for use by low-level library routines.  Higher-level synchronization is
// better done via channels and communication.
//
// Values containing the types defined in this package should not be copied.
package syncimport ("sync/atomic""unsafe"
)// A Mutex is a mutual exclusion lock.
// Mutexes can be created as part of other structures;
// the zero value for a Mutex is an unlocked mutex.
type Mutex struct {state int32sema  uint32
}// A Locker represents an object that can be locked and unlocked.
type Locker interface {Lock()Unlock()
}const (//status只分为三部分，mutexLocked表示锁是否已经被其它协程占用//mutexWoken表示是否唤起协程，让协程开始抢占锁//mutexWaiterShift表示阻塞等待锁的协程个数，占int32的高30位mutexLocked = 1 << iota // mutex is lockedmutexWokenmutexWaiterShift = iota
)// Lock locks m.
// If the lock is already in use, the calling goroutine
// blocks until the mutex is available.
//加锁过程仍然以将status的Locked位置为1位加锁成功
func (m *Mutex) Lock() {// Fast path: grab unlocked mutex.//如果锁没有被占用、没有唤醒的协程、没有等待加锁的协程，直接加锁，成功便返回if atomic.CompareAndSwapInt32(&m.state, 0, mutexLocked) {if raceenabled {//为false，不用管raceAcquire(unsafe.Pointer(m))}return}//协程主动发起的加锁请求，肯定不是被唤醒的，所以awoke为falseawoke := falsefor {old := m.state//无论如何先尝试加锁。因为如果没被占用，肯定是要加锁的。如果被占用，因为使用CAS，所以Locked位也需要为1new := old | mutexLocked//如果锁是被占用的，则将等待协程值加1if old&mutexLocked != 0 {new = old + 1<<mutexWaiterShift}//如果是被唤醒的，CAS操作之后自己就到终止状态（获得锁或者阻塞），所以需要将Woken位置0if awoke {// The goroutine has been woken from sleep,// so we need to reset the flag in either case.new &^= mutexWoken}//CAS操作// old                            new//(0,1)不是唤醒的，被占用了					(+1,0,1)//(1,1)是唤醒的，被占用了           (+1,0,1)//(0,0)不是唤醒的，未被占用         (+0,0,1)//(1,0)  是唤醒的，未被占用         (+0,0,1)if atomic.CompareAndSwapInt32(&m.state, old, new) {if old&mutexLocked == 0 { //如果CAS操作时，锁是未被占用，则加锁成功break}runtime_Semacquire(&m.sema) //否则，将当前协程阻塞awoke = true //当该协程被唤醒是，用awoke进行标记}}if raceenabled {raceAcquire(unsafe.Pointer(m))}
}// Unlock unlocks m.
// It is a run-time error if m is not locked on entry to Unlock.
//
// A locked Mutex is not associated with a particular goroutine.
// It is allowed for one goroutine to lock a Mutex and then
// arrange for another goroutine to unlock it.
// 解锁逻辑也比go1.13版本简单很多
func (m *Mutex) Unlock() {if raceenabled {//默认为false，不用管_ = m.stateraceRelease(unsafe.Pointer(m))}// Fast path: drop lock bit.//将值减一，其实就是将locked位置为0new := atomic.AddInt32(&m.state, -mutexLocked)//小技巧，检查解锁的锁是否未被加锁，如果是这种情况，就panicif (new+mutexLocked)&mutexLocked == 0 {panic("sync: unlock of unlocked mutex")}//因为atomic.AddInt32操作m.state使用的是指针，如果没有其它协程操作m.state的话，new=old=m.stateold := newfor {// If there are no waiters or a goroutine has already// been woken or grabbed the lock, no need to wake anyone.//如果没有等待加锁的协程或者当前锁已经唤醒了其它协程，直接返回if old>>mutexWaiterShift == 0 || old&(mutexLocked|mutexWoken) != 0 {return}// Grab the right to wake someone.//要唤醒等待协程了，所以需要将woken位置为1，同时将等待数量减一new = (old - 1<<mutexWaiterShift) | mutexWokenif atomic.CompareAndSwapInt32(&m.state, old, new) {runtime_Semrelease(&m.sema)//如果解锁成功，则唤醒一个等待的协程return}old = m.state//否则继续尝试解锁}
}

关注点

Unlock中atomic.AddInt32(&m.state, -mutexLocked)是对m.state的指针真行操作，所以其实对m.state进行了真正的减一，这也是CAS操作时atomic.CompareAndSwapInt32(&m.state, old, new)，m.state和old值可能一样的原因
大家课本上学到的信号量伪代码一般是这样的

wait(S)：while S<=0 do no-op；S:=S-1；signal(S)：S:=S+1；

感觉和Mutex中的sema不一样。有这种疑问很正常，因为准确的说，Mutex才是真正意义上的信号量，Mutex中的sema虽然翻译为信号量，但其实只是为了将协程阻塞到以sema为地址的队列上，细节可查看semacquire1、semrelease1。

type Mutex struct {state int32sema  uint32
}

课本上的伪代码：可以加锁，值减一；不可以加锁，死循环。

Mutex具体实现：可以加锁，Locked位置为1，不可以加锁，协程阻塞。

存在等待协程饿死的情况

读写锁

上一篇文章没有写读写锁，主要还是因为go1.13的代码阅读起来太过困难。这次看老版本的代码，简单的多，所以顺便写一下读写锁。

说明

读写锁结构体为：

type RWMutex struct {w           Mutex  // held if there are pending writerswriterSem   uint32 // semaphore for writers to wait for completing readersreaderSem   uint32 // semaphore for readers to wait for completing writersreaderCount int32  // number of pending readersreaderWait  int32  // number of departing readers
}

Mutex：复用写锁，应对读写锁变为写锁的情况。获得写锁首先要获取该锁，如果有一个写锁在进行，那么再到来的写锁将会阻塞于此
writerSem：写阻塞等待的信号量，最后一个读者释放锁时会释放信号量
readerSem：读阻塞的协程等待的信号量，持有写锁的协程释放锁后会释放信号量
readerCount：记录读者个数
readerWait：记录写阻塞时读者个数。表示加写锁的时候有几个读锁，当对应数量的读锁解锁完毕后，写锁开始被唤起。主要目的为防止写锁饿死。

源码

// Copyright 2009 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.package syncimport ("sync/atomic""unsafe"
)// An RWMutex is a reader/writer mutual exclusion lock.
// The lock can be held by an arbitrary number of readers
// or a single writer.
// RWMutexes can be created as part of other
// structures; the zero value for a RWMutex is
// an unlocked mutex.
type RWMutex struct {w           Mutex  // held if there are pending writerswriterSem   uint32 // semaphore for writers to wait for completing readersreaderSem   uint32 // semaphore for readers to wait for completing writersreaderCount int32  // number of pending readersreaderWait  int32  // number of departing readers
}
//读协程最大个数。
//这个值主要和readerCount相比较。如果加写锁，则将readerCount减去rwmutexMaxReaders
//这是一个小技巧，既能看出是否加了写锁，也能使用这个值方便的还原出读协程的数量
const rwmutexMaxReaders = 1 << 30// RLock locks rw for reading.
// 加读锁
func (rw *RWMutex) RLock() {if raceenabled {_ = rw.w.stateraceDisable()}//加读锁，直接将readerCount值加1//什么情况下readerCount会小于0呢，有协程请求写锁了。这种情况下，加读锁的协程需要阻塞//因为加读锁的协程到达时间比加写锁的晚，所以得等写锁用完了才能执行，否则加写锁的协程会饿死，不讲武德if atomic.AddInt32(&rw.readerCount, 1) < 0 {// A writer is pending, wait for it.runtime_Semacquire(&rw.readerSem)}if raceenabled {//为false，不用管raceEnable()raceAcquire(unsafe.Pointer(&rw.readerSem))}
}// RUnlock undoes a single RLock call;
// it does not affect other simultaneous readers.
// It is a run-time error if rw is not locked for reading
// on entry to RUnlock.
//解读锁
func (rw *RWMutex) RUnlock() {if raceenabled {//为false，不用管_ = rw.w.stateraceReleaseMerge(unsafe.Pointer(&rw.writerSem))raceDisable()}//将加读锁的协程数量值减一//如果没有尝试加写锁的协程，值r肯定大于0，则解锁结束//如果有协程加写锁，就需要看一下自己是否是请求写锁前的最后一个读锁协程了，是的话就要唤醒加写锁协程if r := atomic.AddInt32(&rw.readerCount, -1); r < 0 {//说明解锁了未加锁mutex，会panicif r+1 == 0 || r+1 == -rwmutexMaxReaders {raceEnable()panic("sync: RUnlock of unlocked RWMutex")}// A writer is pending.//自己是否是请求写锁前的最后一个读锁协程了，是的话就要唤醒加写锁协程if atomic.AddInt32(&rw.readerWait, -1) == 0 {// The last reader unblocks the writer.runtime_Semrelease(&rw.writerSem)}}if raceenabled {raceEnable()}
}// Lock locks rw for writing.
// If the lock is already locked for reading or writing,
// Lock blocks until the lock is available.
// To ensure that the lock eventually becomes available,
// a blocked Lock call excludes new readers from acquiring
// the lock.
func (rw *RWMutex) Lock() {if raceenabled {_ = rw.w.stateraceDisable()}// First, resolve competition with other writers.//调用mutex加写锁。如果是第一个加写锁的协程，肯定能成功。如果不是第一个，就阻塞了rw.w.Lock()// Announce to readers there is a pending writer.//这个操作实现两个功能，一是将readerCount变为负值，另一个是计算出加写锁时读锁个数r := atomic.AddInt32(&rw.readerCount, -rwmutexMaxReaders) + rwmutexMaxReaders// Wait for active readers.//说明加写锁的时候还有读协程在，则加写锁的协程阻塞if r != 0 && atomic.AddInt32(&rw.readerWait, r) != 0 {runtime_Semacquire(&rw.writerSem)}//当加写锁前的所有读锁协程都释放了，加写锁协程会被唤起if raceenabled {raceEnable()raceAcquire(unsafe.Pointer(&rw.readerSem))raceAcquire(unsafe.Pointer(&rw.writerSem))}
}// Unlock unlocks rw for writing.  It is a run-time error if rw is
// not locked for writing on entry to Unlock.
//
// As with Mutexes, a locked RWMutex is not associated with a particular
// goroutine.  One goroutine may RLock (Lock) an RWMutex and then
// arrange for another goroutine to RUnlock (Unlock) it.
func (rw *RWMutex) Unlock() {if raceenabled {_ = rw.w.stateraceRelease(unsafe.Pointer(&rw.readerSem))raceRelease(unsafe.Pointer(&rw.writerSem))raceDisable()}// Announce to readers there is no active writer.//写锁释放的时候，将readerCount变为正值r := atomic.AddInt32(&rw.readerCount, rwmutexMaxReaders)//如果释放未加锁的mutex，panicif r >= rwmutexMaxReaders {raceEnable()panic("sync: Unlock of unlocked RWMutex")}// Unblock blocked readers, if any.//顺序唤起所有等待的读协程for i := 0; i < int(r); i++ {runtime_Semrelease(&rw.readerSem)}// Allow other writers to proceed.rw.w.Unlock()if raceenabled {raceEnable()}
}// RLocker returns a Locker interface that implements
// the Lock and Unlock methods by calling rw.RLock and rw.RUnlock.
func (rw *RWMutex) RLocker() Locker {return (*rlocker)(rw)
}type rlocker RWMutexfunc (r *rlocker) Lock()   { (*RWMutex)(r).RLock() }
func (r *rlocker) Unlock() { (*RWMutex)(r).RUnlock() }

整个流程如下图

关注点

对于写锁，有两个阻塞。一是Mutex自身阻塞，另一个是RWMutex的writerSem阻塞。两者的作用不一样。Mutex用于处理写锁与写锁之间的关系，writerSem用于处理写锁与读锁之间的关系
源码使用readerWait记录请求写锁时读锁协程个数，当这些协程都释放锁，写锁加锁成功，防止写锁饿死