当前位置： 首页 > article >正文

[Golang] Sync

article 2024/9/20 4:46:04

[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")
}

执行结果：

image-20240916175936050

先把计数器设置为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()

}

执行结果：

image-20240916181728455

只有第一次调用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)
}

执行结果：

image-20240916182645164

最后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)
}

执行结果：

image-20240916183516253

注意：加锁之后不用忘记解锁，否则会造成其他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...")
}

执行结果：

image-20240916184736568

如果把带有锁结构的变量赋值给其他变量，锁的状态会复制。所以复制之后的锁已经有了原来的锁状态，那么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()
}

执行结果：

image-20240916190424536

两个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()
}

执行结果：

image-20240916192225248

说明内置的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()
}

执行结果：

image-20240916192118056

解决方案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)
}

执行结果：

image-20240916193324191

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)
}

执行结果：

image-20240916194456421

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)
}

执行结果：

image-20240916195800443

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)
}

执行结果：

image-20240916204315463

程序逻辑：

我们先初始化一个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)
}

执行结果：

image-20240916204503090

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的，释放时机是随机的。


http://www.kler.cn/news/306939.html

相关文章：

  • HarmonyOS开发之全局状态管理
  • 天融信把桌面explorer.exe删了，导致开机之后无windows桌面，只能看到鼠标解决方法
  • C++基础面试题 | 什么是C++中的虚继承？
  • LabVIEW机动车动态制动性能校准系统
  • spring项目中如何通过redis的setnx实现互斥锁解决缓存缓存击穿问题
  • [项目][WebServer][HttpServer]详细讲解
  • 一码空传临时网盘PHP源码，支持提取码功能
  • 数据中台进化为数据飞轮的必要
  • 【笔记】自动驾驶预测与决策规划_Part2_基于模型的预测方法
  • 初学Linux（学习笔记）
  • Vue.js入门系列（二十九）：深入理解编程式路由导航、路由组件缓存与路由守卫
  • 【C++】入门基础（下）
  • Java项目基于docker 部署配置
  • 关于新版本 tidb dashboard API 调用说明
  • 评价类——熵权法（Entropy Weight Method, EWM），完全客观评价
  • ansible安全优化篇
  • 在深圳停车场我居然能看到很漂亮的瓦房
  • 707. 设计链表
  • SQL，从每组中的 json 字段中提取唯一值
  • 鸿蒙开发基础
  • Rust Web开发框架对比：Warp与Actix-web
  • SpringBoot + MySQL + MyBatis 实操示例教学
  • 从冯唐的成事心法 看SAP协助企业战略落地到信息化
  • 车载软件架构 --- SOA设计与应用（上）
  • DAY20240913 VUE:深入解析 Vue Router 局部路由守卫：路由独享与组件内部守卫的妙用与区别
  • 自修C++PrimerPlus--类型转换、右值引用、引用中的类对象
  • 虹科方案 | 精准零部件测试！多路汽车开关按键功能检测系统
  • C++实现unordered_map和unordered_set
  • 【Kafka】分区与复制机制：解锁高性能与容错的密钥
  • 交换技术是一种在计算机网络和通信系统中广泛应用的关键技术，它主要通过交换设备（如交换机、路由器等）实现数据的转发和传输