您诸位好啊,我是无尘。又到了愉快的周末,肝了一上午,给大家介绍下 sync 包。除了上一节我们介绍的 channel 通道,还有 sync.Mutex、sync.WaitGroup 这些原始的同步机制,来更加灵活的实现数据同步和控制并发。
所谓资源竞争,就是在程序中,同一块内存同时被多个 goroutine 访问。对于这个共享的资源(内存)每个 goroutine 都有不同的操作,就有可能造成数据紊乱。
示例:
package main
import (
"fmt"
"time"
)
var sum = 0
func main() {
//开启100个协程来让 sum + 1
for i := 1; i <= 100; i++ {
go add()
}
// 睡眠两秒防止程序提前退出
time.Sleep(2 * time.Second)
fmt.Println("sum:",sum)
}
func add(){
sum += 1
}
//运行结果:sum:98 或 sum:99 或 ...
解决这个问题,我们可以给资源进行加锁,让其在同一时刻只能被一个协程来操作。
示例:
package main
import (
"fmt"
"sync"
"time"
)
var sum = 0
var mutex = sync.Mutex{}
func main() {
//开启100个协程来让 sum + 1
for i := 1; i <= 100; i++ {
go add()
}
// 睡眠两秒防止程序提前退出
time.Sleep(2 * time.Second)
fmt.Println("sum:",sum)
}
func add(){
mutex.Lock()
defer mutex.Unlock() //使用defer语句,确保锁一定会被释放
sum += 1
}
示例:
package main
import (
"fmt"
"sync"
"time"
)
var sum = 0
var mutex = sync.Mutex{}
var rwmutex = sync.RWMutex{}
func main() {
//开启100个协程来让 sum + 1
for i := 1; i <= 100; i++ {
go add()
}
for i := 1; i<= 10; i++ {
go fmt.Println("sum:",getSum())
}
// 睡眠两秒防止程序提前退出
time.Sleep(2 * time.Second)
fmt.Println("sum:", sum)
}
func add(){
mutex.Lock()
defer mutex.Unlock() //使用defer语句,确保锁一定会被释放
sum += 1
}
func getSum() int {
rwmutex.RLock() //使用读写锁
defer rwmutex.RUnlock()
return sum
}
示例:
package main
import (
"fmt"
"sync"
)
var sum = 0
var mutex = sync.Mutex{}
var rwmutex = sync.RWMutex{}
func run() {
var wg sync.WaitGroup
//因为要监控110个协程,所以设置计数器为110
wg.Add(110)
for i := 1; i <= 100; i++ {
go func() {
//计数器值减1
defer wg.Done()
add()
}()
}
for i := 1; i <= 10; i++ {
go func() {
//计数器值减1
defer wg.Done()
fmt.Println("sum:", getSum())
}()
}
//一直等待,只要计数器值为0
wg.Wait()
}
func main() {
run()
}
func add() {
mutex.Lock()
defer mutex.Unlock() //使用defer语句,确保锁一定会被释放
sum += 1
}
func getSum() int {
rwmutex.RLock() //使用读写锁
defer rwmutex.RUnlock()
return sum
}
有时候我们只希望代码执行一次,即使是在高并发的场景下,比如创建一个单例。这种情况可以使用 sync.Once 来保证代码只执行一次。
示例:
package main
import (
"fmt"
"sync"
)
func main() {
var once sync.Once
onceBody := func() {
fmt.Println("Only once")
}
//用于等待协程执行完毕
done := make(chan bool)
//启动10个协程执行once.Do(onceBody)
for i := 0; i < 10; i++ {
go func() {
//把要执行的函数(方法)作为参数传给once.Do方法即可
once.Do(onceBody)
done <- true
}()
}
for i := 0; i < 10; i++ {
<-done
}
}
//运行结果:Only once
示例:
package main
import (
"fmt"
"sync"
"time"
)
func main() {
//3个人赛跑,1个裁判员发号施令
cond := sync.NewCond(&sync.Mutex{})
var wg sync.WaitGroup
wg.Add(4) //3选手+1裁判
for i := 1; i <= 3; i++ {
go func(num int) {
defer wg.Done()
fmt.Println(num, "号选手已经就位")
cond.L.Lock()
cond.Wait() //等待发令枪响
fmt.Println(num, "号选手开始跑……")
cond.L.Unlock()
}(i)
}
//等待所有goroutine都进入wait状态
time.Sleep(2 * time.Second)
go func() {
defer wg.Done()
fmt.Println("裁判:“各就各位~~预备~~”")
fmt.Println("啪!!!")
cond.Broadcast() //发令枪响
}()
//防止函数提前返回退出
wg.Wait()
}
运行结果:
3 号选手已经就位
1 号选手已经就位
2 号选手已经就位
裁判:“各就各位~~预备~~”
啪!!!
2 号选手开始跑……
3 号选手开始跑……
1 号选手开始跑……
最后贴一下sync.Cond 几个方法的源码:
// Wait atomically unlocks c.L and suspends execution
// of the calling goroutine. After later resuming execution,
// Wait locks c.L before returning. Unlike in other systems,
// Wait cannot return unless awoken by Broadcast or Signal.
// Wait方法释放锁,并阻塞协程执行。满足条件解除阻塞后,当前协程需要获得锁然后Wait方法返回。
//
// Because c.L is not locked when Wait first resumes, the caller
// typically cannot assume that the condition is true when
// Wait returns. Instead, the caller should Wait in a loop:
// 由于解除阻塞后,当前协程不一定能马上获得锁,因此返回后需要再次检查条件,所以通常
// 使用循环。
// c.L.Lock()
// for !condition() {
// c.Wait()
// }
// ... make use of condition ...
// c.L.Unlock()
//
func (c *Cond) Wait() {
c.checker.check()
t := runtime_notifyListAdd(&c.notify)
c.L.Unlock() // 释放锁
runtime_notifyListWait(&c.notify, t) // 等待满足条件,解除阻塞
c.L.Lock() // 获取锁
}
// Signal wakes one goroutine waiting on c, if there is any.
//
// It is allowed but not required for the caller to hold c.L
// during the call.
func (c *Cond) Signal() {
c.checker.check()
runtime_notifyListNotifyOne(&c.notify)
}
// Broadcast wakes all goroutines waiting on c.
//
// It is allowed but not required for the caller to hold c.L
// during the call.
func (c *Cond) Broadcast() {
c.checker.check()
runtime_notifyListNotifyAll(&c.notify)
}
条件变量的Wait方法主要做了四件事:
注意事项
// c.L.Lock()
// for !condition() {
// c.Wait()
// }
// ... make use of condition ...
// c.L.Unlock()