164 lines
6.6 KiB
Markdown
164 lines
6.6 KiB
Markdown
---
|
||
title: 小记Go里常用的并发控制手段（二）
|
||
tags:
|
||
- Go
|
||
- Golang
|
||
- 并发控制
|
||
- Mutex
|
||
- Lock
|
||
- Sync
|
||
- WaitGroup
|
||
- Cond
|
||
categories:
|
||
- - Go
|
||
- 应用技巧
|
||
keywords: 'go,并发,mutex,lock,sync,控制,cond,waitgroup'
|
||
date: 2022-09-28 13:43:51
|
||
---
|
||
|
||
传统的并发控制在Go的标准库中也是有提供的，而且使用起来也非常简单。但是需要注意的是，这些传统的控制手段在使用的时候，同样会面临传统并发编程中会遇到的所有挑战。<!--more-->对于这些传统的并发控制手段，主要都是由标准库中的`sync`包提供的。`sync`包中提供的主要类型有`Locker`、`Cond`、`Map`、`Mutex`、`Once`、`Pool`、`RWMutex`和`WaitGroup`。其中可能有不少类型的名字听起来都十分的熟悉，这里只选其中比较常用的几个来记录。
|
||
|
||
## `Locker`
|
||
|
||
`Locker`是`sync`包中提供的一个接口，主要用来表示提供了`Lock()`和`Unlock()`方法的可以用来在并发编程中执行锁的功能的对象，例如`sync`包中提供的`Mutex`类型、`RWmutex`类型等。
|
||
|
||
## `Once`
|
||
|
||
`Once`是一个可以被多次调用但是只会执行一次的对象，不论每次传入的参数是否相同，`Once`就是只会执行一次。例如以下这个示例。
|
||
|
||
```go
|
||
func main() {
|
||
var once sync.Once
|
||
fnBody := func() {
|
||
// 执行一些只需要执行一次的方法
|
||
}
|
||
|
||
for i := 0; i < 10; i++ {
|
||
go func() {
|
||
once.Do(fnBody)
|
||
}()
|
||
}
|
||
|
||
// 继续其他的功能
|
||
}
|
||
```
|
||
|
||
!!! note ""
|
||
`Once`对象也适合用来创建单例对象使用。
|
||
|
||
## `Mutex`与`RWMutex`
|
||
|
||
`Mutex`是经典的互斥锁实现，互斥锁在刚被创建的时候是未锁闭状态，而且在使用的时候，一定注意要使用引用的形式（指针）传递互斥锁的实例，否则互斥锁将失去其并发控制功能。`Mutex`类型实现了`Locker`接口，其提供的方法主要可以实现以下功能。
|
||
|
||
- `Lock()`，这是`Locker`接口规定必须要实现的功能，当调用`Lock()`的时候，会尝试获取锁并锁闭，如果不能成功获取到锁并锁闭，那么将会阻塞当前的协程直到成功获取到锁为止。
|
||
- `TryLock()`，用于尝试获取锁，但是在未获取到锁的时候，并不会阻塞当前的协程，而是会发回一个`false`返回值。
|
||
- `Unlock()`，解锁并释放当前获取到的锁。互斥锁被解锁并释放以后，就可以被其他的协程所获取并锁闭。
|
||
|
||
在大部分并发编程中，并不推荐直接使用`Mutex`这种偏底层的并发控制。
|
||
|
||
从`Mutex`类型引申出来的类型是`RWMutex`，即读写锁，主要用于单写多读的情况下。因为大部分的资源在并发条件下，只有写操作是互斥的（影响值的一致性），但是读操作是不互斥的。所以读写锁提供了写锁和读锁两种锁，对应的也提供了两种锁的操作方法。
|
||
|
||
- `Lock()`和`Unlock()`用于对写操作进行加锁和解锁，调用`Lock()`会导致读和写都被锁闭。
|
||
- `RLock()`和`RUnlock()`用于对读操作进行加锁和解锁，调用`RLock()`会导致写操作被锁闭，但是不会影响其他的协程继续调用`RLock()`。`RUnlock()`只会解锁一次`RLock()`，不会影响其他剩余的`RLock()`调用。资源必须在所有的`RLock()`都被解锁以后，才能进行写操作。
|
||
|
||
!!! warning ""
|
||
如果调用`RUnlock()`的次数多于调用`RLock()`的次数，程序将会panic。如果多次锁闭但是不进行解锁，将会导致死锁，所以在使用的时候，锁闭与解锁的操作必须是对称的，并且是可以抵达的。
|
||
|
||
以下是`RWMutex`的一个简单使用示例。
|
||
|
||
```go
|
||
var (
|
||
rwm sync.RWMutex
|
||
val int
|
||
)
|
||
|
||
for i := 0; i < 10; i++ {
|
||
go func() {
|
||
rwm.RLock()
|
||
|
||
fmt.Printf("%d\n", val)
|
||
|
||
rwm.RUnlock()
|
||
time.Sleep(rand.Int(20) * time.Second)
|
||
}()
|
||
}
|
||
|
||
for i := 0; i <= 60; i++ {
|
||
rwm.Lock()
|
||
|
||
val = i
|
||
|
||
rwm.Unlock()
|
||
time.Sleep(1 * time.Second)
|
||
}
|
||
```
|
||
|
||
## `Cond`
|
||
|
||
`Cond`是创建一个条件变量来对协程进行控制，所有受控的协程都会集结在这个条件变量的位置上等待调度。举例来说就是多个协程在等待，一个协程在发送通知事件，这种情况通常出现在多个协程在等待资源准备就绪的场景中，
|
||
|
||
`Cond`的核心是一个`Locker`类型的字段，这个`Locker`类型的字段被用来在各个协程中的完成控制操作，这个`Locker`类型的字段通常都是使用`Mutex`类型或者`RWMutex`类型的实例来充当。`Cond`提供了以下几个方法来进行协程的控制。
|
||
|
||
- `Broadcast()`用于唤醒所有等待`Cond`变量的协程。
|
||
- `Signal()`用于只唤醒一个正在等待`Cond`变量的协程。
|
||
- `Wait()`用于挂起调用协程，让调用`Wait()`的协程等待`Broadcast()`或者`Signal()`。
|
||
|
||
!!! warning ""
|
||
在协程中调用`Wait()`挂起当前协程的时候，需要先调用`Cond`实例中`Locker`对象的锁，这个锁是用来锁闭协程中所使用到的共享资源的。也就是说在使用`Wait()`之前，必须先锁闭`Cond`实例。
|
||
|
||
!!! note ""
|
||
不必担心调用`Wait()`的时候`Cond`中的锁会阻止通知协程对于共享资源的访问。在调用`Wait()`的时候，`Cond`中的所将会自动解锁，当`Wait()`被`Broadcast()`或者`Signal()`结束而返回的时候，会自动再次对`Cond`中的锁进行锁闭。
|
||
|
||
以下是一个使用`Cond`进行协程控制的简单示例。
|
||
|
||
```go
|
||
var (
|
||
m sync.Mutex
|
||
c = sync.NewCond(&m)
|
||
sharedResouce = false
|
||
)
|
||
|
||
for i := 0; i < 4; i++ {
|
||
go func() {
|
||
c.L.Lock() // 锁闭Cond的锁，用来访问共享的sharedResouces
|
||
for sharedResource == false {
|
||
// 完成共享资源就绪前的处理
|
||
c.Wait()
|
||
}
|
||
// 完成资源就绪以后的处理
|
||
c.L.Unlock()
|
||
}()
|
||
}
|
||
|
||
go func() {
|
||
time.Sleep(rand.Int(100) * time.Second)
|
||
c.L.Lock()
|
||
sharedResources = true
|
||
c.Broadcast()
|
||
c.L.Unlock()
|
||
}()
|
||
```
|
||
|
||
## `WaitGroup`
|
||
|
||
`WaitGroup`直译过来就是“等待组”，它的功能就是用来等待一组协程的结束。`WaitGroup`的使用其实很简单，只需要在主协程启动子协程之前调用`Add()`来调整目前需要等待的子协程数量，然后在子协程执行结束以后调用`Done()`，然后再在需要让主协程停下来的地方调用`Wait()`即可。用一个示例说明就是下面这样。
|
||
|
||
```go
|
||
func main() {
|
||
var wg sync.WaitGroup
|
||
|
||
// 初始化需要完成的任务tasks
|
||
|
||
for _, task := range tasks {
|
||
wg.Add(1) // 调整等待组中需要等待任务的数量
|
||
go func() {
|
||
defer wg.Done() // 标记等待组中任务的完成
|
||
|
||
// 执行task
|
||
}()
|
||
}
|
||
|
||
wg.Wait() // 暂停主协程，使其等待子协程的完成
|
||
}
|
||
```
|