English 中文 العربية 日本語 Português

Go: Go 同步原语

Channel 用于 goroutine 间通信，sync 包用于保护共享状态——Go 提供了完整的并发安全工具箱。

当多个 goroutine 需要读写同一个变量时，就需要同步原语来保护。这节课你将掌握 Go sync 包的全部核心工具和选择策略。

1. 你将学到

2. 一个高并发工程师的真实故事

(1) 痛点：100 个 goroutine 同时写计数器，数据全乱

Charlie 是电商平台的后端工程师，他需要统计每秒的订单数量：

"100 个 goroutine 同时处理订单，每个处理完把计数器 +1。上线第一天数据完全不对——统计面板显示每秒 1000 单，但数据库里只有 300 单。老板问我'这 700 单的钱去哪了？'"

他打开代码：

GO 
// 坏代码：并发写共享变量，没有锁
var counter int

func processOrder(orderID string) {
    // ... 处理订单
    counter++  // 非原子操作！等价于：
               // temp = counter
               // temp = temp + 1
               // counter = temp  <- 三个 goroutine 同时执行到这步就乱
}

counter++ 不是原子操作——它分解为读取、加一、写入三步。两个 goroutine 同时读到 counter=10，各自加一后写入，结果 counter=11 而不是 12——这就是经典的数据竞争（data race）。

(2) Go 的解法：Mutex

GO 
// counter.go
package main

import (
    "fmt"
    "sync"
)

type SafeCounter struct {
    mu      sync.Mutex
    value   int
}

func (c *SafeCounter) Increment() {
    c.mu.Lock()
    c.value++  // 受锁保护，同一时间只有一个 goroutine 能执行
    c.mu.Unlock()
}

func (c *SafeCounter) Value() int {
    c.mu.Lock()
    defer c.mu.Unlock()
    return c.value
}

func main() {
    var counter SafeCounter
    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.Printf("最终值: %d (期望 100)\n", counter.Value())
}

输出：

TEXT 📖 仅展示 
最终值: 100 (期望 100)

(3) 收益：有锁 vs 无锁

情况 结果 是否数据竞争
无锁并发 95/100/97 随机 ✅ 有（-race 检测到）
Mutex 保护 100 ❌ 无
atomic 操作 100 ❌ 无
💡 提示： go run -race main.go 可以检测数据竞争。如果并发访问共享变量没有锁保护，-race 会在运行时报出 Warning。建议在 CI/CD 中始终开启 -race。

3. sync.Mutex 互斥锁

(1) Mutex 基础

GO 
package main

import (
    "fmt"
    "sync"
)

type Account struct {
    mu      sync.Mutex
    balance float64
}

func (a *Account) Deposit(amount float64) {
    a.mu.Lock()
    a.balance += amount
    a.mu.Unlock()
}

func (a *Account) Withdraw(amount float64) error {
    a.mu.Lock()
    defer a.mu.Unlock()
    if a.balance < amount {
        return fmt.Errorf("余额不足: have %.2f, need %.2f", a.balance, amount)
    }
    a.balance -= amount
    return nil
}

func (a *Account) Balance() float64 {
    a.mu.Lock()
    defer a.mu.Unlock()
    return a.balance
}

func main() {
    acc := Account{balance: 1000}
    var wg sync.WaitGroup

    // 10 个并发存款
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            acc.Deposit(100)
        }()
    }
    wg.Wait()

    fmt.Printf("余额: %.2f\n", acc.Balance())
}

▶ 示例：defer Unlock 模式（推荐）

GO 
// 推荐：Lock 后立即 defer Unlock
func (a *Account) SafeMethod() {
    a.mu.Lock()
    defer a.mu.Unlock()
    // 所有操作都在锁内
    a.balance += 100
    a.balance -= 50
    // 即使中间有 panic，defer 也能确保 Unlock
}

// 不推荐：手动 Unlock
func (a *Account) UnsafeMethod() {
    a.mu.Lock()
    a.balance += 100
    a.mu.Unlock()  // 如果中间 return 或 panic，不会 Unlock → 死锁
}
▶ 试一试
🔥 易错： Lock 后必须 Unlock。 永远用 defer a.mu.Unlock() 而不是手动在末尾 Unlock——如果有多个 return 路径，任何一个忘记 Unlock 都会导致死锁。

4. sync.RWMutex 读写锁

(1) 读写锁原理

GO 
package main

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

type Cache struct {
    mu    sync.RWMutex
    data  map[string]string
}

func NewCache() *Cache {
    return &Cache{data: make(map[string]string)}
}

// 读操作：RLock（可并发读，不互斥）
func (c *Cache) Get(key string) (string, bool) {
    c.mu.RLock()
    defer c.mu.RUnlock()
    v, ok := c.data[key]
    return v, ok
}

// 写操作：Lock（互斥，阻止所有读和写）
func (c *Cache) Set(key, value string) {
    c.mu.Lock()
    defer c.mu.Unlock()
    c.data[key] = value
}

func main() {
    cache := NewCache()
    var wg sync.WaitGroup

    // 多个并发读（RLock 不互斥）
    for i := 0; i < 5; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            for j := 0; j < 3; j++ {
                if v, ok := cache.Get("key"); ok {
                    fmt.Printf("Reader %d: %s\n", id, v)
                }
                time.Sleep(10 * time.Millisecond)
            }
        }(i)
    }

    // 一个写
    wg.Add(1)
    go func() {
        defer wg.Done()
        time.Sleep(5 * time.Millisecond)
        cache.Set("key", "value")
        fmt.Println("Writer: 写入完成")
    }()

    wg.Wait()
}

▶ 示例：Mutex vs RWMutex 性能对比

GO 
package main

import (
    "fmt"
    "sync"
    "testing"
)

type DataStore struct {
    mu    sync.RWMutex
    value int
}

func BenchmarkMutexRead(b *testing.B) {
    store := DataStore{}
    b.RunParallel(func(pb *testing.PB) {
        for pb.Next() {
            store.mu.Lock()
            _ = store.value
            store.mu.Unlock()
        }
    })
}

func BenchmarkRWMutexRead(b *testing.B) {
    store := DataStore{}
    b.RunParallel(func(pb *testing.PB) {
        for pb.Next() {
            store.mu.RLock()
            _ = store.value
            store.mu.RUnlock()
        }
    })
}

// go test -bench=. -benchmem
▶ 试一试

(2) Mutex vs RWMutex

特性 Mutex RWMutex
读与读 互斥 不互斥
读与写 互斥 互斥
写与写 互斥 互斥
适用场景 读写比 1:1 读远多于写（如缓存）
读性能 慢（串行化） 快（并发读）

5. sync.Once 单次执行

(1) Once 基础

GO 
package main

import (
    "fmt"
    "sync"
)

var (
    config     map[string]string
    configOnce sync.Once
)

func loadConfig() {
    configOnce.Do(func() {
        fmt.Println("初始化配置...")
        config = map[string]string{
            "host": "localhost",
            "port": "8080",
        }
    })
}

func main() {
    var wg sync.WaitGroup
    // 100 个 goroutine 同时调用，但 init 只执行一次
    for i := 0; i < 100; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            loadConfig()
        }()
    }
    wg.Wait()
    fmt.Printf("config: %v\n", config)
}

输出：

TEXT 📖 仅展示 
初始化配置...
配置: map[host:localhost port:8080]

(2) sync.Once vs 手动标志位

方式 线程安全 代码量
if !initialized { ... } ❌ 有 data race 少
sync.Once ✅ Go 保证 最少
init() function ✅ 包加载时执行 最少
💡 提示： sync.Once 保证 Do 中的函数只执行一次，即使被 1000 个 goroutine 同时调用。内部实现用了 atomic 操作 + 互斥锁，非常高效。

6. sync.Map 并发安全 Map

(1) sync.Map 基础

GO 
package main

import (
    "fmt"
    "sync"
)

func main() {
    var m sync.Map
    var wg sync.WaitGroup

    // 并发写入
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func(n int) {
            defer wg.Done()
            m.Store(fmt.Sprintf("key-%d", n), n*100)
        }(i)
    }
    wg.Wait()

    // 读取
    m.Range(func(key, value interface{}) bool {
        fmt.Printf("%s = %d\n", key, value)
        return true
    })

    // LoadOrStore：存在则返回，不存在则写入
    actual, loaded := m.LoadOrStore("key-0", 999)
    fmt.Printf("LoadOrStore: actual=%d, loaded=%v\n", actual, loaded)
}

▶ 示例：sync.Map vs map+Mutex

GO 
package main

import (
    "fmt"
    "sync"
    "testing"
)

// map + Mutex
type MutexMap struct {
    mu    sync.Mutex
    items map[string]int
}

func (m *MutexMap) Store(key string, value int) {
    m.mu.Lock()
    m.items[key] = value
    m.mu.Unlock()
}

func (m *MutexMap) Load(key string) (int, bool) {
    m.mu.Lock()
    v, ok := m.items[key]
    m.mu.Unlock()
    return v, ok
}

func BenchmarkMutexMap(b *testing.B) {
    m := &MutexMap{items: make(map[string]int)}
    b.RunParallel(func(pb *testing.PB) {
        for pb.Next() {
            m.Store("key", 1)
            m.Load("key")
        }
    })
}

func BenchmarkSyncMap(b *testing.B) {
    var m sync.Map
    b.RunParallel(func(pb *testing.PB) {
        for pb.Next() {
            m.Store("key", 1)
            m.Load("key")
        }
    })
}

// go test -bench=. -benchmem
▶ 试一试

(2) sync.Map 适用场景

场景 推荐 原因
key 写入一次 sync.Map 读优化（写少读多）
key 频繁更新 map+Mutex sync.Map 写略慢
key 集合增长 sync.Map 避免锁整个 map
简单场景 map+Mutex 更直观
高性能需要 具体 benchmark 实测决定

7. sync/atomic 原子操作

(1) atomic 基础

GO 
package main

import (
    "fmt"
    "sync/atomic"
)

func main() {
    var counter int64

    // atomic 加一
    newVal := atomic.AddInt64(&counter, 1)
    fmt.Printf("AddInt64: %d\n", newVal)

    // atomic 读取
    val := atomic.LoadInt64(&counter)
    fmt.Printf("LoadInt64: %d\n", val)

    // atomic 写入
    atomic.StoreInt64(&counter, 100)

    // CAS（Compare And Swap）
    swapped := atomic.CompareAndSwapInt64(&counter, 100, 200)
    fmt.Printf("CAS: swapped=%v, val=%d\n", swapped, atomic.LoadInt64(&counter))
}

▶ 示例：atomic 计数器 vs Mutex

GO 📖 仅展示 
package main

import (
    "fmt"
    "sync"
    "sync/atomic"
    "time"
)

type AtomicCounter struct {
    value int64
}

func (c *AtomicCounter) Increment() {
    atomic.AddInt64(&c.value, 1)
}

func (c *AtomicCounter) Value() int64 {
    return atomic.LoadInt64(&c.value)
}

type MutexCounter struct {
    mu    sync.Mutex
    value int
}

func (c *MutexCounter) Increment() {
    c.mu.Lock()
    c.value++
    c.mu.Unlock()
}

func (c *MutexCounter) Value() int {
    c.mu.Lock()
    defer c.mu.Unlock()
    return c.value
}

func main() {
    var wg sync.WaitGroup
    n := 100000

    // atomic 计数器
    atomicStart := time.Now()
    var ac AtomicCounter
    for i := 0; i < n; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            ac.Increment()
        }()
    }
    wg.Wait()
    fmt.Printf("Atomic: %d (%v)\n", ac.Value(), time.Since(atomicStart))

    // Mutex 计数器
    muStart := time.Now()
    var mc MutexCounter
    for i := 0; i < n; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            mc.Increment()
        }()
    }
    wg.Wait()
    fmt.Printf("Mutex: %d (%v)\n", mc.Value(), time.Since(muStart))
}
逻辑代码 56 行（超过 40 行限制，仅展示）

(2) atomic 常用函数

操作 函数 用途
加/减 AddInt64, AddUint32 计数器
读取 LoadInt64, LoadPointer 安全读取
写入 StoreInt64, StorePointer 安全写入
CAS CompareAndSwapInt64 乐观锁
交换 SwapInt64 原子交换

8. race detector

▶ 示例：数据竞争检测

GO 
// race_example.go
package main

import (
    "fmt"
    "sync"
)

func main() {
    var counter int
    var wg sync.WaitGroup

    for i := 0; i < 100; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            counter++  // 数据竞争！
        }()
    }

    wg.Wait()
    fmt.Println(counter)
}
▶ 试一试
BASH 
$ go run -race race_example.go
==================
WARNING: DATA RACE
Read at 0x00c0000... by goroutine X
  main.main.func1()
      race_example.go:14 +0x...

Previous write at 0x00c0000... by goroutine Y
  main.main.func1()
      race_example.go:14 +0x...
==================

9. sync.Pool 对象池

GO 
package main

import (
    "fmt"
    "sync"
)

type User struct {
    Name string
    Age  int
}

var userPool = sync.Pool{
    New: func() interface{} {
        return &User{}
    },
}

func main() {
    // 从池中获取（避免了每次分配新对象）
    u := userPool.Get().(*User)
    u.Name = "Alice"
    u.Age = 28
    fmt.Printf("使用: %+v\n", u)

    // 使用后放回池中
    userPool.Put(u)

    // 下次 Get 复用之前的对象
    u2 := userPool.Get().(*User)
    fmt.Printf("复用: %+v\n", u2)
}

10. 完整示例：高并发计数器

GO 
// concurrent_counter.go
package main

import (
    "fmt"
    "sync"
    "sync/atomic"
    "time"
)

// ---------- 3 种计数器实现 ----------

type Counter interface {
    Increment()
    Value() int64
}

// AtomicCounter
type AtomicCounter struct {
    value int64
}

func (c *AtomicCounter) Increment() {
    atomic.AddInt64(&c.value, 1)
}

func (c *AtomicCounter) Value() int64 {
    return atomic.LoadInt64(&c.value)
}

// MutexCounter
type MutexCounter struct {
    mu    sync.Mutex
    value int64
}

func (c *MutexCounter) Increment() {
    c.mu.Lock()
    c.value++
    c.mu.Unlock()
}

func (c *MutexCounter) Value() int64 {
    c.mu.Lock()
    defer c.mu.Unlock()
    return c.value
}

// ChannelCounter
type ChannelCounter struct {
    ch    chan struct{}
    value int64
    done  chan struct{}
}

func NewChannelCounter() *ChannelCounter {
    c := &ChannelCounter{
        ch:   make(chan struct{}, 100),
        done: make(chan struct{}),
    }
    go func() {
        for range c.ch {
            c.value++
        }
        close(c.done)
    }()
    return c
}

func (c *ChannelCounter) Increment() {
    c.ch <- struct{}{}
}

func (c *ChannelCounter) Value() int64 {
    return atomic.LoadInt64(&c.value)
}

func (c *ChannelCounter) Close() {
    close(c.ch)
    <-c.done
}

// ---------- 基准测试 ----------

func benchmarkCounter(c Counter, goroutines, increments int) time.Duration {
    var wg sync.WaitGroup
    start := time.Now()

    for g := 0; g < goroutines; g++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for i := 0; i < increments; i++ {
                c.Increment()
            }
        }()
    }

    wg.Wait()
    return time.Since(start)
}

func main() {
    const (
        goroutines = 100
        increments = 10000
    )

    fmt.Printf("基准测试: %d goroutines 各执行 %d 次 Increment\n\n", goroutines, increments)

    // 1. Atomic 计数器
    ac := &AtomicCounter{}
    atomicTime := time.Now()
    atomicDone := make(chan struct{})
    go func() {
        defer close(atomicDone)
        var wg sync.WaitGroup
        for g := 0; g < goroutines; g++ {
            wg.Add(1)
            go func() {
                defer wg.Done()
                for i := 0; i < increments; i++ {
                    ac.Increment()
                }
            }()
        }
        wg.Wait()
    }()
    <-atomicDone
    fmt.Printf("AtomicCounter: %d (期望 %d) — %v\n",
        ac.Value(), goroutines*increments, time.Since(atomicTime))

    // 2. Mutex 计数器
    mc := &MutexCounter{}
    mutexTime := time.Now()
    mutexDone := make(chan struct{})
    go func() {
        defer close(mutexDone)
        var wg sync.WaitGroup
        for g := 0; g < goroutines; g++ {
            wg.Add(1)
            go func() {
                defer wg.Done()
                for i := 0; i < increments; i++ {
                    mc.Increment()
                }
            }()
        }
        wg.Wait()
    }()
    <-mutexDone
    fmt.Printf("MutexCounter: %d (期望 %d) — %v\n",
        mc.Value(), goroutines*increments, time.Since(mutexTime))

    // 3. Channel 计数器
    cc := NewChannelCounter()
    channelTime := time.Now()
    channelDone := make(chan struct{})
    go func() {
        defer close(channelDone)
        var wg sync.WaitGroup
        for g := 0; g < goroutines; g++ {
            wg.Add(1)
            go func() {
                defer wg.Done()
                for i := 0; i < increments; i++ {
                    cc.Increment()
                }
            }()
        }
        wg.Wait()
        cc.Close()
    }()
    <-channelDone
    fmt.Printf("ChannelCounter: %d (期望 %d) — %v\n",
        cc.Value(), goroutines*increments, time.Since(channelTime))

    fmt.Println("\n结论: Atomic > Mutex > Channel（性能从高到低）")
}
100% 
sequenceDiagram
    participant G1 as Goroutine 1
    participant G2 as Goroutine 2
    participant G3 as Goroutine 3
    participant M as Mutex
    
    G1->>M: Lock()
    Note over G1,M: G1 持有锁
    G2->>M: Lock() 阻塞
    G3->>M: Lock() 阻塞
    G1->>M: Unlock()
    M-->>G2: 唤醒
    M-->>G3: 仍阻塞
    G2->>M: Lock() 成功
    Note over G2,M: G2 持有锁
    G2->>M: Unlock()
    M-->>G3: 唤醒
    G3->>M: Lock() 成功
🔥 易错： sync.Mutex 不可复制。如果你把包含 Mutex 的 struct 按值传递（不是指针），Mutex 的状态会被复制——导致意外的解锁行为。总是用 *Counter 指针传递。

❓ 常见问题

Q Mutex 和 RWMutex 怎么选？
A 读远多于写（>10:1）→ RWMutex；读写相当或写多 → Mutex。RWMutex 的 RLock 允许并发读，但 Lock 会互斥所有读和写。选错会降低性能。
Q sync.Once 怎么用？
A sync.Once.Do(f) 保证 f 只执行一次，即使 1000 个 goroutine 同时调用。典型用途：懒加载配置、单例模式、只初始化一次的资源。内部用 atomic + Mutex 实现。
Q sync.Map 适用什么场景？
A 官方建议两个场景：(1) key 只写一次但被多次读取（cache）；(2) 多个 goroutine 读/写/遍历不同的 key 集合。除此之外，map + RWMutex 更简单且性能不差。
Q atomic 适用什么场景？
A 简单计数器、状态标志、CAS（Compare And Swap）乐观锁。atomic 比 Mutex 快 10~100 倍，但只能用于 int/uint/pointer 等基本类型。复杂逻辑还是用 Mutex。
Q race detector 怎么用？
A go run -race main.go 或 go test -race ./...。在运行时检测数据竞争——如果有两个 goroutine 并发访问同一变量且至少一个是写操作，就会报 Warning。CI/CD 建议始终开启。
Q sync.Pool 什么时候用？
A 当需要频繁分配和释放临时对象时（如 JSON 解码的 buffer、protobuf 消息）。Pool 缓存已分配的对象减少 GC 压力。注意：Pool 中的对象可能在任何时候被 GC 回收。
Q Mutex 可以重入吗？
A 不能。Go 的 sync.Mutex 是非重入锁——同一个 goroutine 连续 Lock 两次会导致死锁。如果需要在同一个 goroutine 中再次获取锁，用 sync.Mutex 配合递归调用重构，或用 defer 确保正确解锁。
Q Lock 和 RLock 的顺序重要吗？
A 非常重要。RWMutex 不允许在 RLock 期间调用 Lock——会死锁。也不允许在 Lock 期间调用 RLock（写者优先）。建议：写操作只用 Lock，读操作只用 RLock，不要混用。

📖 小节

📝 作业

  1. 基础题（难度⭐）：用 Mutex 保护一个 map[string]int 的并发读写。启动 10 个 writer goroutine 写入，10 个 reader goroutine 读取，验证无数据竞争（-race 检测通过）。

  2. 进阶题（难度⭐⭐）：对比 4 种计数器的性能（atomic / Mutex / RWMutex / channel），统计 100 goroutine × 10000 次 increment 的耗时和正确性。用 -race 验证无数据竞争。

  3. 挑战题（难度⭐⭐⭐）：实现一个读多写少的高并发缓存：要求 (1) 用 RWMutex 保护；(2) 支持 Get/Set/Delete/Range 操作；(3) 支持 TTL 过期（time.Now().After(expireAt)）；(4) 写操作时阻塞所有读操作，但读操作之间可并发；(5) 用 -race 验证通过。

Web-Tutorial.com

Web-Tutorial 技术团队

由多位开发者共同维护的编程教程平台。每篇教程由对应领域的开发者编写和审核，确保内容准确可靠。如发现任何问题，欢迎向我们反馈。

100%

🙏 帮我们做得更好

我们是刚上线的编程教程站，几个人的小团队，精力有限。页面虽经检查，难免还有疏漏——链接失效、排版错乱、内容有误、语言生硬……

如果您发现了，麻烦告诉我们，我们会在收到反馈后第一时间进行修复，再次感谢您的光临 🙏