[Golang] Sync
文章目录
- [Golang] Sync
- sync.WaitGroup
- sync.Once
- 对比init()
- sync.Lock
- 互斥锁和读写锁
- 互斥锁
- 读写锁
- 死锁问题
- 加锁解锁不成对
- 循环等待
- sync.Map
- sync/atomic
- atomic和mutex的区别
- atomic.value
- sync.Pool
- 如何使用
- sync.Pool使用场景
Golang中我们一般提倡使用通信来共享内存，不使用共享内存来通信，比如Goroutine之间通过channel来协作。而其他语言中，都是通过共享内存加锁机制来保证并发安全的，同样的Golang中也提供对共享内存并发安全机制的支持，它们都在sync包中。
sync.WaitGroup
之前我们已经使用过sync.WaitGroup来替换time.Sleep，让一个协程等待一个协程执行完毕，也就是使用sync.WaitGroup来实现并发任务的同步以及协程任务等待。
使用方式：
sync.WaitGroup是一个对象，里面维护一个计数器，并通过三个方法来配合使用
-
(wg *WaitGroup) Add(delta int) 计数器加cnt
-
(wg *WaitGroup) Done() 计数器减1
-
(wg *WaitGroup) Wait() 阻塞代码运行，直到计数器减为0
package main
import (
"fmt"
"sync"
)
func main() {
var wg sync.WaitGroup
wg.Add(10)
for i := 0; i < 10; i++ {
go myGoroutine(&wg)
}
wg.Wait()
}
func myGoroutine(wg *sync.WaitGroup) {
defer wg.Done()
fmt.Println("myGoroutine")
}
执行结果：
先把计数器设置为10，每运行完一个myGoroutine就把计数器减1，main函数等待计数器值为0，也就是10个子协程打印完10个myGoroutine后，子协程全部退出，主协程才会退出。
ps：计数器的值不能减为负数，不然就会panic。
sync.Once
很多时候，程序中有很多逻辑只需要执行一次，比如配置文件的加载，我们只需要加载一次，让配置保存在内存中，下次直接使用内存中的配置文件即可，这时就会用到sync.Once。
sync.Once可以在代码的任意位置初始化和调用，并且线程安全。对于一个sync.Once变量我们并不会在程序启动时初始化，而是在第一次使用到它时才进行初始化，并且只初始化这一次，初始化后保存在内存中，这就很符合我们刚刚说到的配置文件加载的场景，这其实也就是单例模式中的懒汉模式。毕竟一开始就加载到内存中，长时间不用就浪费了内存。
package main
import (
"fmt"
"sync"
)
type Config struct{}
var instance *Config
var once sync.Once
var cnt int
func InitConfig() *Config {
once.Do(func() {
fmt.Printf("第%d次被调用", cnt)
instance = &Config{}
})
return instance
}
func main() {
cnt = 1
InitConfig()
InitConfig()
go InitConfig()
InitConfig()
go InitConfig()
InitConfig()
InitConfig()
}
执行结果：
只有第一次调用InitConfig()时获取Config指针时才会执行once.Do()语句，执行完后instance就保留在内存中了，后面再次执行时，会直接返回这个instance。
对比init()
init()：适用于程序启动时的初始化，确保在主函数执行前完成初始化任务。比如，日志配置初始化，在程序启动时配置日志文件。sync.Once：适用于延迟初始化且在并发环境下只需要执行一次的初始化任务。比如，配置文件加载、数据库连接池初始化。
sync.Lock
并发编程中资源的竞争，Golang给出了两种解决方案：锁和原子操作。
package main
import (
"fmt"
"sync"
)
func main() {
n := 10000
sum := 0
var wg sync.WaitGroup
wg.Add(n)
for i := 0; i < n; i++ {
go func() {
defer wg.Done()
sum += 1
}()
}
wg.Wait()
fmt.Println(sum)
}
执行结果：
最后10000个协程执行完之后，sum并不是1000，这就出现了并发问题。同一时间多个Goroutine对sum做+1操作，但是不是在在前一个协程执行完的基础上做的累加，这样前一个协程的执行就会被后一个协程的执行结果覆盖了。
互斥锁和读写锁
互斥锁
互斥锁：同一时间只允许一个goroutine对共享资源进行访问。
var lock sync.Lock
func (m *Mutex) Lock() // 加锁
func (m *Mutex) UnLock() // 解锁
我们对上面的代码稍加修改：
package main
import (
"fmt"
"sync"
)
func main() {
n := 10000
sum := 0
var wg sync.WaitGroup
wg.Add(n)
mu := sync.Mutex{}
for i := 0; i < n; i++ {
go func() {
mu.Lock()
defer wg.Done()
sum += 1
mu.Unlock()
}()
}
wg.Wait()
fmt.Println(sum)
}
执行结果：
注意：加锁之后不用忘记解锁，否则会造成其他goroutine一直阻塞。
读写锁
把读操作和写操作分离，一般用于大量读操作、少量写操作的情况。
var mr sync.RWMutex
func (rw *RWMutex) Lock() // 对写锁加锁
func (rw *RWMutex) UnLock() // 对写锁解锁
func (rw *RWMutex) RLock() // 对读锁加锁
func (rw *RWMutex) RUnLock() // 对读锁解锁
读写锁的使用：多个Goroutine可以同时读，但是只有一个Goroutine能写；共享资源要么在被一个或多个Goroutine读，要么在被一个Goroutine写，读写不能同时进行。
死锁问题
死锁：在两个以上的Goroutine执行过程中，因为争抢共享资源处于互相等待的状态，如果没有外部干涉将一直处于这个阻塞的状态。
加锁解锁不成对
这种场景一般是对锁进行拷贝的使用：
package main
import (
"fmt"
"sync"
)
func main() {
var mu sync.Mutex
mu.Lock()
defer mu.Unlock()
copyMutex1(mu)
copyMutex2(&mu)
fmt.Println("main end...")
}
func copyMutex1(mu sync.Mutex) {
mu.Lock()
defer mu.Unlock()
fmt.Println("copyMutex1 end...")
}
func copyMutex2(mu *sync.Mutex) {
mu.Lock()
defer mu.Unlock()
fmt.Println("copyMutex2 end...")
}
执行结果：
如果把带有锁结构的变量赋值给其他变量，锁的状态会复制。所以复制之后的锁已经有了原来的锁状态，那么copyMutex()中执行mu.Lock()会被一直阻塞，因为外面的main函数中已经Lock()过了一次但是还没有Unlock()。这就导致了copyMutex()内锁一直在等待Lock()，而main()内一直在等待解锁，这就导致了死锁。
所以在使用锁时，我们要避免拷贝锁，并且Lock()和UnLock要成对出现。
循环等待
package main
import (
"sync"
"time"
)
func main() {
var wg sync.WaitGroup
wg.Add(2)
var mu1, mu2 sync.Mutex
go func() {
defer wg.Done()
mu1.Lock()
defer mu1.Unlock()
time.Sleep(time.Second)
mu2.Lock()
defer mu2.Unlock()
}()
go func() {
defer wg.Done()
mu2.Lock()
defer mu2.Unlock()
time.Sleep(time.Second)
mu1.Lock()
defer mu1.Unlock()
}()
wg.Wait()
}
执行结果：
两个Goroutine在加第二个锁时，都会等待对方释放锁，造成了循环等待，一直阻塞，形成了死锁。
sync.Map
golang中内置的Map不是并发安全的，多个goroutine同时操作map时会有并发问题。
比如：
package main
import (
"fmt"
"sync"
)
func main() {
var wg sync.WaitGroup
wg.Add(10)
mp := make(map[int]int)
for i := 0; i < 10; i++ {
go func(num int) {
defer wg.Done()
mp[num] = num + 1
fmt.Printf("key=%v, value=%v", num, mp[num])
}(i)
}
wg.Wait()
}
执行结果：
说明内置的map不能并发操作。
解决方案1：
加锁：
package main
import (
"fmt"
"sync"
)
func main() {
var wg sync.WaitGroup
wg.Add(10)
var mu sync.Mutex
mp := make(map[int]int)
for i := 0; i < 10; i++ {
go func(num int) {
defer wg.Done()
mu.Lock()
defer mu.Unlock()
mp[num] = num + 1
fmt.Printf("key=%v, value=%v\n", num, mp[num])
}(i)
}
wg.Wait()
}
执行结果：
解决方案2：
使用并发安全的map：sync.Map
比如：
package main
import (
"fmt"
"sync"
)
func main() {
var mp sync.Map
// 1.写入
mp.Store("name", "张三")
mp.Store("age", 18)
// 2.读取
age, _ := mp.Load("age")
fmt.Println(age.(int)) // 断言age为int类型
// 3.遍历
mp.Range(func(key, value any) bool {
fmt.Printf("key = %v, value = %v\n", key, value)
return true
})
// 4.删除
mp.Delete("age")
age, ok := mp.Load("age")
fmt.Println(age, ok)
// 5.读取或写入
mp.LoadOrStore("name", "李四")
name, _ := mp.Load("name")
fmt.Printf("name = %v", name)
}
执行结果：
sync/atomic
之前说了锁，现在说另一种解决并发安全的策略：atomic原子操作。
// T的类型为 int32、int64、uint32、uint64和uintptr的任意一种
func AddT(addr *T, delta T)(new T)
func StoreT(addr *T, val T)
func LoadT(addr *T) (val T)
func SwapT(addr *T, new T) (old T)
func CompareAndSwap(addr *T, old,new T) (swapped bool)
例如：
package main
import (
"fmt"
"sync"
"sync/atomic"
)
func main() {
var wg sync.WaitGroup
wg.Add(100)
var num int32 = 0
for i := 0; i < 100; i++ {
go func() {
defer wg.Done()
atomic.AddInt32(&num, 1)
}()
}
wg.Wait()
fmt.Println(num)
}
执行结果：
atomic和mutex的区别
使用方式：mutex通常用于保护一段代码执行逻辑；atomic一般用于对变量的操作
底层实现：mutex由操作系统调度器实现；atomic操作有底层硬件指令支持，保证cpu在执行上不中断。
atomic.value
atomic也支持了对struct这种复合类型进行原子操作。
比如：
package main
import (
"fmt"
"sync/atomic"
)
type Student struct {
Name string
Age int
}
func main() {
st1 := Student{
Name: "张三",
Age: 18,
}
st2 := Student{
Name: "李四",
Age: 20,
}
st3 := Student{
Name: "王五",
Age: 22,
}
var v atomic.Value
v.Store(st1) // 1.写入
st := v.Load().(Student) // 2.读取
fmt.Println(st)
old := v.Swap(st2) // 3.交换
st = v.Load().(Student)
fmt.Println("after swap :", st)
fmt.Println("old :", old)
swapped := v.CompareAndSwap(st1, st3) // 4.比较并交换
fmt.Println("交换: ", swapped)
st = v.Load().(Student)
fmt.Println(st)
swapped = v.CompareAndSwap(st2, st3) // 4.比较并交换
fmt.Println("交换: ", swapped)
st = v.Load().(Student)
fmt.Println(st)
}
执行结果：
sync.Pool
sync.Pool是sync包下的一个内存池组件，用来实现对象的复用，避免重复创建相同的对象，造成频繁的内存分配和gc，以达到提升程序性能的目的。虽然池子中的对象可以被复用，但是sync.Pool并不会永久保存这个对象，池中的对象会在一定时间后被gc回收，这个时间是随机的。所以不能用sync.Pool来持久化存储对象。
如何使用
New() // sync.Pool的构造函数，用于指定sync.Pool中缓存的数据类型，
// 如果调用Get()时，池中没有元素就会调用New()方法创建一个新对象
Get() // 从对象池取对象
Put() // 向对象池中放对象，下次Get()时可以复用
例如：
package main
import (
"fmt"
"sync"
)
type Student struct {
Name string
Age int
}
func main() {
pool := sync.Pool{
New: func() interface{} {
return &Student{
Name: "张三",
Age: 18,
}
},
}
st := pool.Get().(*Student) //(*Student)断言
println(st.Name, st.Age)
fmt.Printf("st addr = %p\n", st)
pool.Put(st)
st2 := pool.Get().(*Student)
println(st2.Name, st2.Age)
fmt.Printf("st addr = %p\n", st2)
}
执行结果：
程序逻辑：
我们先初始化一个sync.Pool对象，初始化New()方法，用于创建对象，这里是返回一个*Student。
第一次调用Get()时，池中没有对象，所以会调用New()方法创建一个，由于返回类型为interface{}所以需要我们断言一下
使用完后，再调用Put()方法，把对象放回池中，再调用pool.Get取对象，此时我们可能看到的对象地址是同一个，如果看到同一个说明sync.Pool有缓存对象的功能。
ps：由于 sync.Pool 的设计是为了在高并发环境下工作，它的行为可能不是完全可预测的。在某些情况下，即使你将对象放回了池中，下次获取时也可能得到一个新的对象。
一般我们如果在修改对象字段后，回收前记得Reset，否则取到的对象是同一个，但是字段内容变化了。
比如：
package main
import (
"fmt"
"sync"
)
type Student struct {
Name string
Age int
}
func main() {
pool := sync.Pool{
New: func() interface{} {
return &Student{
Name: "张三",
Age: 18,
}
},
}
st := pool.Get().(*Student)
println(st.Name, st.Age)
fmt.Printf("st addr = %p\n", st)
// 修改
st.Name = "李四"
st.Age = 20
pool.Put(st)
st2 := pool.Get().(*Student)
println(st2.Name, st2.Age)
fmt.Printf("st2 addr = %p\n", st2)
}
执行结果：
sync.Pool没有提供Reset函数，一般需要我们进行手写：
package main
import (
"fmt"
"sync"
)
type Student struct {
Name string
Age int
}
func main() {
pool := sync.Pool{
New: func() interface{} {
return &Student{
Name: "张三",
Age: 18,
}
},
}
st := pool.Get().(*Student)
println(st.Name, st.Age)
fmt.Printf("st addr = %p\n", st)
// 修改对象状态
st.Name = "李四"
st.Age = 20
// 在放回池中之前重置对象状态
resetStudent(st)
pool.Put(st)
st2 := pool.Get().(*Student)
println(st2.Name, st2.Age)
fmt.Printf("st2 addr = %p\n", st2)
}
// resetStudent 用于重置 Student 对象的状态
func resetStudent(s *Student) {
s.Name = "张三"
s.Age = 18
}
sync.Pool使用场景
sync.Pool通过复用对象来降低gc带来的性能损耗，高并发场景下，每个Goroutine都可能频繁创建一些大对象，造成gc压力很大。所以在高并发场景下出现gc问题时，可以使用sync.Pool减少gc负担。
sync.Pool不能存储带状态的对象，比如Socket连接、数据库连接，因为池中的对象随时可能被gc回收释放；sync.Pool不适合控制缓存对象个数的场景，因为sync.Pool中的对象个数是随机变化的，池中的对象随时有可能被gc的，释放时机是随机的。
