【Go并发编程】01 | Mutex（资源并发访问问题之利器） golang

文章图片

Go 并发编程的第一谈，我们谈谈简单的并发访问
并发访问问题，实属很常见了，比如多个 Goroutine 并发操作同一个资源，eg: 计数器；同时更新用户的账户信息；秒杀系统；往同一个 buffer 中并发写入数据等等。如果没有互斥控制，就会出现一些异常情况，比如计数器的计数不准确、用户的账户可能出现透支、秒杀系统出现超卖、buffer 中的数据混乱等，后果都很严重。
那么如何解决这些问题呢？可能你猜对了，就是用互斥锁，那在 Go 语言里，就是 Mutex。接下来我们一起详细了解互斥锁的实现机制，以及 Go 标准库的互斥锁 Mutex 的基本使用方法。
好了，我们先来看看互斥锁的实现机制。
互斥锁的实现机制互斥锁是并发控制的一个基本手段，是为了避免竞争而建立的一种并发控制机制。在学习它的具体实现原理前，我们要先搞懂一个概念，就是临界区（并发编程中，如果程序中的一部分会被并发访问或修改，那么，为了避免并发访问导致的意想不到的结果，这部分程序需要被保护起来，即这部分被保护起来的程序）
如果很多线程同步访问临界区，就会造成访问或操作错误，这当然不是我们希望看到的结果。所以，我们可以使用互斥锁，限定临界区只能同时由一个线程持有。当临界区由一个线程持有的时候，其它线程如果想进入这个临界区，就会返回失败，或者是等待。直到持有的线程退出临界区，这些等待线程中的某一个才有机会接着持有这个临界区。

文章图片

你看，互斥锁就很好地解决了资源竞争问题，有人也把互斥锁叫做排它锁。那在 Go 标准库中，它提供了 Mutex 来实现互斥锁这个功能。Mutex 是使用最广泛的同步原语（Synchronization primitives，有人也叫做并发原语。关于同步原语，并没有一个严格的定义，你可以把它看作解决并发问题的一个基础的数据结构。
Mutex 的基本使用方法在正式看 Mutex 用法之前呢，我想先给你交代一件事：Locker 接口。在 Go 的标准库中，package sync 提供了锁相关的一系列同步原语，这个 package 还定义了一个 Locker 的接口，Mutex 就实现了这个接口。
Locker 的接口定义了锁同步原语的方法集：

type Locker interface { Lock() Unlock()}

可以看到，Go 定义的锁接口的方法集很简单，就是请求锁（Lock）和释放锁（Unlock）这两个方法，秉承了 Go 语言一贯的简洁风格。
但是，这个接口在实际项目应用得不多，因为我们一般会直接使用具体的同步原语，而不是通过接口。简单来说，互斥锁 Mutex 就提供两个方法 Lock 和 Unlock：进入临界区之前调用 Lock 方法，退出临界区的时候调用 Unlock 方法：

func(m *Mutex)Lock() func(m *Mutex)Unlock()

当一个 goroutine 通过调用 Lock 方法获得了这个锁的拥有权后，其它请求锁的 goroutine 就会阻塞在 Lock 方法的调用上，直到锁被释放并且自己获取到了这个锁的拥有权。
看到这儿，你可能会问，为啥一定要加锁呢？别急，我带你来看一个并发访问场景中不使用锁的例子，看看实现起来会出现什么状况。在这个例子中，我们创建了 10 个 goroutine，同时不断地对一个变量（count）进行加 1 操作，每个 goroutine 负责执行 10 万次的加 1 操作，我们期望的最后计数的结果是 10 * 100000 = 1000000 (一百万)。

package mainimport ( "fmt" "sync" )func main() { var count = 0 // 使用WaitGroup等待10个goroutine完成 var wg sync.WaitGroup wg.Add(10) for i := 0; i < 10; i++ { go func() { defer wg.Done() // 对变量count执行10次加1 for j := 0; j < 100000; j++ { count++ } }() } // 等待10个goroutine完成 wg.Wait() fmt.Println(count) return }

在这段代码中，我们使用 sync.WaitGroup 来等待所有的 goroutine 执行完毕后，再输出最终的结果。sync.WaitGroup 现在你只需要知道，我们使用它来控制等待一组 goroutine 全部做完任务。
但是，每次运行，你都可能得到不同的结果，基本上不会得到理想中的一百万的结果。

文章图片

这是为什么呢？
其实，这是因为，count++ 不是一个原子操作，它至少包含几个步骤，比如读取变量 count 的当前值，对这个值加 1，把结果再保存到 count 中。因为不是原子操作，就可能有并发的问题。
比如，10 个 goroutine 同时读取到 count 的值为 9527，接着各自按照自己的逻辑加 1，值变成了 9528，然后把这个结果再写回到 count 变量。但是，实际上，此时我们增加的总数应该是 10 才对，这里却只增加了 1，好多计数都被“吞”掉了。这是并发访问共享数据的常见错误。

// count++操作的汇编代码 MOVQ"".count(SB), AX LEAQ1(AX), CX MOVQCX, "".count(SB)

这个问题，有经验的开发人员还是比较容易发现的，但是，很多时候，并发问题隐藏得非常深，即使是有经验的人，也不太容易发现或者 Debug 出来。针对这个问题，Go 提供了一个检测并发访问共享资源是否有问题的工具： race detector，它可以帮助我们自动发现程序有没有 data race 的问题。Go race detector 是基于 Google 的 C/C++ sanitizers 技术实现的，编译器通过探测所有的内存访问，加入代码能监视对这些内存地址的访问（读还是写）。在代码运行的时候，race detector 就能监控到对共享变量的非同步访问，出现 race 的时候，就会打印出警告信息。
我们来看看这个工具怎么用。
在编译（compile）、测试（test）或者运行（run）Go 代码的时候，加上 race 参数，就有可能发现并发问题。比如在上面的例子中，我们可以加上 race 参数运行，检测一下是不是有并发问题。如果你 go run -race counter.go，就会输出警告信息。

文章图片

文章图片

这个警告不但会告诉你有并发问题，而且还会告诉你哪个 goroutine 在哪一行对哪个变量有写操作，同时，哪个 goroutine 在哪一行对哪个变量有读操作，就是这些并发的读写访问，引起了 data race。
虽然这个工具使用起来很方便，但是，因为它的实现方式，只能通过真正对实际地址进行读写访问的时候才能探测，所以它并不能在编译的时候发现 data race 的问题。而且，在运行的时候，只有在触发了 data race 之后，才能检测到，如果碰巧没有触发（比如一个 data race 问题只能在 2 月 14 号零点或者 11 月 11 号零点才出现），是检测不出来的。
而且，把开启了 race 的程序部署在线上，还是比较影响性能的。运行 go tool compile -race -S counter.go，可以查看计数器例子的代码:

...... rel 5+4 t=17 TLS+0 rel 43+4 t=8 runtime.racefuncenter+0 rel 50+4 t=16 type.int+0 rel 59+4 t=8 runtime.newobject+0 rel 78+4 t=8 runtime.racewrite+0 rel 97+4 t=16 type.sync.WaitGroup+0 rel 106+4 t=8 runtime.newobject+0 rel 134+4 t=8 runtime.racewriterange+0 rel 172+4 t=8 sync.(*WaitGroup).Add+0 rel 195+4 t=16 "".main.func1·f+0 rel 225+4 t=8 runtime.newproc+0 rel 253+4 t=8 sync.(*WaitGroup).Wait+0 rel 267+4 t=8 runtime.raceread+0 rel 284+4 t=8 runtime.convT64+0 rel 304+4 t=16 type.int+0 rel 321+4 t=16 os.Stdout+0 rel 330+4 t=8 runtime.raceread+0 rel 337+4 t=16 os.Stdout+0 rel 344+4 t=16 go.itab.*os.File,io.Writer+0 rel 386+4 t=8 fmt.Fprintln+0 ......

在编译的代码中，增加了 runtime.racefuncenter、runtime.raceread、runtime.racewrite、runtime.racefuncexit 等检测 data race 的方法。通过这些插入的指令，Go race detector 工具就能够成功地检测出 data race 问题了。
总结一下，通过在编译的时候插入一些指令，在运行时通过这些插入的指令检测并发读写从而发现 data race 问题，就是这个工具的实现机制。既然这个例子存在 data race 问题，我们就要想办法来解决它。这个时候，我们这节课的主角 Mutex 就要登场了，它可以轻松地消除掉 data race。
具体怎么做呢？下面，我就结合这个例子，来具体给你讲一讲 Mutex 的基本用法。
我们知道，这里的共享资源是 count 变量，临界区是 count++，只要在临界区前面获取锁，在离开临界区的时候释放锁，就能完美地解决 data race 的问题了。

package mainimport ( "fmt" "sync" )func main() { // 互斥锁保护计数器 var mu sync.Mutex // 计数器的值 var count = 0 // 辅助变量，用来确认所有的goroutine都完成 var wg sync.WaitGroup wg.Add(10)// 启动10个gourontine for i := 0; i < 10; i++ { go func() { defer wg.Done() // 累加10万次 for j := 0; j < 100000; j++ { mu.Lock() count++ mu.Unlock() } }() } wg.Wait() fmt.Println(count)return }

如果你再运行一下程序，就会发现，data race 警告没有了，系统干脆地输出了 1000000：

文章图片

怎么样，使用 Mutex 是不是非常高效？效果很惊喜。
这里有一点需要注意：Mutex 的零值是还没有 goroutine 等待的未加锁的状态，所以你不需要额外的初始化，直接声明变量（如 var mu sync.Mutex）即可。
【【Go并发编程】01 | Mutex（资源并发访问问题之利器）】那 Mutex 还有哪些用法呢？
很多情况下，Mutex 会嵌入到其它 struct 中使用，比如下面的方式：

type Counter struct { Musync.Mutex Count uint64 }

在初始化嵌入的 struct 时，也不必初始化这个 Mutex 字段，不会因为没有初始化出现空指针或者是无法获取到锁的情况。
有时候，我们还可以采用嵌入字段的方式。通过嵌入字段，你可以在这个 struct 上直接调用 Lock/Unlock 方法。

package mainimport ( "fmt" "sync" )func main() { // 互斥锁保护计数器 var counter Counter // 辅助变量，用来确认所有的goroutine都完成 var wg sync.WaitGroup wg.Add(10)// 启动10个gourontine for i := 0; i < 10; i++ { go func() { defer wg.Done() // 累加10万次 for j := 0; j < 100000; j++ { counter.Lock() counter.Count++ counter.Unlock() } }() } wg.Wait() fmt.Println(counter.Count)return }type Counter struct { sync.Mutex Count uint64 }

如果嵌入的 struct 有多个字段，我们一般会把 Mutex 放在要控制的字段上面，然后使用空格把字段分隔开来。即使你不这样做，代码也可以正常编译，只不过，用这种风格去写的话，逻辑会更清晰，也更易于维护。
甚至，你还可以把获取锁、释放锁、计数加一的逻辑封装成一个方法，对外不需要暴露锁等逻辑：

package mainimport ( "fmt" "sync" )func main() { // 互斥锁保护计数器 var counter Counter // 辅助变量，用来确认所有的goroutine都完成 var wg sync.WaitGroup wg.Add(10)// 启动10个gourontine for i := 0; i < 10; i++ { go func() { defer wg.Done() // 累加10万次 for j := 0; j < 100000; j++ { counter.Incr() // 受到锁保护的方法 } }() } wg.Wait() fmt.Println(counter.Count)return }// 线程安全的计数器类型 type Counter struct { CounterType int Namestring musync.Mutex countuint64 }// 加1的方法，内部使用互斥锁保护 func (c *Counter) Incr() { c.mu.Lock() defer c.mu.Unlock() c.count++ }// 得到计数器的值，也需要锁保护 func (c *Counter) Count() uint64 { c.mu.Lock() defer c.mu.Unlock() return c.count}

总结本次介绍了并发问题的背景知识、标准库中 Mutex 的使用和最佳实践、通过 race detector 工具发现计数器程序的问题以及修复方法。相信你已经大致了解了 Mutex 这个同步原语。在项目开发的初始阶段，我们可能并没有仔细地考虑资源的并发问题，因为在初始阶段，我们还不确定这个资源是否被共享。经过更加深入的设计，或者新功能的增加、代码的完善，这个时候，我们就需要考虑共享资源的并发问题了。当然，如果你能在初始阶段预见到资源会被共享并发访问就更好了。
意识到共享资源的并发访问的早晚不重要，重要的是，一旦你意识到这个问题，你就要及时通过互斥锁等手段去解决。
比如 Docker issue 37583、35517、32826、30696等、kubernetes issue 72361、71617等，都是后来发现的 data race 而采用互斥锁 Mutex 进行修复的。
小试牛刀你已经知道，如果 Mutex 已经被一个 goroutine 获取了锁，其它等待中的 goroutine 们只能一直等待。那么，等这个锁释放后，等待中的 goroutine 中哪一个会优先获取 Mutex 呢？
解析：等待的goroutine们是以FIFO排队的,类似于图书馆借书排队
1）当Mutex处于正常模式时，若此时没有新goroutine与队头goroutine竞争，则队头goroutine获得。若有新goroutine竞争大概率新goroutine获得。
2）当队头goroutine竞争锁失败1ms后，它会将Mutex调整为饥饿模式。进入饥饿模式后，锁的所有权会直接从解锁goroutine移交给队头goroutine，此时新来的goroutine直接放入队尾。
3）当一个goroutine获取锁后，如果发现自己满足下列条件中的任何一个
#1它是队列中最后一个，#2它等待锁的时间少于1ms，则将锁切换回正常模式。