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

Go语言并发编程：sync包深度解析与实践

news 2026/5/22 4:03:33

Go语言并发编程：sync包深度解析与实践

引言

sync包是Go语言中用于同步的标准库，提供了互斥锁、读写锁、条件变量等并发原语。本文将深入探讨sync包的核心组件和使用模式。

一、sync.Mutex

1.1 基础用法

var mu sync.Mutex var count int func increment() { mu.Lock() defer mu.Unlock() count++ } func getCount() int { mu.Lock() defer mu.Unlock() return count }

1.2 实现原理

type Mutex struct { state int32 sema uint32 } const ( mutexLocked = 1 << iota mutexWoken mutexStarving mutexWaiterShift = iota )

1.3 公平锁与非公平锁

func (m *Mutex) Lock() { if atomic.CompareAndSwapInt32(&m.state, 0, mutexLocked) { return } m.lockSlow() } func (m *Mutex) lockSlow() { // 自旋尝试获取锁 // 如果失败，进入休眠等待 }

二、sync.RWMutex

2.1 读写锁特性

操作	读锁持有	写锁持有
获取读锁	允许	阻塞
获取写锁	阻塞	阻塞

2.2 用法示例

var rwmu sync.RWMutex var data map[string]string func readData(key string) string { rwmu.RLock() defer rwmu.RUnlock() return data[key] } func writeData(key, value string) { rwmu.Lock() defer rwmu.Unlock() data[key] = value }

三、sync.WaitGroup

3.1 基本用法

func main() { var wg sync.WaitGroup for i := 0; i < 5; i++ { wg.Add(1) go func(id int) { defer wg.Done() fmt.Printf("Goroutine %d finished\n", id) }(i) } wg.Wait() fmt.Println("All goroutines finished") }

3.2 实现原理

type WaitGroup struct { noCopy noCopy state1 [3]uint32 } func (wg *WaitGroup) Add(delta int) { statep, semap := wg.state() state := atomic.AddUint64(statep, uint64(delta)<<32) v := int32(state >> 32) w := uint32(state) if v < 0 { panic("sync: negative WaitGroup counter") } }

四、sync.Once

4.1 保证单次执行

var once sync.Once var instance *Database func getDB() *Database { once.Do(func() { instance = &Database{ Host: "localhost", Port: 5432, } }) return instance }

4.2 实现原理

type Once struct { m Mutex done uint32 } func (o *Once) Do(f func()) { if atomic.LoadUint32(&o.done) == 1 { return } o.m.Lock() defer o.m.Unlock() if o.done == 0 { defer atomic.StoreUint32(&o.done, 1) f() } }

五、sync.Cond

5.1 条件变量

var cond = sync.NewCond(&sync.Mutex{}) var queue []int func producer(item int) { cond.L.Lock() queue = append(queue, item) cond.L.Unlock() cond.Signal() } func consumer() { cond.L.Lock() for len(queue) == 0 { cond.Wait() } item := queue[0] queue = queue[1:] cond.L.Unlock() fmt.Printf("Consumed: %d\n", item) }

5.2 Signal vs Broadcast

func broadcastAll() { cond.L.Lock() // 修改共享状态 cond.L.Unlock() cond.Broadcast() // 唤醒所有等待者 }

六、sync.Map

6.1 并发安全Map

var m sync.Map func main() { m.Store("key1", "value1") m.Store("key2", "value2") if val, ok := m.Load("key1"); ok { fmt.Println(val) } m.Range(func(key, value interface{}) bool { fmt.Printf("%s: %s\n", key, value) return true }) }

6.2 实现原理

type Map struct { mu Mutex read atomic.Pointer[readOnly] dirty map[any]*entry misses int } type readOnly struct { m map[any]*entry amended bool }

七、sync.Pool

7.1 对象复用

var bufferPool = sync.Pool{ New: func() interface{} { return make([]byte, 1024) }, } func process(data []byte) { buf := bufferPool.Get().([]byte) defer bufferPool.Put(buf) copy(buf, data) // 处理数据 }

7.2 优化场景

type Object struct { Data []byte } var objectPool = sync.Pool{ New: func() interface{} { return &Object{ Data: make([]byte, 4096), } }, } func createObject() *Object { obj := objectPool.Get().(*Object) obj.Data = obj.Data[:0] // 重置数据 return obj } func releaseObject(obj *Object) { objectPool.Put(obj) }

八、实战：线程安全的数据结构

type SafeQueue struct { mu sync.Mutex queue []interface{} cond *sync.Cond } func NewSafeQueue() *SafeQueue { q := &SafeQueue{ queue: make([]interface{}, 0), } q.cond = sync.NewCond(&q.mu) return q } func (q *SafeQueue) Enqueue(item interface{}) { q.mu.Lock() q.queue = append(q.queue, item) q.mu.Unlock() q.cond.Signal() } func (q *SafeQueue) Dequeue() interface{} { q.mu.Lock() defer q.mu.Unlock() for len(q.queue) == 0 { q.cond.Wait() } item := q.queue[0] q.queue = q.queue[1:] return item }

九、性能优化建议

9.1 减少锁竞争

// 错误：单一锁导致高竞争 var mu sync.Mutex var data map[string]int // 正确：分片锁 type ShardedMap struct { shards int mu []sync.Mutex data []map[string]int }

9.2 使用读写锁优化读多写少场景

var rwmu sync.RWMutex // 读操作 func read() { rwmu.RLock() defer rwmu.RUnlock() // 读取数据 } // 写操作 func write() { rwmu.Lock() defer rwmu.Unlock() // 写入数据 }