3. 同步原语详解
当 Channel 不足以满足需求时，sync 包提供了丰富的同步原语。
3.1 Mutex (互斥锁)
3.1.1 基本用法
go
package main
import (
"fmt"
"sync"
)
type Counter struct {
mu sync.Mutex
count int
}
func (c *Counter) Increment() {
c.mu.Lock()
defer c.mu.Unlock()
c.count++
}
func (c *Counter) Value() int {
c.mu.Lock()
defer c.mu.Unlock()
return c.count
}
func main() {
counter := &Counter{}
var wg sync.WaitGroup
// 100 个 goroutine 同时增加计数
for i := 0; i < 100; i++ {
wg.Add(1)
go func() {
defer wg.Done()
counter.Increment()
}()
}
wg.Wait()
fmt.Println("最终计数:", counter.Value()) // 100
}3.1.2 读写互斥锁 (RWMutex)
go
package main
import (
"fmt"
"sync"
"time"
)
type Cache struct {
mu sync.RWMutex
data map[string]string
}
func (c *Cache) Get(key string) (string, bool) {
c.mu.RLock()
defer c.mu.RUnlock()
val, ok := c.data[key]
return val, ok
}
func (c *Cache) Set(key, value string) {
c.mu.Lock()
defer c.mu.Unlock()
c.data[key] = value
}
func main() {
cache := &Cache{data: make(map[string]string)}
// 多个读操作可以并发
for i := 0; i < 10; i++ {
go func() {
for j := 0; j < 100; j++ {
cache.Get("key")
}
}()
}
// 写操作会阻塞所有读写
go func() {
for i := 0; i < 10; i++ {
cache.Set("key", fmt.Sprintf("value-%d", i))
time.Sleep(time.Millisecond)
}
}()
time.Sleep(time.Second)
}3.2 WaitGroup
3.2.1 基本用法
go
package main
import (
"fmt"
"sync"
"time"
)
func main() {
var wg sync.WaitGroup
for i := 0; i < 3; i++ {
wg.Add(1)
go func(id int) {
defer wg.Done()
fmt.Printf("任务 %d 开始\n", id)
time.Sleep(time.Second)
fmt.Printf("任务 %d 完成\n", id)
}(i)
}
wg.Wait()
fmt.Println("所有任务完成!")
}3.2.2 错误收集
go
package main
import (
"errors"
"fmt"
"sync"
)
func main() {
var wg sync.WaitGroup
errs := make(chan error, 3)
tasks := []func() error{
func() error { return nil },
func() error { return errors.New("task 2 failed") },
func() error { return nil },
}
for i, task := range tasks {
wg.Add(1)
go func(id int, fn func() error) {
defer wg.Done()
if err := fn(); err != nil {
errs <- fmt.Errorf("task %d: %w", id, err)
}
}(i, task)
}
wg.Wait()
close(errs)
for err := range errs {
fmt.Println("错误:", err)
}
}3.3 Cond (条件变量)
3.3.1 基本用法
go
package main
import (
"fmt"
"sync"
"time"
)
type Pool struct {
mu sync.Mutex
cond *sync.Cond
size int
max int
}
func NewPool(max int) *Pool {
p := &Pool{max: max}
p.cond = sync.NewCond(&p.mu)
return p
}
func (p *Pool) Acquire() {
p.mu.Lock()
for p.size >= p.max {
p.cond.Wait()
}
p.size++
p.mu.Unlock()
fmt.Printf("获取资源，当前大小：%d\n", p.size)
}
func (p *Pool) Release() {
p.mu.Lock()
p.size--
p.cond.Signal()
p.mu.Unlock()
fmt.Printf("释放资源，当前大小：%d\n", p.size)
}
func main() {
pool := NewPool(3)
for i := 0; i < 5; i++ {
go func(id int) {
pool.Acquire()
time.Sleep(2 * time.Second)
pool.Release()
}(i)
}
time.Sleep(10 * time.Second)
}3.4 Once
3.4.1 单次执行
go
package main
import (
"fmt"
"sync"
"sync/atomic"
)
var (
once sync.Once
counter int64
)
func initOnce() {
atomic.AddInt64(&counter, 1)
fmt.Println("初始化执行!")
}
func main() {
for i := 0; i < 10; i++ {
once.Do(initOnce)
}
fmt.Printf("计数器：%d\n", counter) // 1
}3.4.2 单例模式
go
package main
import "sync"
type Singleton struct {
data string
}
var (
instance *Singleton
once sync.Once
)
func GetInstance() *Singleton {
once.Do(func() {
instance = &Singleton{data: "hello"}
})
return instance
}3.5 ErrGroup
3.5.1 并发执行任务
go
package main
import (
"errors"
"fmt"
"golang.org/x/sync/errgroup"
)
func main() {
var g errgroup.Group
var result []string
g.Go(func() error {
result = append(result, "task1")
return nil
})
g.Go(func() error {
result = append(result, "task2")
return errors.New("task2 failed")
})
g.Go(func() error {
result = append(result, "task3")
return nil
})
if err := g.Wait(); err != nil {
fmt.Println("错误:", err)
}
fmt.Println("结果:", result)
}3.5.2 并发限制
go
package main
import (
"context"
"fmt"
"golang.org/x/sync/errgroup"
"time"
)
func main() {
semaphore := make(chan struct{}, 3)
g, ctx := errgroup.WithContext(context.Background())
tasks := []string{"task1", "task2", "task3", "task4", "task5"}
for _, task := range tasks {
task := task
g.Go(func() error {
semaphore <- struct{}{}
defer func() { <-semaphore }()
fmt.Printf("开始 %s\n", task)
time.Sleep(time.Second)
fmt.Printf("完成 %s\n", task)
return nil
})
}
if err := g.Wait(); err != nil {
fmt.Println("错误:", err)
}
}3.6 实际案例：TiDB 中的并发控制
3.6.1 Latch (轻量级锁)
go
// tikv/latch/latch.go (简化)
type Latch struct {
slots [256]slot
}
type slot struct {
ch chan struct{}
ref int
}
// 获取锁
func (l *Latch) Acquire(key uint64) {
idx := key % 256
s := &l.slots[idx]
s.mu.Lock()
if s.ref == 0 {
s.ch = make(chan struct{}, 1)
}
s.ref++
s.mu.Unlock()
<-s.ch
}
// 释放锁
func (l *Latch) Release(key uint64) {
idx := key % 256
s := &l.slots[idx]
s.mu.Lock()
s.ref--
if s.ref == 0 {
close(s.ch)
}
s.mu.Unlock()
select {
case s.ch <- struct{}{}:
default:
}
}3.7 本章小结
┌─────────────────────────────────────────────────────────────┐
│ 本章总结 │
├─────────────────────────────────────────────────────────────┤
│ ✓ Mutex/RWMutex: 保护共享资源 │
│ └── 读多写少用 RWMutex │
│ │
│ ✓ WaitGroup: 等待一组任务完成 │
│ │
│ ✓ Cond: 条件变量，线程等待/唤醒 │
│ │
│ ✓ Once: 单次执行 │
│ │
│ ✓ ErrGroup: 并发任务 + 错误处理 + 并发限制 │
│ │
│ ✓ TiDB 等项目的实际应用 │
└─────────────────────────────────────────────────────────────┘