文章

并发原语

#golang#sync#context#atomic

并发原语

包总览

包核心类型用途
syncMutex, RWMutex, WaitGroup, Once, Cond, Map, Pool同步原语
sync/atomicInt32/64, Uint32/64, Pointer, Bool, Value原子操作
contextContext, WithCancel, WithTimeout, WithValue取消/超时/传值

并发模型（goroutine/channel/select）详见 02-并发模型，本篇聚焦同步原语。

sync 包详解

Mutex

var mu sync.Mutex

// 方式1：手动 Lock/Unlock
mu.Lock()
counter++
mu.Unlock()

// 方式2：defer（推荐，防止遗漏 Unlock）
mu.Lock()
defer mu.Unlock()
counter++
// 封装为安全计数器
type Counter struct {
    mu    sync.Mutex
    count int
}

func (c *Counter) Inc() {
    c.mu.Lock()
    defer c.mu.Unlock()
    c.count++
}

func (c *Counter) Get() int {
    c.mu.Lock()
    defer c.mu.Unlock()
    return c.count
}

RWMutex

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

func (c *Cache) Get(key string) (string, bool) {
    c.mu.RLock()         // 读锁（多个 goroutine 可同时持有）
    defer c.mu.RUnlock()
    val, ok := c.data[key]
    return val, ok
}

func (c *Cache) Set(key, val string) {
    c.mu.Lock()          // 写锁（排他）
    defer c.mu.Unlock()
    c.data[key] = val
}
对比MutexRWMutex
读并发✗✓
写并发✗✗
读写互斥✓✓
适用读写均衡读多写少（>10:1）
开销小稍大（维护读者计数）

WaitGroup

// 基本用法
var wg sync.WaitGroup

for i := 0; i < 10; i++ {
    wg.Add(1)
    go func(id int) {
        defer wg.Done()
        doWork(id)
    }(i)
}
wg.Wait()

// 封装模式
func RunConcurrently[T any](items []T, worker func(T)) {
    var wg sync.WaitGroup
    for _, item := range items {
        wg.Add(1)
        go func(item T) {
            defer wg.Done()
            worker(item)
        }(item)
    }
    wg.Wait()
}

// 带限流
func RunWithLimit[T any](items []T, limit int, worker func(T)) {
    sem := make(chan struct{}, limit)
    var wg sync.WaitGroup

    for _, item := range items {
        wg.Add(1)
        sem <- struct{}{}
        go func(item T) {
            defer wg.Done()
            defer func() { <-sem }()
            worker(item)
        }(item)
    }
    wg.Wait()
}

Once

var (
    once    sync.Once
    dbConn  *sql.DB
)

func GetDB() *sql.DB {
    once.Do(func() {
        var err error
        dbConn, err = sql.Open("mysql", dsn)
        if err != nil {
            log.Fatal(err)
        }
    })
    return dbConn
}

// 即使多个 goroutine 同时调用，init 函数也只执行一次
// 之后的调用直接跳过 Do 中的函数

Cond

// 条件变量：用于"等待某个条件成立"的场景
type BlockingQueue struct {
    mu    sync.Mutex
    cond  *sync.Cond
    items []interface{}
    cap   int
}

func NewBlockingQueue(cap int) *BlockingQueue {
    q := &BlockingQueue{cap: cap}
    q.cond = sync.NewCond(&q.mu)
    return q
}

func (q *BlockingQueue) Put(item interface{}) {
    q.mu.Lock()
    defer q.mu.Unlock()

    // 等待队列不满
    for len(q.items) >= q.cap {
        q.cond.Wait()  // 释放锁 + 等待 + 被唤醒后重新获取锁
    }
    q.items = append(q.items, item)
    q.cond.Signal()  // 通知一个等待的消费者
}

func (q *BlockingQueue) Get() interface{} {
    q.mu.Lock()
    defer q.mu.Unlock()

    // 等待队列不空
    for len(q.items) == 0 {
        q.cond.Wait()
    }
    item := q.items[0]
    q.items = q.items[1:]
    q.cond.Signal()  // 通知一个等待的生产者
    return item
}

sync.Map

var m sync.Map

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

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

// 读取或写入（原子）
val, loaded := m.LoadOrStore("key", "default")
// 如果 key 存在，返回已有值；不存在则存入 default

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

// 遍历
m.Range(func(key, value interface{}) bool {
    fmt.Printf("%v=%v\n", key, value)
    return true  // 返回 false 停止遍历
})

// 载入并删除
val, loaded := m.LoadAndDelete("key")

// 更新（Go 1.20+）
m.Swap("key", "newvalue")
对比map + RWMutexsync.Map
读性能中（需获取读锁）高（无锁读路径）
写性能中低（写时复制）
类型安全✓（泛型）✗（interface{}）
适用读写均衡读多写少、key 稳定不变
内存少多（维护 read 和 dirty 两个 map）

sync.Pool

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

func ProcessRequest(data []byte) string {
    buf := bufPool.Get().(*bytes.Buffer)
    defer func() {
        buf.Reset()
        bufPool.Put(buf)
    }()

    buf.Write(data)
    // 处理...
    return buf.String()
}
// JSON 处理中的 Pool 使用
var jsonBufPool = sync.Pool{
    New: func() interface{} {
        b := make([]byte, 0, 4096)
        return &b
    },
}

func MarshalJSON(v interface{}) ([]byte, error) {
    bufPtr := jsonBufPool.Get().(*[]byte)
    buf := (*bufPtr)[:0]
    defer func() {
        jsonBufPool.Put(bufPtr)
    }()

    // 使用 buf 进行 JSON 编码...
    return buf, nil
}

Pool 注意事项：

  • Pool 对象可能在任意时间被 GC 回收，不保证持久存在
  • 不要用 Pool 存储有状态的对象
  • Pool 适合复用临时对象，减少 GC 压力

sync/atomic 包

基本原子操作

import "sync/atomic"

var count int64

// 加
atomic.AddInt64(&count, 1)        // count++
atomic.AddInt64(&count, -1)       // count--

// 读取
val := atomic.LoadInt64(&count)

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

// 比较并交换（CAS）
swapped := atomic.CompareAndSwapInt64(&count, 100, 200)
// 如果 count==100，则设为 200，返回 true
// 如果 count!=100，不做修改，返回 false

// 交换
old := atomic.SwapInt64(&count, 200)
// 设为 200，返回旧值

atomic.Value（通用类型）

var config atomic.Value

// 存储（类型必须一致）
config.Store(&Config{Timeout: 30 * time.Second})

// 读取
cfg := config.Load().(*Config)

atomic.Bool / atomic.Int64 / atomic.Uint64（Go 1.19+）

// Go 1.19+ 提供了类型安全的原子类型
var ready atomic.Bool
var counter atomic.Int64

ready.Store(true)
if ready.Load() {
    // ...
}

counter.Add(1)
fmt.Println(counter.Load())

// CAS
if counter.CompareAndSwap(10, 20) {
    // 如果当前值是 10，设为 20
}

atomic vs Mutex 对比

对比atomicMutex
操作粒度单个变量代码块
性能~5ns~25ns
适用计数器、标志位复杂临界区
复合操作✗（需要 CAS 循环）✓
可组合性差好

CAS 实现无锁队列

// 无锁栈
type Stack struct {
    head atomic.Pointer[Node]
}

type Node struct {
    value interface{}
    next  *Node
}

func (s *Stack) Push(v interface{}) {
    node := &Node{value: v}
    for {
        oldHead := s.head.Load()
        node.next = oldHead
        if s.head.CompareAndSwap(oldHead, node) {
            return
        }
    }
}

func (s *Stack) Pop() (interface{}, bool) {
    for {
        oldHead := s.head.Load()
        if oldHead == nil {
            return nil, false
        }
        if s.head.CompareAndSwap(oldHead, oldHead.next) {
            return oldHead.value, true
        }
    }
}

context 包

Context 核心方法

// context.Background()：根 context（通常在 main 或请求入口）
// context.TODO()：占位（不确定用哪个时）

// WithCancel：手动取消
ctx, cancel := context.WithCancel(context.Background())
defer cancel()  // 必须调用 cancel，避免资源泄漏

// WithTimeout：超时取消
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()

// WithDeadline：截止时间取消
deadline := time.Now().Add(10 * time.Second)
ctx, cancel := context.WithDeadline(context.Background(), deadline)
defer cancel()

// WithValue：携带值（谨慎使用）
ctx = context.WithValue(ctx, "userID", 12345)

Context 传播

// context 在调用链中传递，取消信号自动传播
func Handler(w http.ResponseWriter, r *http.Request) {
    ctx := r.Context()  // HTTP 请求自带 context

    // 传递给下游
    user, err := svc.GetUser(ctx, userID)
    if err != nil {
        // 检查是否是取消导致的
        if errors.Is(err, context.Canceled) {
            return  // 客户端断开连接
        }
        http.Error(w, err.Error(), 500)
        return
    }
}

func (s *UserService) GetUser(ctx context.Context, id string) (*User, error) {
    // 继续传递
    entity, err := s.repo.FindByID(ctx, id)
    if err != nil {
        return nil, err
    }
    return entity, nil
}

func (r *UserRepo) FindByID(ctx context.Context, id string) (*User, error) {
    // 在 DB 查询中使用 context
    var u User
    err := r.db.WithContext(ctx).First(&u, "id = ?", id).Error
    if err != nil {
        return nil, err
    }
    return &u, nil
}

Context 检查取消

// 方式1：select 监听 ctx.Done()
func longRunning(ctx context.Context) error {
    for {
        select {
        case <-ctx.Done():
            return ctx.Err()  // context.Canceled 或 context.DeadlineExceeded
        default:
            // 继续工作
            if err := processChunk(); err != nil {
                return err
            }
        }
    }
}

// 方式2：在循环中检查
func batchProcess(ctx context.Context, items []Item) error {
    for i, item := range items {
        if err := ctx.Err(); err != nil {
            return err  // context 被取消
        }
        if err := process(item); err != nil {
            return fmt.Errorf("process item %d: %w", i, err)
        }
    }
    return nil
}

// 方式3：在 IO 操作中使用
func readWithTimeout(ctx context.Context, r io.Reader) ([]byte, error) {
    done := make(chan struct{})
    var data []byte
    var err error

    go func() {
        data, err = io.ReadAll(r)
        close(done)
    }()

    select {
    case <-done:
        return data, err
    case <-ctx.Done():
        return nil, ctx.Err()
    }
}

Context 传值最佳实践

// ❌ 不要用 context 传业务参数
ctx = context.WithValue(ctx, "user", user)

// ✓ 用自定义类型作为 key（避免冲突）
type ctxKey string

const (
    ctxKeyUser   ctxKey = "user"
    ctxKeyTraceID ctxKey = "trace_id"
)

func WithUser(ctx context.Context, u *User) context.Context {
    return context.WithValue(ctx, ctxKeyUser, u)
}

func UserFromContext(ctx context.Context) (*User, bool) {
    u, ok := ctx.Value(ctxKeyUser).(*User)
    return u, ok
}
适合用 context 传的值不适合的值
请求 ID / Trace ID业务数据
认证信息（用户 ID）函数参数
日志上下文配置
Tenant ID依赖注入的组件

Context 错误类型

错误触发条件检查方式
context.Canceledcancel() 被调用errors.Is(err, context.Canceled)
context.DeadlineExceeded超时到达errors.Is(err, context.DeadlineExceeded)