并发编程规范:goroutine、channel 与 sync 的纪律

goroutine 的创建成本低:一个 go 语句即可启动,初始栈仅需几 KB。但创建成本低意味着终止必须显式设计——每个 goroutine 都是需要被负责回收的资源。多数并发故障并非源于高深特性,而是几类基础疏漏:goroutine 未定义退出机制、接收方关闭 channel、select 遗漏取消信号、WaitGroup.Add 的调用位置错误。

另一类常见错误是把并发当作性能开关,先引入并发再寻找理由。结构化并发(structured concurrency)的要求正相反:每个 goroutine 都必须有明确的 owner、可预测的退出路径与正确的错误传播;不满足这些条件,不应启动 goroutine。取消与超时的细节见 context 使用规范,本文不再展开。

goroutine 生命周期:先定义退出路径

启动任何 goroutine 之前,必须先确定其退出路径。 没有 context、done channel 或 WaitGroup 约束的 fire-and-forget goroutine 无法停止也无法等待,每次调用都会累积一个实例,持续占用内存与调度资源:

// ❌ fire-and-forget:无法停止、无法等待,进程生命周期内持续泄漏
func startWorker() {
    go func() {
        for {
            doWork() // 无退出条件
        }
    }()
}

// ✅ 尊重 context 取消,调用方可以等待
func startWorker(ctx context.Context) *sync.WaitGroup {
    var wg sync.WaitGroup
    wg.Add(1)
    go func() {
        defer wg.Done()
        for {
            select {
            case <-ctx.Done():
                return
            default:
                doWork(ctx)
            }
        }
    }()
    return &wg
}

另一项必须处理的机制:子 goroutine 中的 panic 会导致整个进程终止,而非仅结束该 goroutine。因此生产代码应在每个 goroutine 的入口函数中通过 defer 注册 recover,并记录捕获到的 panic 值。

在引入并发之前,还应确认同步实现确实无法满足需求——能用同步代码解决的问题不应引入并发。启动一个 goroutine 前,三个问题必须有明确答案:它如何退出?能否被通知停止(传入 context 或 done channel)?能否被等待(sync.WaitGroup 或 errgroup)?

channel 的关闭责任与所有权

channel 应由发送方关闭,接收方不应调用 close。 接收方关闭后,发送方再写入会直接 panic。惯用写法是生产者在结束时 defer close(ch):

// 生产者拥有并关闭输出 channel
func generate(ctx context.Context) <-chan int {
    ch := make(chan int)
    go func() {
        defer close(ch) // 发送方负责关闭
        for i := 0; ; i++ {
            select {
            case ch <- i:
            case <-ctx.Done():
                return
            }
        }
    }()
    return ch
}

函数签名应尽量使用单向类型(chan<- / <-chan)。接收双向 chan int 参数时,调用方可能误发送也可能误关闭;单向类型让编译器在构建期拒绝这类操作。这也是所有权原则的体现:创建者拥有 channel,再通过单向类型将发送权与接收权分配给不同角色。

channel 应传递值的副本,不应传递指针。"通过通信共享内存"的前提是所有权随消息转移(所有权转移);发送指针会形成隐式的共享内存,破坏该模型,并引入数据竞争(data race)。需要传递大对象时,应传递不可变值,或在注释中明确约定"接收方不得修改"。

默认应使用无缓冲 channel:它使发送方与接收方同步,交接点明确。缓冲大小 1 保留给信号 channel(如 done := make(chan struct{}, 1))。更大的缓冲会掩盖背压(backpressure)问题;使用时必须有测量依据,并在注释中说明为什么是这个 N、缓冲填满之后的语义是什么。

select 中的取消信号

select 必须包含 ctx.Done() 分支,发送分支也不例外。 阻塞在 out <- v 上的 goroutine,是下游提前退出后最常见的泄漏来源:

// 读和写都监听取消
func process(ctx context.Context, in <-chan Task, out chan<- Result) {
    for {
        select {
        case <-ctx.Done():
            return
        case task, ok := <-in:
            if !ok {
                return // 上游已关闭
            }
            select {
            case out <- handle(ctx, task):
            case <-ctx.Done():
                return
            }
        }
    }
}

热循环中不应反复调用 time.After:它在每次迭代都新建一个定时器,引入额外的分配与 GC 压力。应改用 time.NewTimer + Reset:

// ❌ 每次迭代都分配一个新 timer
for {
    select {
    case msg := <-ch:
        handle(msg)
    case <-time.After(5 * time.Second):
        handleTimeout()
    }
}

// ✅ 复用一个 timer(Go 1.23+)
timer := time.NewTimer(5 * time.Second)
defer timer.Stop()
for {
    select {
    case msg := <-ch:
        handle(msg)
        timer.Reset(5 * time.Second)
    case <-timer.C:
        handleTimeout()
        timer.Reset(5 * time.Second)
    }
}

Go 1.23 起,Stop 之后从 timer.C 接收保证阻塞、不会读到陈旧值,旧版本所需的"先接收一次以排空通道"步骤不再必要;旧代码仍需处理 Stop 返回 false 时的残留数据。

channel、mutex 与 atomic 的选择

选择依据如下:数据的所有权需要随传递而转移时,使用 channel;仅需保护一块共享状态时,使用互斥锁;仅涉及计数或标志位时,使用 atomic。不应使用 mutex 模拟消息传递,也不应使用 channel 保护单个共享字段——两种写法都会混淆并发模型并降低可读性。

并发编排:WaitGroup 与 errgroup

wg.Add 必须在对应的 go 语句之前调用。 写在 goroutine 内部会与 Wait 形成竞争条件(race condition),Wait 可能在 Add 执行前就返回。Go 1.25+ 可直接使用 wg.Go,不存在 Add/Done 配对出错的空间:

// wg.Go 自动登记,没有 Add/Wait 顺序 bug 的空间
var wg sync.WaitGroup
for _, item := range items {
    wg.Go(func() { process(item) })
}
wg.Wait()

传入 wg.Go 的函数不应 panic——panic 仍会导致整个进程终止。WaitGroup 本身不传播错误,也不会取消兄弟任务;只要涉及错误、取消或并发限制,就应使用 errgroup:

// 首错取消 + 每个 goroutine 只写自己的下标(无竞争)
func fetchAll(ctx context.Context, urls []string) ([]Response, error) {
    g, ctx := errgroup.WithContext(ctx)
    results := make([]Response, len(urls))
    for i, url := range urls {
        g.Go(func() error {
            resp, err := fetch(ctx, url)
            if err != nil {
                return fmt.Errorf("fetching %s: %w", url, err)
            }
            results[i] = resp
            return nil
        })
    }
    if err := g.Wait(); err != nil {
        return nil, err // 错误处理细则见错误处理规范一文
    }
    return results, nil
}

并发量应使用 errgroup.SetLimit(n) 限制,避免无界启动 goroutine。绝大多数场景下它已取代手写 worker pool;仅当需要 per-worker 状态(连接、缓冲)、自定义背压或优先级调度、排空在途任务时才需要手写。若需绕开 errgroup 做更细粒度的控制,可将缓冲 channel 用作信号量(semaphore):sem := make(chan struct{}, 10),启动前 sem <- struct{}{} 获取,defer func() { <-sem }() 释放;需要错误传播时仍应优先 errgroup。

sync 原语速查

锁的声明位置同样有规范:mutex 应作为非导出字段嵌入,紧贴在它保护的字段上方,并用注释说明保护对象——mu sync.Mutex // protects entries。这显著优于通过阅读代码推断保护范围。

一次性初始化的最简形态:

// 多个 goroutine 调用 loadConfig(),配置也只解析一次
var loadConfig = sync.OnceValue(func() *Config {
    return parseConfig("config.yaml")
})

pipeline 与 fan-in

pipeline 是由 channel 串联的一串阶段,每个阶段是一(组)goroutine:从上游读取、处理、向下游写出。以下三条纪律缺一不可,违反任何一条,该阶段都会成为泄漏点:

• 每个阶段都必须 select ctx.Done(),保证提前取消时能够停止。
• 每个阶段只关闭自己的输出 channel,与上述所有权规则一致;下游通过 for v := range in 接收关闭信号。
• fan-in 的输出由唯一的聚合 goroutine 关闭:多个发送者共享输出时,每个发送者登记一个 WaitGroup 计数,由独立 goroutine 在 wg.Wait() 之后 close(out):

// fan-in:合并多个输入 channel,收齐后统一关闭
func fanIn(ctx context.Context, ins ...<-chan Result) <-chan Result {
    out := make(chan Result)
    var wg sync.WaitGroup
    for _, in := range ins {
        wg.Add(1)
        go func(in <-chan Result) {
            defer wg.Done()
            for v := range in {
                select {
                case out <- v:
                case <-ctx.Done():
                    return
                }
            }
        }(in)
    }
    go func() { wg.Wait(); close(out) }()
    return out
}

fan-out 相反:多个 goroutine 读取同一个输入 channel,将 CPU 密集型处理分布到多个核心,再经 fan-in 汇合。

数据变换不需要并发时,应优先使用 Go 1.23+ 的 range-over-func 迭代器,而非启动 goroutine:数据源本身是顺序的(slice、数据库游标)、处理是 CPU 密集且无并行收益时,迭代器没有 goroutine 与 channel 的开销,并且天然惰性。只有阶段涉及 I/O、需要真并行或吞吐不均时,才有必要引入 pipeline。需要流畅 API 的顺序变换,可参考 samber/ro 这类库。

goroutine 泄漏检测

应在 TestMain 中接入 go.uber.org/goleak,使 goroutine 泄漏在 CI 阶段即可暴露:

func TestMain(m *testing.M) {
    goleak.VerifyTestMain(m)
}

此外,CI 中应始终执行 go test -race ./...。数据竞争几乎无法通过 code review 发现,竞态检测器能够可靠识别。

线上排查依赖以下手段:runtime.NumGoroutine() 观察数量趋势;/debug/pprof/goroutine?debug=2 导出全量栈,定位阻塞点;Go 1.27 起 goroutine leak profile 在 runtime/pprof 正式可用(Go 1.26 还是 GOEXPERIMENT 实验特性),/debug/pprof/goroutineleak 端点可直接指出哪个 goroutine 阻塞在哪一次收发操作上。泄漏或死锁发生后的定位与复盘,见 故障排查方法论。

速查清单

编写并发代码与代码审查时逐项核对:

• 启动前必须确定退出机制;没有退出路径的 goroutine 一律不启动。
• 每个 goroutine 的入口应注册 recover:子 goroutine 的 panic 会终止整个进程。
• 能用同步实现解决的问题,不引入并发。
• 只有发送方关闭 channel;多发送者场景用 WaitGroup 加聚合 goroutine 统一关闭。
• 函数签名使用单向 channel,把发送与关闭约束交给编译器。
• channel 传递值副本,不传指针;大对象传不可变值。
• 默认无缓冲;缓冲 1 留给信号 channel;更大的缓冲必须有测量依据和注释。
• select 必须带 ctx.Done() 分支,发送分支也不例外。
• 热循环避免 time.After,使用 NewTimer + Reset(Go 1.23+ 无需排空通道)。
• wg.Add 必须在 go 之前;Go 1.25+ 直接使用 wg.Go。
• 需要错误传播、取消、限流时使用 errgroup,避免无界启动 goroutine。
• 临界区保持简短,不跨 I/O 持锁;RWMutex 禁止锁升级;mutex 声明紧贴其保护的字段。
• 计数与标志使用类型化 atomic(Go 1.19+);读多写少才使用 sync.Map;昂贵调用去重使用 singleflight。
• sync.Pool 对象 Put 前先 Reset,不存持久状态。
• 顺序数据变换优先 range-over-func;不需要并发的工作不启动 goroutine。
• goleak 接入 TestMain,go test -race ./... 纳入 CI;线上通过 pprof 的 goroutine 与 goroutineleak 端点排查。

延伸阅读:context 使用规范 详述取消与超时的传播;errgroup 返回错误的包装与判断,见 错误处理规范。