go语言进阶之同步原语

同步原语

资源竞争

定义与实现

在Go语言中，资源竞争指多个goroutine同时访问共享资源，导致程序的行为不可预测或者不一致。

资源竞争通常发生在对同一变量进行读写操作时，如果没有正确的同步机制来控制访问可能会引发资源竞争

package main

import (
    "fmt"
    "sync"
)

var counter int // 共享资源

func increment(wg *sync.WaitGroup) {
    defer wg.Done()
    counter++
}

func main() {
    var wg sync.WaitGroup
    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go increment(&wg)
    }
    wg.Wait()
    fmt.Println("Final counter:", counter) // 可能不是 1000
}

![[Pasted image 20241029003944.png]]
可以运行后看到发生了资源竞争，导致不为1000。

原因

1. 多个goroutine同时写入或读取共享资源，缺乏恰当的同步机制
2. 非原子操作，对共享变量的操作如果不是原子的，会导致在操作过程中被其他goroutine中断，导致数据错误

与安全的关系

1. 可能会导致数据一致性问题，影响应用数据
2. 并发攻击，比如对ddos可以利用并发性来压垮系统，导致资源竞争和其他问题。
3. 状态泄露，如果资源竞争导致应用状态不一样，可能可以利用这些漏洞来获取敏感信息或执行恶意操作。

同步原语

同步原语用于控制多个goroutine之间的访问顺序和协调，正好可以解决上面的资源竞争问题。

互斥锁(sync.Mutex)

确保同一时间只有一个goroutine可以访问资源

实现

package main  
  
import (  
    "fmt"  
    "sync")  
  
var (  
    mu      sync.Mutex  
    counter int  
)  
  
func increment() {  
    mu.Lock()   //加锁  
    counter++   //访问共享资源  
    mu.Unlock() //解锁  
}  
  
func main() {  
    for i := 1; i <= 1000; i++ {  
       increment()  
    }  
    fmt.Println(counter)  
  
}

可以看到现在输出为1000

![[Pasted image 20241029011834.png]]

sync.RWMutex

RWMutex是Go语言的读写互斥锁，用于处理读多写少的场景，他允许多个goroutine并发地读取共享资源，但是写操作时之允许一个goroutine进行写操作。

package main  
  
import (  
    "fmt"  
    "sync")  
  
var (  
    rwmu    sync.RWMutex // 创建读写互斥锁  
    counter int          // 共享计数器  
)  
  
func read(wg *sync.WaitGroup) {  
    defer wg.Done()                  // 完成时通知 WaitGroup    rwmu.RLock()                     // 获取读锁  
    fmt.Println("Counter:", counter) // 读取共享资源  
    rwmu.RUnlock()                   // 释放读锁  
}  
  
func write(wg *sync.WaitGroup) {  
    defer wg.Done() // 完成时通知 WaitGroup    rwmu.Lock()     // 获取写锁  
    counter++       // 修改共享资源  
    rwmu.Unlock()   // 释放写锁  
}  
  
func main() {  
    var wg sync.WaitGroup  
  
    // 启动多个读和写的 goroutine    for i := 0; i < 5; i++ {  
       wg.Add(1)  
       go read(&wg) // 启动读取 goroutine    }  
  
    for i := 0; i < 5; i++ {  
       wg.Add(1)  
       go write(&wg) // 启动写入 goroutine    }  
  
    wg.Wait() // 等待所有 goroutine 完成  
}

这个程序由于read在write前所以可以更好的体现read和write是同时进行的

输出为

![[Pasted image 20241029014147.png]]

这个的输出不固定，因为无法确定程序每一次的执行速度。

sync.WaitGroup

WaiGroup 常用于等待一组goroutine完成，它可以让主goroutine等待其他多个goroutine的完成

package main

import (
    "fmt"
    "sync"
	"time"
)

func worker(id int, wg *sync.WaitGroup) {
    defer wg.Done() // 在函数结束时调用 Done()
    fmt.Printf("Worker %d is working...\n", id)
    time.Sleep(1 * time.Second) // 模拟一些工作
    fmt.Printf("Worker %d is done.\n", id)
}

func main() {
    var wg sync.WaitGroup

    for i := 1; i <= 5; i++ {
        wg.Add(1) // 增加 WaitGroup 计数
        go worker(i, &wg) // 启动 goroutine
    }

    wg.Wait() // 等待所有 goroutine 完成
    fmt.Println("All workers are done.")
}

这个函数主要的作用就是等待所有线程执行完再结束主线程，如果当主线程提前结束，其他线程就算没有完成执行操作也无法继续执行了

sync.Once

Once函数和它的名字一样，主要作用就是确保某段代码只执行一次，无论有多少个goroutine尝试调用它，都只执行一次。

package main

import (
    "fmt"
    "sync"
)

var once sync.Once

func initialize() {
    fmt.Println("正在初始化...")
}

func worker(id int) {
    // 仅调用一次初始化函数
    once.Do(initialize)
    fmt.Printf("工作者 %d 正在工作\n", id)
}

func main() {
    var wg sync.WaitGroup

    for i := 1; i <= 5; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            worker(id)
        }(i)
    }

    wg.Wait() // 等待所有 goroutine 完成
    fmt.Println("所有工作者完成")
}

这段代码运行后可以发现，初始化操作只运行了一次，当有资源初始化等操作时，使用Once函数是一个不错的选择。

sync.Cond

Cond函数是一个用于实现条件变量的同步原语。它通常和Mutex和RWMutex一起使用，允许goroutines在特定条件下进行等待和通知

package main

import (
	"fmt"
	"sync"
	"time"
)

// Resource 结构体包含一个互斥锁和一个条件变量
type Resource struct {
	mu    sync.Mutex   // 用于保护共享资源的互斥锁
	cond  *sync.Cond   // 条件变量，用于协程之间的同步
	value int          // 共享资源的值
}

// NewResource 创建一个新的 Resource 实例
func NewResource() *Resource {
	r := &Resource{}
	r.cond = sync.NewCond(&r.mu) // 初始化条件变量，关联互斥锁
	return r
}

// SetValue 设置资源的值，并通知等待的协程
func (r *Resource) SetValue(val int) {
	r.mu.Lock()         // 获取互斥锁
	r.value = val       // 设置资源的值
	r.cond.Broadcast()  // 唤醒所有等待的协程
	r.mu.Unlock()       // 释放互斥锁
}

// GetValue 获取资源的值，如果值为0则等待
func (r *Resource) GetValue() int {
	r.mu.Lock()                       // 获取互斥锁
	for r.value == 0 {                // 如果值为0，则等待
		r.cond.Wait()                 // 等待条件变量信号
	}
	val := r.value                    // 获取资源的值
	r.mu.Unlock()                     // 释放互斥锁
	return val                        // 返回获取的值
}

func main() {
	resource := NewResource() // 创建一个新的资源实例

	// 启动一个协程，设置资源的值
	go func() {
		time.Sleep(1 * time.Second) // 等待1秒以确保 GetValue 先调用
		fmt.Println("设置值为42")
		resource.SetValue(42) // 设置值为42，并通知等待的协程
	}()

	// 主协程获取资源的值
	val := resource.GetValue() // 获取值
	fmt.Printf("获取的值是: %d\n", val) // 打印获取的值
}

这个函数主要的作用就是使用Wait方法阻塞协程，然后用Signal()或Broadcast()方法唤醒一个等待时间最长的协程；或者唤醒全部正在等待的协程。