golang-concurrency-patterns

Go 并发模式（生产环境）

概述

当 goroutine 的生命周期明确、通过 context.Context 传播取消信号，并且共享状态得到保护（使用 channel 或锁）时，Go 并发才能扩展。应用这些模式来构建可靠的服务，并避免常见故障：goroutine 泄漏、死锁和数据竞争。

快速入门

默认构建模块

• 使用 context 驱动取消和截止时间。
• 使用 errgroup.WithContext 实现扇出/扇入，并在出错时提前中止。
• 限制并发量（避免创建无界 goroutine），可使用信号量或工作池。
• 优先使用不可变数据；否则使用互斥锁保护共享状态，或让单个 goroutine 拥有该状态。

避免

• 在请求处理程序中启动即不管的 goroutine。
• 在热循环中使用 time.After。
• 从接收方关闭 channel。
• 在没有同步机制的情况下跨 goroutine 共享可变变量。

核心概念

Goroutine 生命周期

将 goroutine 视为资源，为其明确指定所有者和关闭条件。

✅ 正确：通过 context 停止 goroutine

ctx, cancel := context.WithCancel(context.Background())
defer cancel()

go func() {
    ticker := time.NewTicker(250 * time.Millisecond)
    defer ticker.Stop()

    for {
        select {
        case <-ctx.Done():
            return
        case <-ticker.C:
            // do work
        }
    }
}()

❌ 错误：没有停止条件的 goroutine

go func() {
    for {
        doWork() // leaks forever
    }
}()

Channel 与互斥锁（有意识地选择）

• 使用 channel 来建模状态的所有权/串行化，或用于流水线处理工作。
• 使用 互斥锁 来保护具有简单读写模式的共享内存状态。

✅ 正确：由一个 goroutine 拥有 map

type req struct {
    key   string
    reply chan<- int
}

func mapOwner(ctx context.Context, in <-chan req) {
    m := map[string]int{}
    for {
        select {
        case <-ctx.Done():
            return
        case r := <-in:
            r.reply <- m[r.key]
        }
    }
}

✅ 正确：使用互斥锁保护共享 map

type SafeMap struct {
    mu sync.RWMutex
    m  map[string]int
}

func (s *SafeMap) Get(k string) (int, bool) {
    s.mu.RLock()
    defer s.mu.RUnlock()
    v, ok := s.m[k]
    return v, ok
}

模式

1) 使用 errgroup 实现扇出/扇入与取消

使用 errgroup.WithContext 并发运行任务，在出错时取消其他任务，并等待任务完成。

✅ 正确：首次出错时取消

g, ctx := errgroup.WithContext(ctx)

for _, id := range ids {
    id := id // capture
    g.Go(func() error {
        return process(ctx, id)
    })
}

if err := g.Wait(); err != nil {
    return err
}

❌ 错误：WaitGroup 会丢失首个错误，也不会传播取消信号

var wg sync.WaitGroup
for _, id := range ids {
    wg.Add(1)
    go func() {
        defer wg.Done()
        _ = process(context.Background(), id) // ignores caller ctx + captures id
    }()
}
wg.Wait()

2) 限制并发量（信号量模式）

限制并行度，以防止 CPU/内存耗尽和下游过载。

✅ 正确：有界扇出

limit := make(chan struct{}, 8) // max 8 concurrent
g, ctx := errgroup.WithContext(ctx)

for _, id := range ids {
    id := id
    g.Go(func() error {
        select {
        case <-ctx.Done():
            return ctx.Err()
        case limit <- struct{}{}:
        }
        defer func() { <-limit }()

        return process(ctx, id)
    })
}

return g.Wait()

3) 工作池（稳定吞吐量）

使用固定数量的工作线程，以实现稳定的吞吐量和可预测的资源使用。

✅ 正确：通过 context 停止工作池

type Job struct{ ID string }

func runPool(ctx context.Context, jobs <-chan Job, workers int) error {
    g, ctx := errgroup.WithContext(ctx)

    for i := 0; i < workers; i++ {
        g.Go(func() error {
            for {
                select {
                case <-ctx.Done():
                    return ctx.Err()
                case j, ok := <-jobs:
                    if !ok {
                        return nil
                    }
                    if err := handleJob(ctx, j); err != nil {
                        return err
                    }
                }
            }
        })
    }

    return g.Wait()
}

4) 流水线阶段（阶段间扇出）

优先使用单向 channel，并且只从发送方关闭 channel。

✅ 正确：发送方关闭

func stageA(ctx context.Context, out chan<- int) {
    defer close(out)
    for i := 0; i < 10; i++ {
        select {
        case <-ctx.Done():
            return
        case out <- i:
        }
    }
}

❌ 错误：接收方关闭

func stageB(in <-chan int) {
    close(in) // compile error in<-chan; also wrong ownership model
}

5) 避免周期性工作发生泄漏（time.Ticker 与 time.After）

循环中使用 time.NewTicker；避免在热路径中分配 time.After。

✅ 正确：使用 ticker

t := time.NewTicker(1 * time.Second)
defer t.Stop()

for {
    select {
    case <-ctx.Done():
        return
    case <-t.C:
        poll()
    }
}

❌ 错误：在循环中使用 time.After

for {
    select {
    case <-ctx.Done():
        return
    case <-time.After(1 * time.Second):
        poll()
    }
}

决策树

Channel 与互斥锁

• 需要 所有权/串行化（单个写入方、消息传递）→ 使用 channel + 所有者 goroutine
• 需要 共享缓存/map，有许多读取方且只进行简单更新 → 使用 RWMutex
• 需要 简单计数器，且竞争较少 → 使用 atomic

WaitGroup 与 errgroup

• 需要 错误传播 + 取消其他任务 → 使用 errgroup.WithContext
• 只需等待，错误由其他部分处理 → 使用 sync.WaitGroup

带缓冲与不带缓冲的 channel

• 需要 反压 和同步交接 → 使用 不带缓冲的 channel
• 需要 吸收已知大小的突发流量 → 使用 带缓冲的 channel（有目的地设置缓冲区大小）
• 不确定 → 从不带缓冲的 channel 开始并测量；只有在确定存在瓶颈时才添加缓冲区

测试与验证

竞态检测器与不稳定测试控制

使用竞态检测器运行目标测试，并在调试时禁用缓存：

go test -race ./...
go test -run TestName -race -count=1 ./...

使用超时防止测试挂起

✅ 正确：通过 context 设置测试级超时

func TestSomething(t *testing.T) {
    ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
    defer cancel()

    if err := doThing(ctx); err != nil {
        t.Fatal(err)
    }
}

故障排查

症状：死锁（测试挂起，goroutine 阻塞）

操作：

• 在阻塞操作周围添加超时（context.WithTimeout）。
• 检查 channel 所有权：只有发送方会关闭 channel；接收方在 ok == false 时停止。
• 检查信号量模式中是否遗漏了 <-limit 释放操作。

症状：数据竞争（go test -race 报告）

操作：

• 找出被多个 goroutine 修改的共享变量。
• 添加互斥锁，或转换为所有权模型（由单个 goroutine 拥有状态）。
• 避免写入捕获的循环变量。

症状：goroutine 泄漏（内存增长、关闭缓慢）

操作：

• 确保每个 goroutine 都会监听 ctx.Done()。
• 确保停止 time.Ticker，并由发送方关闭通道。
• 避免在请求路径中使用 context.Background()；传递调用方的上下文。

资源

• Go 博客：并发模式：https://go.dev/blog/pipelines
• errgroup：https://pkg.go.dev/golang.org/x/sync/errgroup
• Go 内存模型：https://go.dev/ref/mem