并发编程

发布于 2026-07-06 22:29 · 最后编辑于 2026-07-31 15:51 · 字数 4,403 👁 108 次阅读

#Go #并发 #memory-model #atomic #channel #goroutine #sync #context

系统梳理 Go 并发编程的完整工具箱：从 GMP 调度模型出发，覆盖 sync 包全家桶（Mutex/RWMutex/Cond/WaitGroup/Once/Pool/Map）、原子操作、context 取消传播，扩展原语（Semaphore/SingleFlight/errgroup），以及 Go 内存模型的 happens-before 规则。

章节	说明
GMP 调度模型	goroutine、系统线程、处理器三者的关系
goroutine 基础	启动、主 goroutine 退出、goroutine 泄漏
sync.Mutex 与 sync.RWMutex	互斥锁、读写锁的正确使用与禁忌
Mutex 内部实现与演进	四阶段演进、饥饿模式、state 字段解析
Mutex 4 种易错场景	Lock/Unlock 不配对、复制、重入、死锁
sync.Cond	条件变量的等待/通知机制
sync.WaitGroup	一对多协作，等待所有 goroutine 完成
sync.Once	保证代码只执行一次
sync.Pool	临时对象复用池，降低 GC 压力
sync.Map	并发安全字典，读多写少场景
原子操作 sync/atomic	无锁并发，适合简单计数器
context.Context	取消信号传播、超时控制、跨 goroutine 传值
Channel 应用模式	消息传递、信号通知、任务编排经典模式
Go 内存模型	happens-before 规则，保证并发读写可见性
扩展并发原语	Semaphore、SingleFlight、errgroup
并发模式速查	常见场景的工具选择

GMP 调度模型

Go GMP 调度模型

Go 运行时实现了用户级线程（goroutine），由 GMP 调度器管理：

G（Goroutine）— 用户级线程，由 Go 运行时管理，创建成本极低（初始栈 ~2KB）
M（Machine）  — 系统线程，由 OS 管理
P（Processor）— 逻辑处理器，持有 goroutine 队列，是 G 和 M 之间的中介

graph LR
    G1["G1"] --> P1["P1"]
    G2["G2"] --> P1
    G3["G3"] --> P2["P2"]
    G4["G4"] --> P2
    P1 --> M1["M1（OS Thread）"]
    P2 --> M2["M2（OS Thread）"]
    style P1 fill:#cfc,stroke:#060
    style P2 fill:#cfc,stroke:#060

P 的数量由 GOMAXPROCS 控制（默认 = CPU 核数），决定了真正的并行度。

调度要点：

当 G 发生系统调用（如 I/O）时，M 与 P 解绑，P 接管其他 G 继续运行，M 等待系统调用返回
G 被抢占：Go 1.14+ 支持异步抢占，长时间运行的 G 不再独占 P
Work stealing：P 的本地队列空了，会从其他 P 的队列"偷"任务

goroutine 基础

启动

go func() {
    // 在新 goroutine 中执行
}()

go myFunc(arg1, arg2)  // 函数调用前加 go

主 goroutine 退出即程序退出

func main() {
    go longTask()   // 启动后台任务
    // 如果 main 直接返回，longTask 不会执行完
}

// 解决方案1：time.Sleep（不推荐，不精确）
// 解决方案2：sync.WaitGroup
// 解决方案3：channel 阻塞等待

goroutine 泄漏

goroutine 泄漏是 Go 中最常见的资源泄漏：

// ❌ 泄漏：ch 没有值时 goroutine 永远阻塞
func leak() {
    ch := make(chan int)
    go func() {
        val := <-ch  // 永远阻塞，goroutine 无法退出
        _ = val
    }()
}

// ✅ 修复：用 context 或 done channel 传递退出信号
func noLeak(ctx context.Context) {
    ch := make(chan int)
    go func() {
        select {
        case val := <-ch:
            _ = val
        case <-ctx.Done():  // 收到取消信号，退出
            return
        }
    }()
}

sync.Mutex 与 sync.RWMutex

../../assets/03 GoLang 并发编程/file 20260605123653215

sync.Mutex（互斥锁）

var mu sync.Mutex
var count int

func increment() {
    mu.Lock()
    defer mu.Unlock()  // 推荐用 defer，确保解锁
    count++
}

使用禁忌：

不能复制已使用过的 Mutex（会连同锁定状态一起复制，导致死锁）
不能在不同 goroutine 中 Lock/Unlock（Lock 和 Unlock 必须成对出现在同一 goroutine 中）
避免嵌套加锁（同一 goroutine 重复 Lock → 死锁，Mutex 不可重入）

sync.RWMutex（读写锁）

var rwmu sync.RWMutex
var data map[string]int

// 读（共享锁，允许多个 goroutine 同时读）
func read(key string) int {
    rwmu.RLock()
    defer rwmu.RUnlock()
    return data[key]
}

// 写（排他锁）
func write(key string, val int) {
    rwmu.Lock()
    defer rwmu.Unlock()
    data[key] = val
}

适用场景：读多写少（如配置缓存、注册表）。读远多于写时，RWMutex 优于 Mutex。

写锁饥饿问题：若读操作持续不断，写锁可能长时间等待。Go 的实现中，已有写锁等待时，新的读锁请求会被阻塞（防止写饥饿）。

sync.Cond

条件变量用于协调 goroutine 之间的等待/通知：

var mu sync.Mutex
var ready bool
cond := sync.NewCond(&mu)

// 等待方（消费者）
go func() {
    cond.L.Lock()
    for !ready {         // 必须用 for，防止虚假唤醒
        cond.Wait()      // 原子地：释放锁 → 等待 → 重新加锁
    }
    // 执行业务逻辑
    cond.L.Unlock()
}()

// 通知方（生产者）
mu.Lock()
ready = true
cond.Signal()    // 唤醒一个等待的 goroutine
// cond.Broadcast()  // 唤醒所有等待的 goroutine
mu.Unlock()

Wait() 的三个步骤（原子执行）：

把当前 goroutine 加入通知队列
解锁关联的锁
挂起 goroutine（暂停执行）

被唤醒后，Wait() 会重新加锁才返回。

适用场景：生产者-消费者模型、多个 goroutine 等待某个条件成立。

sync.WaitGroup

协调一个 goroutine 等待多个 goroutine 完成：

var wg sync.WaitGroup

wg.Add(3)  // 先统一 Add
for i := 0; i < 3; i++ {
    go func(id int) {
        defer wg.Done()  // goroutine 结束时 -1
        // 执行任务
    }(i)
}
wg.Wait()  // 阻塞直到计数器归零

标准模式：先统一 Add，再并发 Done，最后 Wait。

禁忌：

Add 和 Wait 并发调用（可能引发 panic）
计数器降为负数（panic）
复制已使用过的 WaitGroup

sync.Once

保证某段代码在多 goroutine 环境中只执行一次（常用于单例初始化）：

var once sync.Once
var instance *MyService

func getInstance() *MyService {
    once.Do(func() {
        instance = &MyService{}
        instance.init()
    })
    return instance
}

特点：

Do 的函数执行完后，后续调用直接跳过（即使传入不同函数）
若 Do 的函数内部 panic，Once 仍视为已完成（后续调用不会重试）

sync.Pool

临时对象复用池，降低频繁申请/释放对象时的 GC 压力：

var bufPool = sync.Pool{
    New: func() interface{} {
        return new(bytes.Buffer)
    },
}

func processRequest(data []byte) {
    buf := bufPool.Get().(*bytes.Buffer)
    buf.Reset()
    defer bufPool.Put(buf)  // 用完归还
    
    buf.Write(data)
    // 使用 buf
}

关键特性：

GC 时 Pool 中的对象可能被回收（不保证持久存储）
不适合存储连接（连接有状态，不宜随意重用）
不能用 Pool 缓存 channel 等有显式关闭语义的对象
典型用途：fmt 包内部大量使用 Pool 复用 pp 对象，encoding/json 用 Pool 复用 Buffer

sync.Map

并发安全的字典，适合读多写少或键集合稳定的场景：

var m sync.Map

// 写入
m.Store("key", "value")

// 读取
v, ok := m.Load("key")

// 不存在则写入（原子操作）
v, loaded := m.LoadOrStore("key", "default")

// 读取后删除
v, loaded = m.LoadAndDelete("key")

// 删除
m.Delete("key")

// 遍历（不保证顺序）
m.Range(func(k, v interface{}) bool {
    fmt.Println(k, v)
    return true  // 返回 false 停止遍历
})

内部实现（双层结构）：

read：只读的 atomic.Value，无锁读取，适合高频读
dirty：加锁的 map，写操作在这里

适用 vs 不适用：

适用	不适用
读多写少	频繁写入
键集合稳定（如注册表、路由表）	频繁新增/删除键
多个 goroutine 读不同的键	需要迭代计算的聚合操作

频繁写入时，sync.Mutex + 普通 map 性能更好。

原子操作 sync/atomic

对基本数值类型的无锁原子操作，性能高于 Mutex：

import "sync/atomic"

var count int64

// 原子加
atomic.AddInt64(&count, 1)

// 原子读（防止读到中间状态）
v := atomic.LoadInt64(&count)

// 原子写
atomic.StoreInt64(&count, 100)

// CAS（Compare-And-Swap，乐观锁基础）
old, new := int64(10), int64(20)
swapped := atomic.CompareAndSwapInt64(&count, old, new)
// 只有当前值 == old 时才将其改为 new，返回是否成功

Go 1.19+ 新增泛型原子类型：

var atomicInt atomic.Int64  // 比函数式 API 更安全
atomicInt.Add(1)
atomicInt.Load()
atomicInt.Store(100)
atomicInt.CompareAndSwap(10, 20)

适用场景：简单计数器、状态标志（int32 表示 0/1）。对于复杂的多字段协调，仍需 Mutex。

Mutex 内部实现与演进

Mutex 的演进经历四个阶段（Russ Cox 2008 年初版 → 当前 Go 版本）：

阶段	特征	问题
初版	CAS flag，等待者阻塞	新 goroutine 不能抢锁
给新人机会	新 goroutine 可参与竞争（自旋）	等待者可能长时间饥饿
多给些机会	被唤醒的 goroutine + 新 goroutine 都可以竞争	饥饿更严重
解决饥饿（现版本）	引入饥饿模式，等待超 1ms 直接转为饥饿模式	—

state 字段结构

Mutex.state（32位 int）：
  bit 0：是否被锁定（mutexLocked）
  bit 1：是否有 goroutine 被唤醒（mutexWoken）
  bit 2：是否处于饥饿模式（mutexStarving）
  bit 3-31：等待锁的 goroutine 数量

正常模式 vs 饥饿模式

正常模式：等待的 goroutine 按 FIFO 排队，但唤醒后需与新进来的 goroutine 竞争（新来的有 CPU 优势）
饥饿模式（等待超过 1ms）：锁直接交给队列最前面的 goroutine，新来的不参与竞争，保证公平性
持有锁后若发现自己是队列最后一个，或等待时间 < 1ms，退出饥饿模式

Mutex 4 种易错场景

1. Lock/Unlock 不成对（死锁）

// ❌ 发生 panic 时 Unlock 可能不被调用
mu.Lock()
// panic here
mu.Unlock()

// ✅ 用 defer 保证配对
mu.Lock()
defer mu.Unlock()

2. 复制已使用的 Mutex

type SafeCounter struct {
    mu sync.Mutex
    v  int
}

// ❌ 值传递复制了 Mutex 的内部状态
func bad(c SafeCounter) {
    c.mu.Lock()
    defer c.mu.Unlock()
    c.v++
}

// ✅ 传指针
func good(c *SafeCounter) {
    c.mu.Lock()
    defer c.mu.Unlock()
    c.v++
}

go vet 会检测 Mutex 的复制问题。

3. 重入（Go 的 Mutex 不可重入）

// ❌ 死锁：同一 goroutine 重复 Lock
func (t *Thing) Foo() {
    t.mu.Lock()
    defer t.mu.Unlock()
    t.Bar()  // Bar 内部也 Lock → 死锁
}

func (t *Thing) Bar() {
    t.mu.Lock()
    defer t.mu.Unlock()
    // ...
}

// ✅ 拆分有锁/无锁版本
func (t *Thing) Bar() {
    t.mu.Lock()
    defer t.mu.Unlock()
    t.barLocked()  // 不加锁的内部版本
}

func (t *Thing) barLocked() { ... }  // 调用方保证已持有锁

4. 死锁（多锁顺序不一致）

// ❌ goroutine A 持有 mu1 等待 mu2，goroutine B 持有 mu2 等待 mu1 → 死锁
go func() { mu1.Lock(); mu2.Lock() /* ... */ }()
go func() { mu2.Lock(); mu1.Lock() /* ... */ }()

// ✅ 固定加锁顺序（所有地方都先 mu1 再 mu2）

context.Context

context 包用于在 goroutine 树中传播取消信号、截止时间和请求范围的值：

四种创建方式

// 1. 根 context（不可取消）
ctx := context.Background()

// 2. 可取消 context
ctx, cancel := context.WithCancel(parent)
defer cancel()  // 务必调用 cancel，否则资源泄漏

// 3. 超时 context
ctx, cancel := context.WithTimeout(parent, 3*time.Second)
defer cancel()

// 4. 截止时间 context
ctx, cancel := context.WithDeadline(parent, time.Now().Add(3*time.Second))
defer cancel()

在 goroutine 中监听取消

func doWork(ctx context.Context) {
    for {
        select {
        case <-ctx.Done():
            // context 被取消或超时
            fmt.Println("cancelled:", ctx.Err())
            return
        default:
            // 正常工作
            time.Sleep(100 * time.Millisecond)
        }
    }
}

传递请求范围的值

// 设置值（key 建议用自定义类型，避免与其他包冲突）
type ctxKey string
ctx = context.WithValue(ctx, ctxKey("requestID"), "req-123")

// 读取值
reqID, ok := ctx.Value(ctxKey("requestID")).(string)

context.WithValue 使用原则：只传递请求范围的元数据（如 trace ID、用户认证信息），不要用它传递函数的必要参数（那应该显式传参）。

取消传播链

parent ctx
    ├── child1 ctx（WithCancel）
    │       └── grandchild ctx
    └── child2 ctx（WithTimeout）

parent 取消 → child1、child2 及其所有子孙都被取消

Channel 应用模式

Go Channel 类型对比

模式1：数据传递（流水线）

func gen(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        for _, n := range nums { out <- n }
        close(out)
    }()
    return out
}

func sq(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        for n := range in { out <- n * n }
        close(out)
    }()
    return out
}

// 流水线组合
c := sq(sq(gen(2, 3)))

模式2：信号通知（done channel）

done := make(chan struct{})  // 用空结构体，不占内存

// 发信号
close(done)      // 广播给所有监听者
done <- struct{}{}  // 通知单个

// 监听
select {
case <-done:
    return
default:
    // 继续工作
}

模式3：Fan-out / Fan-in（分发与汇聚）

// Fan-out：一个输入通道，分发给多个 worker
func fanOut(in <-chan int, n int) []<-chan int {
    outs := make([]<-chan int, n)
    for i := 0; i < n; i++ {
        outs[i] = worker(in)
    }
    return outs
}

// Fan-in：多个输入通道合并为一个
func fanIn(cs ...<-chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup
    for _, c := range cs {
        wg.Add(1)
        go func(c <-chan int) {
            defer wg.Done()
            for v := range c { out <- v }
        }(c)
    }
    go func() { wg.Wait(); close(out) }()
    return out
}

模式4：超时控制

select {
case result := <-ch:
    fmt.Println(result)
case <-time.After(3 * time.Second):
    fmt.Println("timeout")
}

模式5：限流（令牌桶）

// 利用带缓冲 channel 实现并发数限制
limit := make(chan struct{}, 10)  // 最多 10 个并发

for _, task := range tasks {
    limit <- struct{}{}  // 占一个槽
    go func(t Task) {
        defer func() { <-limit }()  // 释放槽
        process(t)
    }(task)
}

Go 内存模型

Go 内存模型定义了在什么条件下，一个 goroutine 对变量的写操作对另一个 goroutine 的读操作可见。核心概念是 happens-before（先行发生）关系。

happens-before 规则

操作	happens-before 关系
`go` 语句	goroutine 启动 happens-before goroutine 内的代码
`ch <- v` 发送	happens-before 对应的 `v := <-ch` 接收完成
关闭通道 `close(ch)`	happens-before `<-ch` 返回零值
带缓冲通道：第 n 次接收	happens-before 第 n+cap 次发送完成
`sync.Mutex.Unlock()`	happens-before 下一次 `Lock()` 返回
`sync/atomic` 操作	提供 sequential consistency（顺序一致性）

常见的可见性陷阱

// ❌ 不保证可见性：x 的写对另一个 goroutine 不可见
var x int
go func() { x = 1 }()
fmt.Println(x)  // 可能打印 0

// ✅ 通过 channel 建立 happens-before 关系
done := make(chan struct{})
go func() {
    x = 1
    done <- struct{}{}  // 写 x happens-before 发送 done
}()
<-done
fmt.Println(x)  // 保证打印 1

// ✅ 使用 sync/atomic 保证可见性
var x int64
go func() { atomic.StoreInt64(&x, 1) }()
// 在适当同步后
fmt.Println(atomic.LoadInt64(&x))

实践原则：不要依赖偶然的顺序。只要有并发读写，就必须通过 channel、mutex 或 atomic 建立明确的 happens-before 关系。

扩展并发原语

这些原语在 golang.org/x/sync 包中（需 go get），用于处理 sync 标准库无法覆盖的场景。

Semaphore（信号量）

控制同时访问资源的 goroutine 数量上限（广义互斥锁，N=1 时等价于 Mutex）：

import "golang.org/x/sync/semaphore"

// 创建最多允许 10 个并发的信号量
sem := semaphore.NewWeighted(10)

ctx := context.Background()
// 获取 1 个资源（阻塞直到有空位）
if err := sem.Acquire(ctx, 1); err != nil {
    return err
}
defer sem.Release(1)

// 尝试获取（非阻塞）
if sem.TryAcquire(1) {
    defer sem.Release(1)
    // ...
}

适用场景：限制数据库连接并发数、限制 HTTP 请求并发数、批量任务并行度控制。

SingleFlight（请求合并）

合并同一时刻对同一 key 的重复请求，只执行一次，结果共享给所有等待者：

import "golang.org/x/sync/singleflight"

var g singleflight.Group

func getData(key string) (interface{}, error) {
    // 无论有多少并发请求同一 key，函数只执行一次
    v, err, shared := g.Do(key, func() (interface{}, error) {
        return db.Query(key)  // 只有一个 goroutine 真正执行查询
    })
    fmt.Printf("shared=%v\n", shared)  // true 表示结果被多个调用者共享
    return v, err
}

适用场景：防止缓存击穿（缓存失效时大量请求同时打到数据库）。

errgroup（带错误的 WaitGroup）

WaitGroup 的增强版，支持收集第一个错误并取消所有 goroutine：

import "golang.org/x/sync/errgroup"

g, ctx := errgroup.WithContext(context.Background())

for _, url := range urls {
    url := url  // 循环变量捕获
    g.Go(func() error {
        resp, err := http.Get(url)
        if err != nil {
            return err  // 返回 error 会触发 ctx 取消
        }
        defer resp.Body.Close()
        return process(resp)
    })
}

// 等待所有 goroutine 完成，返回第一个非 nil error
if err := g.Wait(); err != nil {
    log.Fatal(err)
}

与 WaitGroup 的区别：

errgroup.Go 启动 goroutine，WaitGroup.Add(1) + go + Done()
errgroup.Wait() 返回第一个错误
errgroup.WithContext 版本：任意 goroutine 返回错误 → ctx 被取消 → 其他 goroutine 通过 ctx.Done() 得知

并发模式速查

场景	推荐工具	说明
保护共享数据（读写均衡）	`sync.Mutex`	简单可靠
保护共享数据（读多写少）	`sync.RWMutex`	并发读性能更好
等待多个 goroutine 完成	`sync.WaitGroup`	标准模式
单例初始化	`sync.Once`	并发安全的懒加载
goroutine 间等待条件	`sync.Cond`	生产者-消费者
临时对象复用	`sync.Pool`	降低 GC 压力
并发安全读多写少字典	`sync.Map`	键集合稳定时
简单计数器/状态标志	`sync/atomic`	无锁，性能最高
取消/超时传播	`context.Context`	所有新 goroutine 都应接收 ctx
goroutine 间传递数据	`channel`	Go 的并发哲学
限制并发数量	`semaphore.Weighted`	广义 Mutex，控制资源池
防缓存击穿	`singleflight.Group`	相同 key 只执行一次
并发任务 + 错误收集	`errgroup.Group`	WaitGroup + error 增强版

参考资料

《Go 语言核心 36 讲》— 第 16-35 讲（郝林）
《Go 并发编程实战课》— 第 1-20 讲（鸟窝）
The Go Memory Model
Go Concurrency Patterns: Context
Go sync package
golang.org/x/sync
../../05 计算机基础/07 并发与协程/02 协程的实现机制（GMP 模型的底层原理：有栈协程 + M:N 调度）
../../05 计算机基础/07 并发与协程/03 各语言的协程权衡（Go goroutine 在各语言横向对比中的位置）

← 返回列表

暂无评论，来留下第一条吧。

登录或注册后才能发表评论