667 lines
26 KiB
Markdown
667 lines
26 KiB
Markdown
|
# 34｜并发：如何使用共享变量？
|
|||
|
|
|
|||
|
|
你好，我是Tony Bai。
|
|||
|
|
|
|||
|
|
在前面的讲解中，我们学习了Go的并发实现方案，知道了Go基于Tony Hoare的**CSP并发模型**理论，实现了Goroutine、channel等并发原语。
|
|||
|
|
|
|||
|
|
并且，Go语言之父Rob Pike还有一句经典名言：“不要通过共享内存来通信，应该通过通信来共享内存（Don’t communicate by sharing memory, share memory by communicating）”，这就奠定了Go应用并发设计的主流风格：**使用channel进行不同Goroutine间的通信**。
|
|||
|
|
|
|||
|
|
不过，Go也并没有彻底放弃基于共享内存的并发模型，而是在提供CSP并发模型原语的同时，还通过标准库的sync包，提供了针对传统的、基于共享内存并发模型的低级同步原语，包括：互斥锁（sync.Mutex）、读写锁（sync.RWMutex）、条件变量（sync.Cond）等，并通过atomic包提供了原子操作原语等等。显然，基于共享内存的并发模型在Go语言中依然有它的“用武之地”。
|
|||
|
|
|
|||
|
|
所以，在并发的最后一讲，我们就围绕sync包中的几个同步结构与对应的方法，聊聊基于共享内存的并发模型在Go中的应用。
|
|||
|
|
|
|||
|
|
我们先来看看在哪些场景下，我们需要用到sync包提供的低级同步原语。
|
|||
|
|
|
|||
|
|
## sync包低级同步原语可以用在哪？
|
|||
|
|
|
|||
|
|
这里我要先强调一句，一般情况下，我建议你优先使用CSP并发模型进行并发程序设计。但是在下面一些场景中，我们依然需要sync包提供的低级同步原语。
|
|||
|
|
|
|||
|
|
**首先是需要高性能的临界区（critical section）同步机制场景。**
|
|||
|
|
|
|||
|
|
在Go中，channel并发原语也可以用于对数据对象访问的同步，我们可以把channel看成是一种高级的同步原语，它自身的实现也是建构在低级同步原语之上的。也正因为如此，channel自身的性能与低级同步原语相比要略微逊色，开销要更大。
|
|||
|
|
|
|||
|
|
这里，关于sync.Mutex和channel各自实现的临界区同步机制，我做了一个简单的性能基准测试对比，通过对比结果，我们可以很容易看出两者的性能差异：
|
|||
|
|
|
|||
|
|
```plain
|
|||
|
|
var cs = 0 // 模拟临界区要保护的数据
|
|||
|
|
var mu sync.Mutex
|
|||
|
|
var c = make(chan struct{}, 1)
|
|||
|
|
|
|||
|
|
func criticalSectionSyncByMutex() {
|
|||
|
|
mu.Lock()
|
|||
|
|
cs++
|
|||
|
|
mu.Unlock()
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func criticalSectionSyncByChan() {
|
|||
|
|
c <- struct{}{}
|
|||
|
|
cs++
|
|||
|
|
<-c
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func BenchmarkCriticalSectionSyncByMutex(b *testing.B) {
|
|||
|
|
for n := 0; n < b.N; n++ {
|
|||
|
|
criticalSectionSyncByMutex()
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func BenchmarkCriticalSectionSyncByMutexInParallel(b *testing.B) {
|
|||
|
|
b.RunParallel(func(pb *testing.PB) {
|
|||
|
|
for pb.Next() {
|
|||
|
|
criticalSectionSyncByMutex()
|
|||
|
|
}
|
|||
|
|
})
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func BenchmarkCriticalSectionSyncByChan(b *testing.B) {
|
|||
|
|
for n := 0; n < b.N; n++ {
|
|||
|
|
criticalSectionSyncByChan()
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func BenchmarkCriticalSectionSyncByChanInParallel(b *testing.B) {
|
|||
|
|
b.RunParallel(func(pb *testing.PB) {
|
|||
|
|
for pb.Next() {
|
|||
|
|
criticalSectionSyncByChan()
|
|||
|
|
}
|
|||
|
|
})
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
|
|||
|
|
运行这个对比测试（Go 1.17），我们得到：
|
|||
|
|
|
|||
|
|
```plain
|
|||
|
|
$go test -bench .
|
|||
|
|
goos: darwin
|
|||
|
|
goarch: amd64
|
|||
|
|
... ...
|
|||
|
|
BenchmarkCriticalSectionSyncByMutex-8 88083549 13.64 ns/op
|
|||
|
|
BenchmarkCriticalSectionSyncByMutexInParallel-8 22337848 55.29 ns/op
|
|||
|
|
BenchmarkCriticalSectionSyncByChan-8 28172056 42.48 ns/op
|
|||
|
|
BenchmarkCriticalSectionSyncByChanInParallel-8 5722972 208.1 ns/op
|
|||
|
|
PASS
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
|
|||
|
|
通过这个对比实验，我们可以看到，无论是在单Goroutine情况下，还是在并发测试情况下，`sync.Mutex`实现的同步机制的性能，都要比channel实现的高出三倍多。
|
|||
|
|
|
|||
|
|
因此，通常在需要高性能的临界区（critical section）同步机制的情况下，sync包提供的低级同步原语更为适合。
|
|||
|
|
|
|||
|
|
**第二种就是在不想转移结构体对象所有权，但又要保证结构体内部状态数据的同步访问的场景。**
|
|||
|
|
|
|||
|
|
基于channel的并发设计，有一个特点：在Goroutine间通过channel转移数据对象的所有权。所以，只有拥有数据对象所有权（从channel接收到该数据）的Goroutine才可以对该数据对象进行状态变更。
|
|||
|
|
|
|||
|
|
如果你的设计中没有转移结构体对象所有权，但又要保证结构体内部状态数据在多个Goroutine之间同步访问，那么你可以使用sync包提供的低级同步原语来实现，比如最常用的`sync.Mutex`。
|
|||
|
|
|
|||
|
|
了解了这些应用场景之后，接着我们就来看看如何使用sync包中的各个同步结构，不过在使用之前，我们需要先看看一个sync包中同步原语使用的注意事项。
|
|||
|
|
|
|||
|
|
## sync包中同步原语使用的注意事项
|
|||
|
|
|
|||
|
|
在sync包的注释中（在`$GOROOT/src/sync/mutex.go`文件的头部注释），我们看到这样一行说明：
|
|||
|
|
|
|||
|
|
```plain
|
|||
|
|
// Values containing the types defined in this package should not be copied.
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
|
|||
|
|
翻译过来就是：“不应复制那些包含了此包中类型的值”。
|
|||
|
|
|
|||
|
|
在sync包的其他源文件中，我们同样看到类似的一些注释：
|
|||
|
|
|
|||
|
|
```plain
|
|||
|
|
|
|||
|
|
// $GOROOT/src/sync/mutex.go
|
|||
|
|
// A Mutex must not be copied after first use. （禁止复制首次使用后的Mutex）
|
|||
|
|
|
|||
|
|
// $GOROOT/src/sync/rwmutex.go
|
|||
|
|
// A RWMutex must not be copied after first use.（禁止复制首次使用后的RWMutex）
|
|||
|
|
|
|||
|
|
// $GOROOT/src/sync/cond.go
|
|||
|
|
// A Cond must not be copied after first use.（禁止复制首次使用后的Cond）
|
|||
|
|
... ...
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
|
|||
|
|
那么，为什么首次使用Mutex等sync包中定义的结构类型后，我们不应该再对它们进行复制操作呢？我们以Mutex这个同步原语为例，看看它的实现是怎样的。
|
|||
|
|
|
|||
|
|
Go标准库中sync.Mutex的定义是这样的：
|
|||
|
|
|
|||
|
|
```plain
|
|||
|
|
// $GOROOT/src/sync/mutex.go
|
|||
|
|
type Mutex struct {
|
|||
|
|
state int32
|
|||
|
|
sema uint32
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
|
|||
|
|
我们看到，Mutex的定义非常简单，由两个整型字段state和sema组成：
|
|||
|
|
|
|||
|
|
* state：表示当前互斥锁的状态；
|
|||
|
|
* sema：用于控制锁状态的信号量。
|
|||
|
|
|
|||
|
|
初始情况下，Mutex的实例处于**Unlocked**状态（state和sema均为0）。对Mutex实例的复制也就是两个整型字段的复制。一旦发生复制，原变量与副本就是两个单独的内存块，各自发挥同步作用，互相就没有了关联。
|
|||
|
|
|
|||
|
|
如果发生复制后，你仍然认为原变量与副本保护的是同一个数据对象，那可就大错特错了。我们来看一个例子：
|
|||
|
|
|
|||
|
|
```plain
|
|||
|
|
func main() {
|
|||
|
|
var wg sync.WaitGroup
|
|||
|
|
i := 0
|
|||
|
|
var mu sync.Mutex // 负责对i的同步访问
|
|||
|
|
|
|||
|
|
wg.Add(1)
|
|||
|
|
// g1
|
|||
|
|
go func(mu1 sync.Mutex) {
|
|||
|
|
mu1.Lock()
|
|||
|
|
i = 10
|
|||
|
|
time.Sleep(10 * time.Second)
|
|||
|
|
fmt.Printf("g1: i = %d\n", i)
|
|||
|
|
mu1.Unlock()
|
|||
|
|
wg.Done()
|
|||
|
|
}(mu)
|
|||
|
|
|
|||
|
|
time.Sleep(time.Second)
|
|||
|
|
|
|||
|
|
mu.Lock()
|
|||
|
|
i = 1
|
|||
|
|
fmt.Printf("g0: i = %d\n", i)
|
|||
|
|
mu.Unlock()
|
|||
|
|
|
|||
|
|
wg.Wait()
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
|
|||
|
|
在这个例子中，我们使用一个sync.Mutex类型变量mu来同步对整型变量i的访问。我们创建一个新Goroutine：g1，g1通过函数参数得到mu的一份拷贝mu1，然后g1会通过mu1来同步对整型变量i的访问。
|
|||
|
|
|
|||
|
|
那么，g0通过mu和g1通过mu的拷贝mu1，是否能实现对同一个变量i的同步访问呢？我们来看看运行这个示例的运行结果：
|
|||
|
|
|
|||
|
|
```plain
|
|||
|
|
g0: i = 1
|
|||
|
|
g1: i = 1
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
|
|||
|
|
从结果来看，这个程序并没有实现对i的同步访问，第9行g1对mu1的加锁操作，并没能阻塞第19行g0对mu的加锁。于是，g1刚刚将i赋值为10后，g0就又将i赋值为1了。
|
|||
|
|
|
|||
|
|
出现这种结果的原因就是我们前面分析的情况，一旦Mutex类型变量被拷贝，原变量与副本就各自发挥作用，互相没有关联了。甚至，如果拷贝的时机不对，比如在一个mutex处于locked的状态时对它进行了拷贝，就会对副本进行加锁操作，将导致加锁的Goroutine永远阻塞下去。
|
|||
|
|
|
|||
|
|
通过前面这个例子，我们可以很直观地看到：如果对使用过的、sync包中的类型的示例进行复制，并使用了复制后得到的副本，将导致不可预期的结果。所以，在使用sync包中的类型的时候，我们推荐通过**闭包**方式，或者是**传递类型实例（或包裹该类型的类型实例）的地址（指针）**的方式进行。这就是使用sync包时最值得我们注意的事项。
|
|||
|
|
|
|||
|
|
接下来，我们就来逐个分析日常使用较多的sync包中同步原语。我们先来看看互斥锁与读写锁。
|
|||
|
|
|
|||
|
|
## 互斥锁（Mutex）还是读写锁（RWMutex）？
|
|||
|
|
|
|||
|
|
sync包提供了两种用于临界区同步的原语：互斥锁（Mutex）和读写锁（RWMutex）。它们都是零值可用的数据类型，也就是不需要显式初始化就可以使用，并且使用方法都比较简单。在上面的示例中，我们已经看到了Mutex的应用方法，这里再总结一下：
|
|||
|
|
|
|||
|
|
```plain
|
|||
|
|
var mu sync.Mutex
|
|||
|
|
mu.Lock() // 加锁
|
|||
|
|
doSomething()
|
|||
|
|
mu.Unlock() // 解锁
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
|
|||
|
|
一旦某个Goroutine调用的Mutex执行Lock操作成功，它将成功持有这把互斥锁。这个时候，如果有其他Goroutine执行Lock操作，就会阻塞在这把互斥锁上，直到持有这把锁的Goroutine调用Unlock释放掉这把锁后，才会抢到这把锁的持有权并进入临界区。
|
|||
|
|
|
|||
|
|
由此，我们也可以得到使用互斥锁的两个原则：
|
|||
|
|
|
|||
|
|
* **尽量减少在锁中的操作**。这可以减少其他因Goroutine阻塞而带来的损耗与延迟。
|
|||
|
|
* **一定要记得调用Unlock解锁**。忘记解锁会导致程序局部死锁，甚至是整个程序死锁，会导致严重的后果。同时，我们也可以结合第23讲学习到的defer，优雅地执行解锁操作。
|
|||
|
|
|
|||
|
|
读写锁与互斥锁用法大致相同，只不过多了一组加读锁和解读锁的方法：
|
|||
|
|
|
|||
|
|
```plain
|
|||
|
|
var rwmu sync.RWMutex
|
|||
|
|
rwmu.RLock() //加读锁
|
|||
|
|
readSomething()
|
|||
|
|
rwmu.RUnlock() //解读锁
|
|||
|
|
rwmu.Lock() //加写锁
|
|||
|
|
changeSomething()
|
|||
|
|
rwmu.Unlock() //解写锁
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
|
|||
|
|
写锁与Mutex的行为十分类似，一旦某Goroutine持有写锁，其他Goroutine无论是尝试加读锁，还是加写锁，都会被阻塞在写锁上。
|
|||
|
|
|
|||
|
|
但读锁就宽松多了，一旦某个Goroutine持有读锁，它不会阻塞其他尝试加读锁的Goroutine，但加写锁的Goroutine依然会被阻塞住。
|
|||
|
|
|
|||
|
|
通常，**互斥锁（Mutex）是临时区同步原语的首选**，它常被用来对结构体对象的内部状态、缓存等进行保护，是使用最为广泛的临界区同步原语。相比之下，读写锁的应用就没那么广泛了，只活跃于它擅长的场景下。
|
|||
|
|
|
|||
|
|
那读写锁（RWMutex）究竟擅长在哪种场景下呢？我们先来看一组基准测试：
|
|||
|
|
|
|||
|
|
```plain
|
|||
|
|
var cs1 = 0 // 模拟临界区要保护的数据
|
|||
|
|
var mu1 sync.Mutex
|
|||
|
|
|
|||
|
|
var cs2 = 0 // 模拟临界区要保护的数据
|
|||
|
|
var mu2 sync.RWMutex
|
|||
|
|
|
|||
|
|
func BenchmarkWriteSyncByMutex(b *testing.B) {
|
|||
|
|
b.RunParallel(func(pb *testing.PB) {
|
|||
|
|
for pb.Next() {
|
|||
|
|
mu1.Lock()
|
|||
|
|
cs1++
|
|||
|
|
mu1.Unlock()
|
|||
|
|
}
|
|||
|
|
})
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func BenchmarkReadSyncByMutex(b *testing.B) {
|
|||
|
|
b.RunParallel(func(pb *testing.PB) {
|
|||
|
|
for pb.Next() {
|
|||
|
|
mu1.Lock()
|
|||
|
|
_ = cs1
|
|||
|
|
mu1.Unlock()
|
|||
|
|
}
|
|||
|
|
})
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func BenchmarkReadSyncByRWMutex(b *testing.B) {
|
|||
|
|
b.RunParallel(func(pb *testing.PB) {
|
|||
|
|
for pb.Next() {
|
|||
|
|
mu2.RLock()
|
|||
|
|
_ = cs2
|
|||
|
|
mu2.RUnlock()
|
|||
|
|
}
|
|||
|
|
})
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func BenchmarkWriteSyncByRWMutex(b *testing.B) {
|
|||
|
|
b.RunParallel(func(pb *testing.PB) {
|
|||
|
|
for pb.Next() {
|
|||
|
|
mu2.Lock()
|
|||
|
|
cs2++
|
|||
|
|
mu2.Unlock()
|
|||
|
|
}
|
|||
|
|
})
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
|
|||
|
|
这些基准测试都是并发测试，度量的是Mutex、RWMutex在并发下的读写性能。我们分别在cpu=2、8、16、32的情况下运行这个并发性能测试，测试结果如下：
|
|||
|
|
|
|||
|
|
```plain
|
|||
|
|
goos: darwin
|
|||
|
|
goarch: amd64
|
|||
|
|
... ...
|
|||
|
|
BenchmarkWriteSyncByMutex-2 73423770 16.12 ns/op
|
|||
|
|
BenchmarkReadSyncByMutex-2 84031135 15.08 ns/op
|
|||
|
|
BenchmarkReadSyncByRWMutex-2 37182219 31.87 ns/op
|
|||
|
|
BenchmarkWriteSyncByRWMutex-2 40727782 29.08 ns/op
|
|||
|
|
|
|||
|
|
BenchmarkWriteSyncByMutex-8 22153354 56.39 ns/op
|
|||
|
|
BenchmarkReadSyncByMutex-8 24164278 51.12 ns/op
|
|||
|
|
BenchmarkReadSyncByRWMutex-8 38589122 31.17 ns/op
|
|||
|
|
BenchmarkWriteSyncByRWMutex-8 18482208 65.27 ns/op
|
|||
|
|
|
|||
|
|
BenchmarkWriteSyncByMutex-16 20672842 62.94 ns/op
|
|||
|
|
BenchmarkReadSyncByMutex-16 19247158 62.94 ns/op
|
|||
|
|
BenchmarkReadSyncByRWMutex-16 29978614 39.98 ns/op
|
|||
|
|
BenchmarkWriteSyncByRWMutex-16 16095952 78.19 ns/op
|
|||
|
|
|
|||
|
|
BenchmarkWriteSyncByMutex-32 20539290 60.20 ns/op
|
|||
|
|
BenchmarkReadSyncByMutex-32 18807060 72.61 ns/op
|
|||
|
|
BenchmarkReadSyncByRWMutex-32 29772936 40.45 ns/op
|
|||
|
|
BenchmarkWriteSyncByRWMutex-32 13320544 86.53 ns/op
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
|
|||
|
|
通过测试结果对比，我们得到了一些结论：
|
|||
|
|
|
|||
|
|
* 并发量较小的情况下，Mutex性能最好；随着并发量增大，Mutex的竞争激烈，导致加锁和解锁性能下降；
|
|||
|
|
* RWMutex的读锁性能并没有随着并发量的增大，而发生较大变化，性能始终恒定在40ns左右；
|
|||
|
|
* 在并发量较大的情况下，RWMutex的写锁性能和Mutex、RWMutex读锁相比，是最差的，并且随着并发量增大，RWMutex写锁性能有继续下降趋势。
|
|||
|
|
|
|||
|
|
由此，我们就可以看出，**读写锁适合应用在具有一定并发量且读多写少的场合**。在大量并发读的情况下，多个Goroutine可以同时持有读锁，从而减少在锁竞争中等待的时间。
|
|||
|
|
|
|||
|
|
而互斥锁，即便是读请求的场合，同一时刻也只能有一个Goroutine持有锁，其他Goroutine只能阻塞在加锁操作上等待被调度。
|
|||
|
|
|
|||
|
|
接下来，我们继续看条件变量sync.Cond。
|
|||
|
|
|
|||
|
|
## 条件变量
|
|||
|
|
|
|||
|
|
`sync.Cond`是传统的条件变量原语概念在Go语言中的实现。我们可以把一个条件变量理解为一个容器，这个容器中存放着一个或一组等待着某个条件成立的Goroutine。当条件成立后，这些处于等待状态的Goroutine将得到通知，并被唤醒继续进行后续的工作。这与百米飞人大战赛场上，各位运动员等待裁判员的发令枪声的情形十分类似。
|
|||
|
|
|
|||
|
|
条件变量是同步原语的一种，如果没有条件变量，开发人员可能需要在Goroutine中通过连续轮询的方式，检查某条件是否为真，这种连续轮询非常消耗资源，因为Goroutine在这个过程中是处于活动状态的，但它的工作又没有进展。
|
|||
|
|
|
|||
|
|
这里我们先看一个用`sync.Mutex` 实现对条件轮询等待的例子：
|
|||
|
|
|
|||
|
|
```go
|
|||
|
|
type signal struct{}
|
|||
|
|
|
|||
|
|
var ready bool
|
|||
|
|
|
|||
|
|
func worker(i int) {
|
|||
|
|
fmt.Printf("worker %d: is working...\n", i)
|
|||
|
|
time.Sleep(1 * time.Second)
|
|||
|
|
fmt.Printf("worker %d: works done\n", i)
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func spawnGroup(f func(i int), num int, mu *sync.Mutex) <-chan signal {
|
|||
|
|
c := make(chan signal)
|
|||
|
|
var wg sync.WaitGroup
|
|||
|
|
|
|||
|
|
for i := 0; i < num; i++ {
|
|||
|
|
wg.Add(1)
|
|||
|
|
go func(i int) {
|
|||
|
|
for {
|
|||
|
|
mu.Lock()
|
|||
|
|
if !ready {
|
|||
|
|
mu.Unlock()
|
|||
|
|
time.Sleep(100 * time.Millisecond)
|
|||
|
|
continue
|
|||
|
|
}
|
|||
|
|
mu.Unlock()
|
|||
|
|
fmt.Printf("worker %d: start to work...\n", i)
|
|||
|
|
f(i)
|
|||
|
|
wg.Done()
|
|||
|
|
return
|
|||
|
|
}
|
|||
|
|
}(i + 1)
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
go func() {
|
|||
|
|
wg.Wait()
|
|||
|
|
c <- signal(struct{}{})
|
|||
|
|
}()
|
|||
|
|
return c
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func main() {
|
|||
|
|
fmt.Println("start a group of workers...")
|
|||
|
|
mu := &sync.Mutex{}
|
|||
|
|
c := spawnGroup(worker, 5, mu)
|
|||
|
|
|
|||
|
|
time.Sleep(5 * time.Second) // 模拟ready前的准备工作
|
|||
|
|
fmt.Println("the group of workers start to work...")
|
|||
|
|
|
|||
|
|
mu.Lock()
|
|||
|
|
ready = true
|
|||
|
|
mu.Unlock()
|
|||
|
|
|
|||
|
|
<-c
|
|||
|
|
fmt.Println("the group of workers work done!")
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
|
|||
|
|
就像前面提到的，轮询的方式开销大，轮询间隔设置的不同，条件检查的及时性也会受到影响。
|
|||
|
|
`sync.Cond`为Goroutine在这个场景下提供了另一种可选的、资源消耗更小、使用体验更佳的同步方式。使用条件变量原语，我们可以在实现相同目标的同时，避免对条件的轮询。
|
|||
|
|
|
|||
|
|
我们用`sync.Cond`对上面的例子进行改造，改造后的代码如下：
|
|||
|
|
|
|||
|
|
```plain
|
|||
|
|
type signal struct{}
|
|||
|
|
|
|||
|
|
var ready bool
|
|||
|
|
|
|||
|
|
func worker(i int) {
|
|||
|
|
fmt.Printf("worker %d: is working...\n", i)
|
|||
|
|
time.Sleep(1 * time.Second)
|
|||
|
|
fmt.Printf("worker %d: works done\n", i)
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func spawnGroup(f func(i int), num int, groupSignal *sync.Cond) <-chan signal {
|
|||
|
|
c := make(chan signal)
|
|||
|
|
var wg sync.WaitGroup
|
|||
|
|
|
|||
|
|
for i := 0; i < num; i++ {
|
|||
|
|
wg.Add(1)
|
|||
|
|
go func(i int) {
|
|||
|
|
groupSignal.L.Lock()
|
|||
|
|
for !ready {
|
|||
|
|
groupSignal.Wait()
|
|||
|
|
}
|
|||
|
|
groupSignal.L.Unlock()
|
|||
|
|
fmt.Printf("worker %d: start to work...\n", i)
|
|||
|
|
f(i)
|
|||
|
|
wg.Done()
|
|||
|
|
}(i + 1)
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
go func() {
|
|||
|
|
wg.Wait()
|
|||
|
|
c <- signal(struct{}{})
|
|||
|
|
}()
|
|||
|
|
return c
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func main() {
|
|||
|
|
fmt.Println("start a group of workers...")
|
|||
|
|
groupSignal := sync.NewCond(&sync.Mutex{})
|
|||
|
|
c := spawnGroup(worker, 5, groupSignal)
|
|||
|
|
|
|||
|
|
time.Sleep(5 * time.Second) // 模拟ready前的准备工作
|
|||
|
|
fmt.Println("the group of workers start to work...")
|
|||
|
|
|
|||
|
|
groupSignal.L.Lock()
|
|||
|
|
ready = true
|
|||
|
|
groupSignal.Broadcast()
|
|||
|
|
groupSignal.L.Unlock()
|
|||
|
|
|
|||
|
|
<-c
|
|||
|
|
fmt.Println("the group of workers work done!")
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
|
|||
|
|
我们运行这个示例程序，得到：
|
|||
|
|
|
|||
|
|
```plain
|
|||
|
|
start a group of workers...
|
|||
|
|
the group of workers start to work...
|
|||
|
|
worker 2: start to work...
|
|||
|
|
worker 2: is working...
|
|||
|
|
worker 3: start to work...
|
|||
|
|
worker 3: is working...
|
|||
|
|
worker 1: start to work...
|
|||
|
|
worker 1: is working...
|
|||
|
|
worker 4: start to work...
|
|||
|
|
worker 5: start to work...
|
|||
|
|
worker 5: is working...
|
|||
|
|
worker 4: is working...
|
|||
|
|
worker 4: works done
|
|||
|
|
worker 2: works done
|
|||
|
|
worker 3: works done
|
|||
|
|
worker 1: works done
|
|||
|
|
worker 5: works done
|
|||
|
|
the group of workers work done!
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
|
|||
|
|
我们看到，`sync.Cond`实例的初始化，需要一个满足实现了`sync.Locker`接口的类型实例，通常我们使用`sync.Mutex`。
|
|||
|
|
|
|||
|
|
条件变量需要这个互斥锁来同步临界区，保护用作条件的数据。加锁后，各个等待条件成立的Goroutine判断条件是否成立，如果不成立，则调用`sync.Cond`的Wait方法进入等待状态。Wait方法在Goroutine挂起前会进行Unlock操作。
|
|||
|
|
|
|||
|
|
当main goroutine将`ready`置为true，并调用`sync.Cond`的Broadcast方法后，各个阻塞的Goroutine将被唤醒，并从Wait方法中返回。Wait方法返回前，Wait方法会再次加锁让Goroutine进入临界区。接下来Goroutine会再次对条件数据进行判定，如果条件成立，就会解锁并进入下一个工作阶段；如果条件依旧不成立，那么会再次进入循环体，并调用Wait方法挂起等待。
|
|||
|
|
|
|||
|
|
和`sync.Mutex` 、`sync.RWMutex`等相比，`sync.Cond` 应用的场景更为有限，只有在需要“等待某个条件成立”的场景下，Cond才有用武之地。
|
|||
|
|
|
|||
|
|
其实，面向CSP并发模型的channel原语和面向传统共享内存并发模型的sync包提供的原语，已经能够满足Go语言应用并发设计中99.9%的并发同步需求了。而剩余那0.1%的需求，我们可以使用Go标准库提供的atomic包来实现。
|
|||
|
|
|
|||
|
|
## 原子操作（atomic operations）
|
|||
|
|
|
|||
|
|
atomic包是Go语言给用户提供的原子操作原语的相关接口。原子操作（atomic operations）是相对于普通指令操作而言的。
|
|||
|
|
|
|||
|
|
我们以一个整型变量自增的语句为例说明一下：
|
|||
|
|
|
|||
|
|
```plain
|
|||
|
|
var a int
|
|||
|
|
a++
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
|
|||
|
|
a++这行语句需要3条普通机器指令来完成变量a的自增：
|
|||
|
|
|
|||
|
|
* LOAD：将变量从内存加载到CPU寄存器；
|
|||
|
|
* ADD：执行加法指令；
|
|||
|
|
* STORE：将结果存储回原内存地址中。
|
|||
|
|
|
|||
|
|
这3条普通指令在执行过程中是可以被中断的。而原子操作的指令是不可中断的，它就好比一个事务，要么不执行，一旦执行就一次性全部执行完毕，中间不可分割。也正因为如此，原子操作也可以被用于共享数据的并发同步。
|
|||
|
|
|
|||
|
|
原子操作由底层硬件直接提供支持，是一种硬件实现的指令级的“事务”，因此相对于操作系统层面和Go运行时层面提供的同步技术而言，它更为原始。
|
|||
|
|
|
|||
|
|
atomic包封装了CPU实现的部分原子操作指令，为用户层提供体验良好的原子操作函数，因此atomic包中提供的原语更接近硬件底层，也更为低级，它也常被用于实现更为高级的并发同步技术，比如channel和sync包中的同步原语。
|
|||
|
|
|
|||
|
|
我们以atomic.SwapInt64函数在x86\_64平台上的实现为例，看看这个函数的实现方法：
|
|||
|
|
|
|||
|
|
```go
|
|||
|
|
// $GOROOT/src/sync/atomic/doc.go
|
|||
|
|
func SwapInt64(addr *int64, new int64) (old int64)
|
|||
|
|
|
|||
|
|
// $GOROOT/src/sync/atomic/asm.s
|
|||
|
|
|
|||
|
|
TEXT ·SwapInt64(SB),NOSPLIT,$0
|
|||
|
|
JMP runtime∕internal∕atomic·Xchg64(SB)
|
|||
|
|
|
|||
|
|
// $GOROOT/src/runtime/internal/atomic/asm_amd64.s
|
|||
|
|
TEXT runtime∕internal∕atomic·Xchg64(SB), NOSPLIT, $0-24
|
|||
|
|
MOVQ ptr+0(FP), BX
|
|||
|
|
MOVQ new+8(FP), AX
|
|||
|
|
XCHGQ AX, 0(BX)
|
|||
|
|
MOVQ AX, ret+16(FP)
|
|||
|
|
RET
|
|||
|
|
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
|
|||
|
|
从函数SwapInt64的实现中，我们可以看到：它基本就是对x86\_64 CPU实现的原子操作指令`XCHGQ`的直接封装。
|
|||
|
|
|
|||
|
|
原子操作的特性，让atomic包也可以被用作对共享数据的并发同步，那么和更为高级的channel以及sync包中原语相比，我们究竟该怎么选择呢？
|
|||
|
|
|
|||
|
|
我们先来看看atomic包提供了哪些能力。
|
|||
|
|
|
|||
|
|
atomic包提供了两大类原子操作接口，一类是针对整型变量的，包括有符号整型、无符号整型以及对应的指针类型；另外一类是针对自定义类型的。因此，第一类原子操作接口的存在让atomic包天然适合去实现某一个共享整型变量的并发同步。
|
|||
|
|
|
|||
|
|
我们再看一个例子：
|
|||
|
|
|
|||
|
|
```go
|
|||
|
|
var n1 int64
|
|||
|
|
|
|||
|
|
func addSyncByAtomic(delta int64) int64 {
|
|||
|
|
return atomic.AddInt64(&n1, delta)
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func readSyncByAtomic() int64 {
|
|||
|
|
return atomic.LoadInt64(&n1)
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
var n2 int64
|
|||
|
|
var rwmu sync.RWMutex
|
|||
|
|
|
|||
|
|
func addSyncByRWMutex(delta int64) {
|
|||
|
|
rwmu.Lock()
|
|||
|
|
n2 += delta
|
|||
|
|
rwmu.Unlock()
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func readSyncByRWMutex() int64 {
|
|||
|
|
var n int64
|
|||
|
|
rwmu.RLock()
|
|||
|
|
n = n2
|
|||
|
|
rwmu.RUnlock()
|
|||
|
|
return n
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func BenchmarkAddSyncByAtomic(b *testing.B) {
|
|||
|
|
b.RunParallel(func(pb *testing.PB) {
|
|||
|
|
for pb.Next() {
|
|||
|
|
addSyncByAtomic(1)
|
|||
|
|
}
|
|||
|
|
})
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func BenchmarkReadSyncByAtomic(b *testing.B) {
|
|||
|
|
b.RunParallel(func(pb *testing.PB) {
|
|||
|
|
for pb.Next() {
|
|||
|
|
readSyncByAtomic()
|
|||
|
|
}
|
|||
|
|
})
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func BenchmarkAddSyncByRWMutex(b *testing.B) {
|
|||
|
|
b.RunParallel(func(pb *testing.PB) {
|
|||
|
|
for pb.Next() {
|
|||
|
|
addSyncByRWMutex(1)
|
|||
|
|
}
|
|||
|
|
})
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
func BenchmarkReadSyncByRWMutex(b *testing.B) {
|
|||
|
|
b.RunParallel(func(pb *testing.PB) {
|
|||
|
|
for pb.Next() {
|
|||
|
|
readSyncByRWMutex()
|
|||
|
|
}
|
|||
|
|
})
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
|
|||
|
|
我们分别在cpu=2、 8、16、32的情况下运行上述性能基准测试，得到结果如下：
|
|||
|
|
|
|||
|
|
```go
|
|||
|
|
goos: darwin
|
|||
|
|
goarch: amd64
|
|||
|
|
... ...
|
|||
|
|
BenchmarkAddSyncByAtomic-2 75426774 17.69 ns/op
|
|||
|
|
BenchmarkReadSyncByAtomic-2 1000000000 0.7437 ns/op
|
|||
|
|
BenchmarkAddSyncByRWMutex-2 39041671 30.16 ns/op
|
|||
|
|
BenchmarkReadSyncByRWMutex-2 41325093 28.48 ns/op
|
|||
|
|
|
|||
|
|
BenchmarkAddSyncByAtomic-8 77497987 15.25 ns/op
|
|||
|
|
BenchmarkReadSyncByAtomic-8 1000000000 0.2395 ns/op
|
|||
|
|
BenchmarkAddSyncByRWMutex-8 17702034 67.16 ns/op
|
|||
|
|
BenchmarkReadSyncByRWMutex-8 29966182 40.37 ns/op
|
|||
|
|
|
|||
|
|
BenchmarkAddSyncByAtomic-16 57727968 20.39 ns/op
|
|||
|
|
BenchmarkReadSyncByAtomic-16 1000000000 0.2536 ns/op
|
|||
|
|
BenchmarkAddSyncByRWMutex-16 15029635 78.61 ns/op
|
|||
|
|
BenchmarkReadSyncByRWMutex-16 29722464 40.28 ns/op
|
|||
|
|
|
|||
|
|
BenchmarkAddSyncByAtomic-32 58010497 20.40 ns/op
|
|||
|
|
BenchmarkReadSyncByAtomic-32 1000000000 0.2402 ns/op
|
|||
|
|
BenchmarkAddSyncByRWMutex-32 11748312 93.15 ns/op
|
|||
|
|
BenchmarkReadSyncByRWMutex-32 29845912 40.54 ns/op
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
|
|||
|
|
通过这个运行结果，我们可以得出一些结论：
|
|||
|
|
|
|||
|
|
* 读写锁的性能随着并发量增大的情况，与前面讲解的sync.RWMutex一致；
|
|||
|
|
* 利用原子操作的无锁并发写的性能，随着并发量增大几乎保持恒定；
|
|||
|
|
* 利用原子操作的无锁并发读的性能，随着并发量增大有持续提升的趋势，并且性能是读锁的约200倍。
|
|||
|
|
|
|||
|
|
通过这些结论，我们大致可以看到atomic原子操作的特性：随着并发量提升，使用atomic实现的**共享变量**的并发读写性能表现更为稳定，尤其是原子读操作，和sync包中的读写锁原语比起来，atomic表现出了更好的伸缩性和高性能。
|
|||
|
|
|
|||
|
|
由此，我们也可以看出atomic包更适合**一些对性能十分敏感、并发量较大且读多写少的场合**。
|
|||
|
|
|
|||
|
|
不过，atomic原子操作可用来同步的范围有比较大限制，只能同步一个整型变量或自定义类型变量。如果我们要对一个复杂的临界区数据进行同步，那么首选的依旧是sync包中的原语。
|
|||
|
|
|
|||
|
|
## 小结
|
|||
|
|
|
|||
|
|
好了，今天的课讲到这里就结束了，现在我们一起来回顾一下吧。
|
|||
|
|
|
|||
|
|
虽然Go推荐基于通信来共享内存的并发设计风格，但Go并没有彻底抛弃对基于共享内存并发模型的支持，Go通过标准库的sync包以及atomic包提供了低级同步原语。这些原语有着它们自己的应用场景。
|
|||
|
|
|
|||
|
|
如果我们考虑使用低级同步原语，一般都是因为低级同步原语可以提供**更佳的性能表现**，性能基准测试结果告诉我们，使用低级同步原语的性能可以高出channel许多倍。在性能敏感的场景下，我们依然离不开这些低级同步原语。
|
|||
|
|
|
|||
|
|
在使用sync包提供的同步原语之前，我们一定要牢记这些原语使用的注意事项：**不要复制首次使用后的Mutex/RWMutex/Cond等**。一旦复制，你将很大可能得到意料之外的运行结果。
|
|||
|
|
|
|||
|
|
sync包中的低级同步原语各有各的擅长领域，你可以记住：
|
|||
|
|
|
|||
|
|
* 在具有一定并发量且读多写少的场合使用RWMutex；
|
|||
|
|
* 在需要“等待某个条件成立”的场景下使用Cond；
|
|||
|
|
* 当你不确定使用什么原语时，那就使用Mutex吧。
|
|||
|
|
|
|||
|
|
如果你对同步的性能有极致要求，且并发量较大，读多写少，那么可以考虑一下atomic包提供的原子操作函数。
|
|||
|
|
|
|||
|
|
## 思考题
|
|||
|
|
|
|||
|
|
使用基于共享内存的并发模型时，最令人头疼的可能就是“死锁”问题的存在了。你了解死锁的产生条件么？能编写一个程序模拟一下死锁的发生么？
|
|||
|
|
|
|||
|
|
欢迎你把这节课分享给更多对Go并发感兴趣的朋友。我是Tony Bai，我们下节课见。
|
|||
|
|
|