阿里云 个人网站备案烟台主流网站

当前位置： 首页 > news >正文

阿里云 个人网站备案,烟台主流网站,电商网页模板,建设网站的各种问题文章目录 Mutex锁mutex源码分析LockUnLock mutex两种运行模式mutex normal 正常模式自旋 mutex starvation 饥饿模式 锁的底层实现类型 RWMutexRWMutex 实现其他共享内存线程安全的方式 思考如何设计一个并发更高的锁#xff1f; Mutex锁 mutex源码分析 Locker接口#xff… 文章目录 Mutex锁mutex源码分析LockUnLock mutex两种运行模式mutex normal 正常模式自旋 mutex starvation 饥饿模式 锁的底层实现类型 RWMutexRWMutex 实现其他共享内存线程安全的方式 思考如何设计一个并发更高的锁 Mutex锁 mutex源码分析 Locker接口 type Locker interface {Lock()Unlock() }Mutex 就实现了这个接口Lock请求锁Unlock释放锁 type Mutex struct {state int32 //锁状态保护四部分含义sema uint32 //信号量用于阻塞等待或者唤醒 }Locked表示该 mutex 是否被锁定0 表示没有1 表示处于锁定状态 Woken表示是否有协程被唤醒0 表示没有1 表示有协程处于唤醒状态并且在加锁过程中 StarvingGo1.9 版本之后引入表示 mutex 是否处于饥饿状态0 表示没有1 表示有协程处于饥饿状态 Waiter: 等待锁的协程数量。
方法解析 const (// mutex is locked ,在低位值 1mutexLocked 1 iota//标识有协程被唤醒处于 state 中的第二个 bit 位值 2mutexWoken//标识 mutex 处于饥饿模式处于 state 中的第三个 bit 位值 4mutexStarving// 值 3state 值通过右移三位可以得到 waiter 的数量// 同理state 1 mutexWaiterShift可以累加 waiter 的数量mutexWaiterShift iota// 标识协程处于饥饿状态的最长阻塞时间当前被设置为 1msstarvationThresholdNs 1e6 )Lock func (m *Mutex) Lock() {// Fast path: grab unlocked mutex. //运气好直接加锁成功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() }func (m *Mutex) lockSlow() {var waitStartTime int64// 标识是否处于饥饿模式starving : false// 唤醒标记awoke : false// 自旋次数iter : 0old : m.statefor {// 非饥饿模式下开启自旋操作// 从 runtime_canSpin(iter) 的实现中(runtime/proc.sync_runtime_canSpin)可以知道// 如果 iter 的值大于 4将返回 falseif old(mutexLocked|mutexStarving) mutexLocked runtime_canSpin(iter) {// 如果没有其他 waiter 被唤醒那么将当前协程置为唤醒状态同时 CAS 更新 mutex 的 Woken 位if !awoke oldmutexWoken 0 oldmutexWaiterShift ! 0 atomic.CompareAndSwapInt32(m.state, old, old|mutexWoken) {awoke true}// 开启自旋runtime_doSpin()iter// 重新检查 state 的值old m.statecontinue}new : old// 非饥饿状态if oldmutexStarving 0 {// 当前协程可以直接加锁new | mutexLocked}// mutex 已经被锁住或者处于饥饿模式// 那么当前协程不能获取到锁将会进入等待状态if old(mutexLocked|mutexStarving) ! 0 {// waiter 数量加 1当前协程处于等待状态new 1 mutexWaiterShift}// 当前协程处于饥饿状态并且 mutex 依然被锁住那么设置 mutex 为饥饿模式if starving oldmutexLocked ! 0 {new | mutexStarving}if awoke {if newmutexWoken 0 {throw(sync: inconsistent mutex state)}// 清除唤醒标记// ^ 与非操作mutexWoken: 10 - 01// 此操作之后new 的 Locked 位值是 1如果能够成功写入到 m.state 字段那么当前协程获取锁成功new ^ mutexWoken}// CAS 设置新状态成功if atomic.CompareAndSwapInt32(m.state, old, new) {// 旧的锁状态已经被释放并且处于非饥饿状态// 这个时候当前协程正常请求到了锁就可以直接返回了if old(mutexLocked|mutexStarving) 0 {break}// 处理当前协程的饥饿状态// 如果之前已经处于等待状态了已经在队列里面那么将其加入到队列头部从而可以被高优唤醒queueLifo : waitStartTime ! 0if waitStartTime 0 {// 阻塞开始时间waitStartTime runtime_nanotime()}// P 操作阻塞等待runtime_SemacquireMutex(m.sema, queueLifo, 1)// 唤醒之后如果当前协程等待超过 1ms那么标识当前协程处于饥饿状态starving starving || runtime_nanotime()-waitStartTime starvationThresholdNsold m.state// mutex 已经处于饥饿模式if oldmutexStarving ! 0 {// 1. 如果当前协程被唤醒但是 mutex 还是处于锁住状态// 那么 mutex 处于非法状态//// 2. 或者如果此时 waiter 数量是 0并且 mutex 未被锁住// 代表当前协程没有在 waiters 中但是却想要获取到锁那么 mutex 状态非法if old(mutexLocked|mutexWoken) ! 0 || oldmutexWaiterShift 0 {throw(sync: inconsistent mutex state)}// delta 代表加锁并且将 waiter 数量减 1 两步操作delta : int32(mutexLocked - 1mutexWaiterShift)// 非饥饿状态 或者 当前只剩下一个 waiter 了就是当前协程本身if !starving || oldmutexWaiterShift 1 {// 那么 mutex 退出饥饿模式delta - mutexStarving}// 设置新的状态atomic.AddInt32(m.state, delta)break}awoke trueiter 0} else {old m.state}}if race.Enabled {race.Acquire(unsafe.Pointer(m))} }解锁操作会根据 Mutex.state 的状态来判断需不需要去唤醒其他等待中的协程。 func (m *Mutex) unlockSlow(new int32) {// new - state 字段原子减 1 之后的值如果之前是处于加锁状态那么此时 new 的末位应该是 0// 此时 newmutexLocked 正常情况下会将 new 末位变成 1// 那么如果和 mutexLocked 做与运算之后的结果是 0代表 new 值非法解锁了一个未加锁的 mutexif (newmutexLocked)mutexLocked 0 {throw(sync: unlock of unlocked mutex)}// 如果不是处于饥饿状态if newmutexStarving 0 {old : newfor {// oldmutexWaiterShift 0 代表没有等待加锁的协程了自然不需要执行唤醒操作// oldmutexLocked ! 0 代表已经有协程加锁成功此时没有必要再唤醒一个协程因为它不可能加锁成功// oldmutexWoken ! 0 代表已经有协程被唤醒并且在加锁过程中此时不需要再执行唤醒操作了// oldmutexStarving ! 0 代表已经进入了饥饿状态// 以上四种情况皆不需要执行唤醒操作if oldmutexWaiterShift 0 || old(mutexLocked|mutexWoken|mutexStarving) ! 0 {return}// 唤醒一个等待中的协程将 state woken 位置为 1// old - 1mutexWaiterShift waiter 数量减 1new (old - 1mutexWaiterShift) | mutexWokenif atomic.CompareAndSwapInt32(m.state, old, new) {runtime_Semrelease(m.sema, false, 1)return}old m.state}} else {// 饥饿模式// 将 mutex 的拥有权转移给下一个 waiter并且交出 CPU 时间片从而能够让下一个 waiter 立刻开始执行runtimeSemrelease(m.sema, true, 1)} }UnLock // 解锁操作 func (m *Mutex) Unlock() {if race.Enabled { m.staterace.Release(unsafe.Pointer(m))}// mutexLocked 位设置为 0解锁new : atomic.AddInt32(m.state, -mutexLocked)// 如果此时 state 值不是 0代表其他位不是 0或者出现异常使用导致 mutexLocked 位也不是 0// 此时需要进一步做一些其他操作比如唤醒等待中的协程等if new ! 0 {m.unlockSlow(new)} }mutex两种运行模式 饥饿模式是对公平性和性能的一种平衡它避免了某些 goroutine 长时间的等待锁。在饥饿模式下优先对待的是那些一直在等待的 waiter。 mutex normal 正常模式 默认情况下Mutex的模式为normal。 该模式下协程如果加锁不成功不会立即转入阻塞排队而是判断是否满足自旋的条件如果满足则会启动自旋过程尝试抢锁。 正常模式 高吞吐量
自旋 自旋是一种多线程同步机制当前的进程在进入自旋的过程中会一直保持 CPU 的占用持续检查某个条件是否为真。 在多核的 CPU 上自旋可以避免 Goroutine 的切换使用恰当会对性能带来很大的增益但是使用的不恰当就会拖慢整个程序所以 Goroutine 进入自旋的条件非常苛刻 互斥锁只有在普通模式才能进入自旋runtime.sync_runtime_canSpin 需要返回 true 运行在多 CPU 的机器上当前 Goroutine 为了获取该锁进入自旋的次数小于四次当前机器上至少存在一个正在运行的处理器 P 并且处理的运行队列为空 https://draveness.me/golang/docs/part3-runtime/ch06-concurrency/golang-sync-primitives/ mutex starvation 饥饿模式 自旋过程中能抢到锁一定意味着同一时刻有协程释放了锁我们知道释放锁时如果发现有阻塞等待的协程还会释放一个信号量来唤醒一个等待协程被唤醒的协程得到CPU后开始运行此时发现锁已被抢占了自己只好再次阻塞不过阻塞前会判断自上次阻塞到本次阻塞经过了多长时间如果超过1ms的话会将Mutex标记为饥饿模式然后再阻塞。 处于饥饿模式下不会启动自旋过程也即一旦有协程释放了锁那么一定会唤醒协程被唤醒的协程将会成功获取锁同时也会把等待计数减1。 在饥饿模式下Mutex 的拥有者将直接把锁交给队列最前面的 waiter。新来的 goroutine 不会尝试获取锁即使看起来锁没有被持有它也不会去抢也不会 spin自旋它会乖乖地加入到等待队列的尾部。 如果拥有 Mutex 的 waiter 发现下面两种情况的其中之一它就会把这个 Mutex 转换成正常模式: 此 waiter 已经是队列中的最后一个 waiter 了没有其它的等待锁的 goroutine 了此 waiter 的等待时间小于 1 毫秒ms。 锁的底层实现类型 锁内存总线针对内存的读写操作在总线上控制限制程序的内存访问 锁缓存行同一个缓存行的内容读写操作CPU内部的高速缓存保证一致性 锁作用在一个对象或者变量上。现代CPU会优先在高速缓存查找如果存在这个对象、变量的缓存行数据会使用锁缓存行的方式。否则才使用锁总线的方式。 RWMutex RWMutex 实现 type RWMutex struct {w Mutex // 复用互斥锁能力//写锁信号量 当阻塞写操作的读操作goroutine释放读锁时通过该信号量通知阻塞的写操作的goroutinewriterSem uint32 // 读锁信号量 当写操作goroutine释放写锁时通过该信号量通知阻塞的读操作的goroutinereaderSem uint32 // 当前读操作的数量包含所有已经获取到读锁或者被写操作阻塞的等待获取读锁的读操作数量readerCount int32 // 获取写锁需要等待读锁释放的数量readerWait int32 }通过记录 readerCount 读锁的数量来进行控制当有一个写锁的时候会将读 锁数量设置为负数 130。目的是让新进入的读锁等待之前的写锁释放通知读 锁。同样的当有写锁进行抢占时也会等待之前的读锁都释放完毕才会开始 21 进行后续的操作。 而等写锁释放完之后会将值重新加上 130, 并通知刚才 新进入的读锁rw.readerSem两者互相限制。 const rwmutexMaxReaders 1 30 func (rw *RWMutex) Lock() {// First, resolve competition with other writers.// 写锁也就是互斥锁复用互斥锁的能力来解决与其他写锁的竞争// 如果写锁已经被获取了其他goroutine在获取写锁时会进入自旋或者休眠rw.w.Lock()// 将readerCount设置为负值告诉读锁现在有一个正在等待运行的写锁获取互斥锁成功r : atomic.AddInt32(rw.readerCount, -rwmutexMaxReaders) rwmutexMaxReaders// 获取互斥锁成功并不代表goroutine获取写锁成功我们默认最大有2^30的读操作数目减去这个最大数目// 后仍然不为0则表示前面还有读锁需要等待读锁释放并更新写操作被阻塞时等待的读操作goroutine个数if r ! 0 atomic.AddInt32(rw.readerWait, r) ! 0 {runtime_SemacquireMutex(rw.writerSem, false, 0)} }func (rw *RWMutex) Unlock() {// Announce to readers there is no active writer.// 将readerCount的恢复为正数也就是解除对读锁的互斥r : atomic.AddInt32(rw.readerCount, rwmutexMaxReaders)if r rwmutexMaxReaders {race.Enable()throw(sync: Unlock of unlocked RWMutex)}// 如果后面还有读操作的goroutine则需要唤醒他们for i : 0; i int®; i {runtime_Semrelease(rw.readerSem, false, 0)}// 释放互斥锁写操作的goroutine和读操作的goroutine同时竞争rw.w.Unlock() }读锁 func (rw *RWMutex) RLock() {// 原子操作readerCount 只要值不是负数就表示获取读锁成功if atomic.AddInt32(rw.readerCount, 1) 0 {// 有一个正在等待的写锁为了避免饥饿后面进来的读锁进行阻塞等待runtimeSemacquireMutex(rw.readerSem, false, 0)} } func (rw *RWMutex) RUnlock() {// 将readerCount的值减1如果值等于等于0直接退出即可否则进入rUnlockSlow处理if r : atomic.AddInt32(rw.readerCount, -1); r 0 {// Outlined slow-path to allow the fast-path to be inlinedrw.rUnlockSlow®} }其他共享内存线程安全的方式 官方不太推荐使用锁更多的是通过channel做数据交换 思考 如何设计一个并发更高的锁 在Go语言中使用切片来设计并发更高效的锁是一种常见的做法通常被称为分段锁或分片锁。 这种技术可以在一定程度上减小锁的粒度从而提高并发性能。 package mainimport (fmtsynchash/fnv )const numSegments 16type ConcurrentMap struct {segments []sync.Mutexdata map[interface{}]interface{} }func NewConcurrentMap() *ConcurrentMap {segments : make([]sync.Mutex, numSegments)data : make(map[interface{}]interface{})return ConcurrentMap{segments: segments, data: data} }func (cm *ConcurrentMap) getSegment(key interface{}) *sync.Mutex {hash : hashFunction(key) % numSegmentsreturn cm.segments[hash] }func (cm *ConcurrentMap) Get(key interface{}) interface{} {segment : cm.getSegment(key)segment.Lock()defer segment.Unlock()return cm.data[key] }func (cm *ConcurrentMap) Set(key, value interface{}) {segment : cm.getSegment(key)segment.Lock()defer segment.Unlock()cm.data[key] value }// 假设的哈希函数仅用于示例目的 func hashFunction(key interface{}) int {h : fnv.New32a()// 将键的字节表示写入哈希函数, _ h.Write([]byte(fmt.Sprintf(%v, key)))return int(h.Sum32()) }func main() {concurrentMap : NewConcurrentMap()var wg sync.WaitGroupnumItems : 1000for i : 0; i numItems; i {wg.Add(1)go func(index int) {defer wg.Done()key : fmt.Sprintf(key%d, index)concurrentMap.Set(key, index)}(i)}wg.Wait()// 输出结果for i : 0; i numItems; i {key : fmt.Sprintf(key%d, i)fmt.Printf(%s: %v\n, key, concurrentMap.Get(key))} }