Go 锁资源(一)理论介绍与源码解析

字数统计: 3.3k字   |   阅读时长: 12分
阅读量

一、锁资源介绍

1、乐观锁

乐观锁操作数据的时候很乐观，认为没有人同时修改数据，因此乐观锁不会上锁，在执行更新的时候判断在次期间别人是否修改了数据，如果修改了则进行回滚。使用版本号机制或者CAS算法实现。通常读多写少使用乐观锁。

版本号机制：修改数据的时候获取版本，更新操作进行的时候带上版本，若版本一致才会进行修改

CAS算法：包括3个操作数。

需要读写内存位置(V)，进行比较的预期值(A)，拟写入的新值(B)

操作流程如下：如果内存位置V的值与预期A值相等，则将A值改为B值。否则，不做任何操作。CAS一般是自旋的，如果操作不成功，稍后重试，直到成功为止。值得注意的是CAS是原子操作，虽然是2个步骤，但这是硬件支持的。

CAS的缺点：（一）ABA问题，也就是2个线程依次读内存的值。例如线程1和2，1线程将内存值改为X，2线程将内存值改为Y，2线程将内存值改为X，1线程继续操作。对于1线程看起来这个值没有被动过，但事实上已经被修改。解决的办法就是再带上版本号。（二）开销问题，CAS一直处于自旋状态，消耗CPU。（三）只能保证一个共享变量的原子操作。

乐观锁加锁吗？不加锁，只是修改之前比较内存的值。

2、悲观锁

总是觉得有人会修改数据，所以在修改数据之前先加锁。一旦加锁，只允许一个线程访问共享数据。通常写多读少使用悲观锁。Mysql的读锁、写锁、行锁都是使用悲观锁。

3、自旋锁

与互斥锁类似，在获取不到锁的时候是处于循环忙等，而不是休眠。

4、读写锁

当共享变量被加了写锁，其他线程对该锁加读锁和写锁都是阻塞的；

当共享变量被加了读锁，其他线程对该锁加写锁会阻塞，加读锁会成功。

所以读写锁是一种读共享，写独占的锁，适用与读多写少的情景。

二、mutex互斥锁源码解析

1、mutex工作模式

主要分为正常模式与饥饿模式

一个尝试加锁的goroutine会先自选几次，尝试通过原子操作获得锁，若自旋几次仍不能获得锁，则会通过信号量排队等待，（这种模式有更高的吞吐量，因为频繁的挂起和唤醒goroutine会带来较多的开销但不能无限制的自旋，）所有的等待者都会按照先进先出的顺序排队，但是当锁被释放后，第一个等待者被唤醒后，并不会直接拥有锁，而是需要后来者竞争，也就是那些处于自旋阶段，尚未排队等待的goroutine。这种情况下，后来者更有优势，一方面他们正在CPU上运行，自然比刚被唤醒的goroutine更有优势；另一方面处于自旋状态的goroutine可以有很多，而被唤醒的goroutine每次只有1个，所以被唤醒的goroutine有很大的概率拿不到锁。这种情况下会被重新插入到队列的头部，而不是尾部。

当goroutine加速等待时间超过1ms之后，会把当前mutex从正常模式切换到饥饿模式。在饥饿模式下，mutex的所有权从执行unlock的goroutine直接传递给等待队列头部的goutine。后来者不会自旋也不会尝试获得锁，即使mutex处于unlock的状态，他们会直接到队列的尾部排队等待。当一个等待者获得锁之后，它会在以下2种情况由饥饿模式切换到正常模式。1、等待时间小于1ms，刚来不久。2、是最后一个等待者，等待队列已经空了。后面自然就没有饥饿的goroutine了。所以在正常模式下，自旋和排队是同时存在的。饥饿模式下，没有自旋，都去排队，严格的先来后到。

2、源码分析
2.1 mutex结构体
1
2
3
4
5
6
7
8
9
10
// mutex.go
type Mutex struct {
	state int32 // 互斥锁的状态，加锁和解锁，都是通过atomic包提供原子性的操作该字段
	sema  uint32 // 信号量，主要用作等待队列。用来唤醒goroutine
}

// mutexLocked 锁状态标识，置为1表示已加锁，
// mutexWoken是否已有goroutine被唤醒，置为1表示被唤醒，
// mutexStarving表示mutex的工作模式，0正常，1饥饿模式。
// state其他位(29位)用来记录有多少个goroutine在排队

state是一个32位的bit，每一位都有其意义。

2.2 加锁

分为2种情况：

1）锁处于空闲状态，可以直接加锁。比如第一次goroutine抢占的时候

2）当有goroutine占用锁的时候，开始等待

1
2
3
4
5
6
7
8
9
10
11
12
13
while(0) {
    if 当前mutex = 正常模式 {
			自旋等待  // 注意自旋是有条件的
		} 
    if 当前mutex = 饥饿模式 {
			排队到最后一个乖乖等待 
		} 
	通过信号量挨排队等待 // P申请一个资源
	if gouroutine等待时间>1ms
		设置模式 = 饥饿模式
	if gouroutine等待时间< 1ms || 当前goroutine是最后一个
		设置模式 = 正常模式
}

精彩的部分就看源码吧！

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
func (m *Mutex) Lock() {
	// 若锁处于正常状态，该锁没有被抢占。则直接加锁。比如第一次被goroutine抢占的时候，或者锁处于空闲的时候，也是这种状态。
	if atomic.CompareAndSwapInt32(&m.state, 0, mutexLocked) {
		if race.Enabled {
			race.Acquire(unsafe.Pointer(m))
		}
		return
	}
	// 当有人占有锁的时候，自旋等待
	m.lockSlow()
}

func (m *Mutex) lockSlow() {
	var waitStartTime int64
	// 不是饥饿模式，即正常模式
	starving := false
	// 未被唤醒
	awoke := false
	iter := 0 // 自旋次数
	// 记录当前的状态
	old := m.state
	for {
		// 若mutex已经锁且属于正常模式，尝试自旋。饥饿状态不会自旋
		if old&(mutexLocked|mutexStarving) == mutexLocked && runtime_canSpin(iter) {
			// runtime_canSpin返回正常有如下条件
			//当前Goroutine为了获取该锁进入自旋的次数小于四次；
			//当前机器CPU核数大于1；
			//当前机器上至少存在一个正在运行的处理器 P 并且处理的运行队列为空

			// 正常模式。满足以上条件才可以自旋。
			// 将自己的状态以及锁的状态设置为唤醒，这样当Unlock的时候就不会去唤醒其它被阻塞的goroutine了
			if !awoke && old&mutexWoken == 0 && old>>mutexWaiterShift != 0 &&
				atomic.CompareAndSwapInt32(&m.state, old, old|mutexWoken) {
                    // 这个字段就说明自旋更容易获得锁
				awoke = true
			}
			runtime_doSpin()
			iter++ // 自旋次数++
			old = m.state // 更新状态
			continue
		}
		// 自旋没有获得锁
		// 到了这一步， state的状态可能是：
		// 1. 锁已获取，锁处于正常状态
		// 2. 锁已获取，锁处于饥饿状态
		// 3. 锁空闲， 锁处于正常状态
		// 4. 锁空闲， 锁处于饥饿状态

		new := old // 当前最新状态
		// 若正常状态下，新的goroutine设置锁，尝试通过CAS获得锁；因为是饥饿状态，锁直接给队列的第一个。
		// 正常状态，期望获取锁
		if old&mutexStarving == 0 {
			new |= mutexLocked
		}
		// 饥饿状态，等待队列数量++
		if old&(mutexLocked|mutexStarving) != 0 {
			new += 1 << mutexWaiterShift
		}
		// 如果当前 goroutine处于饥饿状态(等待时间超过1ms)，且锁被其他goroutine获取
		if starving && old&mutexLocked != 0 {
			// 则把锁从正常模式改为饥饿状态
			new |= mutexStarving
		}
		// 如果本goroutine已经设置为唤醒状态, 需要清除new state的唤醒标记, 因为本goroutine要么获得了锁，要么进入休眠，
		// 总之state的新状态不再是woken状态.
		if awoke {
			// The goroutine has been woken from sleep,
			// so we need to reset the flag in either case.
			if new&mutexWoken == 0 {
				throw("sync: inconsistent mutex state")
			}
			new &^= mutexWoken
		}
		// 通过CAS操作更新锁的状态
		if atomic.CompareAndSwapInt32(&m.state, old, new) {
			// 如果说old状态不是饥饿状态也不是被获取状态
			// 那么代表当前goroutine已经通过CAS成功获取了锁
			if old&(mutexLocked|mutexStarving) == 0 {
				break // locked the mutex with CAS
			}
			// If we were already waiting before, queue at the front of the queue.
			// 如果之前等待过，就要放到队列头。
			queueLifo := waitStartTime != 0
			// 若是新来的，放在队尾
			if waitStartTime == 0 {
				// 计算最新的等待时间
				waitStartTime = runtime_nanotime()
			}
			// 竞争失败，使用runtime_SemacquireMutex信号量，保证不会有2个goutine获取
			// 既然未能获取到锁， 那么就使用sleep原语阻塞本goroutine
			// 如果是新来的goroutine,queueLifo=false, 加入到等待队列的尾部，耐心等待
			// 如果是唤醒的goroutine, queueLifo=true, 加入到等待队列的头部
			runtime_SemacquireMutex(&m.sema, queueLifo, 1)
			//  如果当前goroutine已经是饥饿状态了
			//  或者当前goroutine已经等待了1ms（在上面定义常量）以上
			//  就把当前goroutine的状态设置为饥饿
			starving = starving || runtime_nanotime()-waitStartTime > starvationThresholdNs
			// 再次获取一下锁现在的状态
			old = m.state
			// 如果说锁现在是饥饿状态，就代表现在锁是被释放的状态，当前goroutine是被信号量所唤醒的
			// 也就是说，锁被直接交给了当前goroutine
			if old&mutexStarving != 0 {
				if old&(mutexLocked|mutexWoken) != 0 || old>>mutexWaiterShift == 0 {
					throw("sync: inconsistent mutex state")
				}
				// 当前goroutine获得锁，并将等待的goroutine数减1
				delta := int32(mutexLocked - 1<<mutexWaiterShift)
				// 如果本goroutine是最后一个等待者，或者它并不处于饥饿状态，
				// 那么我们需要把锁的state状态设置为正常模式.
				if !starving || old>>mutexWaiterShift == 1 {
					// 退出饥饿模式
					delta -= mutexStarving
				}
				// // 原子性地加上改动的状态
				atomic.AddInt32(&m.state, delta)
				break
			}
			// 如果锁不是饥饿模式，就把当前的goroutine设为被唤醒
			// 并且重置iter(重置spin)
			awoke = true
			iter = 0
		} else {
			// 如果CAS不成功，也就是说没能成功获得锁，锁被别的goroutine获得了或者锁一直没被释放
			// 那么就更新状态，重新开始循环尝试拿锁
			old = m.state
		}
	}

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

分为2种情况

1）若没有等待锁资源的goroutine，成功释放锁

2）有等待的goroutine，把锁给其他goroutine

1
2
3
4
5
6
7
// 不能给未加锁的锁解锁
if 状态 = 正常状态 {
    使用信号量V释放一个资源 // 若有等待的goroutine
}
if 状态 = 饥饿状态 {
    直接唤醒等待队列头部的goroutine
}

精彩部分看源码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
// 这里的释放锁的意思就是如果还有等待该资源的goroutine，会把该锁资源继续让给另外的goroutine(也就是加锁)
func (m *Mutex) Unlock() {
	if race.Enabled {
		_ = m.state
		race.Release(unsafe.Pointer(m))
	}

	// Fast path: drop lock bit.
	// 去除加锁状态
	new := atomic.AddInt32(&m.state, -mutexLocked)
	if new != 0 { // 存在其他等待的goroutine,把锁给其他goroutine
		m.unlockSlow(new)
	}
}

func (m *Mutex) unlockSlow(new int32) {
	// 如果说锁不是处于locked状态，那么对锁执行Unlock会导致panic
	if (new+mutexLocked)&mutexLocked == 0 {
		throw("sync: unlock of unlocked mutex")
	}
	// 如果锁的状态是正常状态（可以存在自旋锁与唤醒锁的抢占）
	if new&mutexStarving == 0 {
		old := new
		for {
			//没有等待的goroutine，或者当前goroutine已经抢到锁，已经被唤醒。即没有任何处于等待的goroutine。什么都不做
			if old>>mutexWaiterShift == 0 || old&(mutexLocked|mutexWoken|mutexStarving) != 0 {
				return
			}
			// 走到这一步的时候，说明锁目前还是空闲状态，并且没有goroutine被唤醒且队列中有goroutine等待拿锁
			// 那么我们就要把锁的状态设置为被唤醒，等待队列-1
			new = (old - 1<<mutexWaiterShift) | mutexWoken
			if atomic.CompareAndSwapInt32(&m.state, old, new) {
				// 如果状态设置成功了，我们就通过信号量去唤醒goroutine
				// V释放一个资源。唤醒一个阻塞的goroutine 但不是第一个等待者，可能有自旋的
				runtime_Semrelease(&m.sema, false, 1)
				return
			}
			// 循环结束的时候，更新一下状态，因为有可能在执行的过程中，状态被修改了(比如被Lock改为了饥饿状态)
			old = m.state
		}
	} else {
		// 饥饿状态，那么我们就直接把锁的所有权通过信号量移交给队列头的goroutine就好了
		// handoff = true表示直接把锁交给队列头部的goroutine
		// V释放一个资源，唤醒等待的goroutine
		runtime_Semrelease(&m.sema, true, 1)
	}
}

后续文章还继续对以下内容进行分析

1、信号量的PV操作

1
2
func runtime_SemacquireMutex(s *uint32, lifo bool, skipframes int)
func runtime_Semrelease(s *uint32, handoff bool, skipframes int)

2、自旋的操作

1
2
func runtime_canSpin(i int) bool
func runtime_doSpin()

3、Rwmutex 读写互斥锁

引用

赏

感谢您的认可

支付宝

扫一扫，分享到微信

微信分享二维码

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true