从入门到深入：并发不只是起 goroutine

上一篇文章我们学会了用 goroutine 起协程、用 channel 传数据、用 WaitGroup 等所有任务跑完。但那只是 Go 并发的「语法」。真实工程里的并发要回答三个更难的问题：并发度多大合适、出错怎么办、重复请求怎么避免打爆下游。

无脑为每个任务起一个 goroutine 是有代价的：goroutine 很便宜但并非免费，每起一个都要分配栈空间；当任务成千上万、每个任务还要占用数据库连接或外部 API 配额时，全量并发反而是灾难。于是有了 worker 池：用固定数量的工人消化一条任务队列，既控制了并发度，也保护了下游资源。

同样地，并发任务里只要有一个失败，是继续等全部跑完再汇总，还是立刻取消其他任务？标准库的 context 配合 errgroup 给出了答案。而面对热点数据，成千上万个并发请求同时打到数据库，用 singleflight 可以把它们合并成一次真正的查询。这些模式不是语法糖，而是 Go 社区在大量生产事故里沉淀下来的工程答案。

语法告诉你「怎么并发」，工程模式告诉你「并发到什么时候该刹车、出错时怎么办」。
本文核心知识点
  • worker 池用固定数量 goroutine 消化任务队列。
  • errgroup并发错误聚合，任一失败取消整组。
  • 扇出扇入多个生产者的结果汇成一条流。
  • singleflight合并并发重复请求，防缓存击穿。
  • race 检测用工具找出数据竞争，而不靠肉眼。
Go 并发模式深入配图：深蓝渐变背景，中央发光的六边形调度核心，四周多个半透明方块 worker 节点通过发光连线相连，左侧彩色小方块经栅栏屏障分发到各节点，右侧光点汇聚成发光箭头指向出口
调度核心把任务分发给固定数量的 worker，结果再汇聚成一条流——worker 池与扇出扇入的视觉写照。

worker 池：把任务队列摊给固定数量的工人

worker 池的基本结构是「一个任务 channel + N 个消费 goroutine」：生产者往 channel 里塞任务，N 个 worker 同时从 channel 取任务执行。channel 本身就是并发安全的队列，所以不需要额外的锁。

func workerPool(jobs <-chan int, workerCount int) {
    var wg sync.WaitGroup
    for w := 0; w < workerCount; w++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for job := range jobs {
                process(job)   // 每个 worker 串行消费
            }
        }()
    }
    wg.Wait()
}

worker 数量怎么定？经验法则是看瓶颈资源：如果是 CPU 密集，用 runtime.GOMAXPROCS 级别；如果是 IO 密集（HTTP、数据库），并发度可以更高，但也要受连接池上限约束。固定 worker 的好处是：队列再长也只是排队，不会瞬间打出 N 倍流量。

关闭时机也很关键：生产者全部塞完后 close(jobs)，所有 worker 消费完剩余任务后，for range 自然退出，wg 归零。这个「生产-消费-关闭」的节奏是 worker 池不出 bug 的核心。

errgroup：并发任务的错误聚合与提前退出

原生 WaitGroup 只等不查错。errgroup 补上了两件事：收集第一个错误，以及通过 context 在出错时取消其余任务。它来自 golang.org/x/sync/errgroup。

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

func runAll(ctx context.Context, urls []string) error {
    g, ctx := errgroup.WithContext(ctx)
    for _, u := range urls {
        u := u
        g.Go(func() error {
            return fetch(ctx, u)   // 任一失败返回 error
        })
    }
    return g.Wait()   // 第一个 error，或 nil
}

errgroup.WithContext 会派生出一个可取消的 ctx：只要任何一个 g.Go 里的函数返回错误，errgroup 内部就会 cancel 这个 ctx，其他任务里监听 ctx 的请求会立刻中断。这种「一个失败、全体撤退」的语义，正是分布式调用里最想要的行为——不白等、不空转。

需要注意 g.Go 不接受可取消的单独任务参数，取消要靠闭包捕获的 ctx 主动配合：函数体内要检查 ctx.Err() 或使用带 ctx 的请求方法，否则取消信号到了，任务还在傻等。errgroup 也支持 SetLimit 限流，效果类似 worker 池的并发上限。

扇出扇入：把并发结果汇成一条流

扇出（fan-out）指把一个任务拆给多个 goroutine 并行处理，扇入（fan-in）指把多个 goroutine 的输出合并到一个 channel 里继续处理。合并的方式是开一个专门的收集 goroutine：

func fanIn(chs ...<-chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup
    for _, ch := range chs {
        wg.Add(1)
        go func(c <-chan int) {
            defer wg.Done()
            for v := range c {
                out <- v
            }
        }(ch)
    }
    go func() { wg.Wait(); close(out) }()
    return out
}

扇出扇入组合起来，就是 Go 最经典的「流水线」模型：读数据 → 并行变换 → 合并 → 下一步。每一段只关心自己的输入输出 channel，段与段之间靠类型解耦。管道里任何一段都可以替换实现而不影响上下游。

写扇入代码时最常踩的坑是：收集 goroutine 往 out 里写，但没人保证 out 会被读，写满就会死锁；或者 wg 没等完就 close(out)，导致下游读到零值。牢记「close 只由生产者执行，且必须在所有写入结束后」，这条纪律能避开大多数 channel 死锁。

singleflight：合并重复请求

缓存击穿是经典问题：某个热点 key 失效的瞬间，几千个并发请求同时打到数据库。每个请求都认为「缓存没数据，我去查库」，数据库瞬间被打爆。singleflight 的思路是：同一时刻对同一个 key 的并发请求，只让第一个真正去查，其余请求共享它的结果。

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

var g singleflight.Group

func getData(ctx context.Context, key string) (string, error) {
    v, err, _ := g.Do(key, func() (interface{}, error) {
        return queryDB(ctx, key)   // 只有第一个请求真正执行
    })
    return v.(string), err
}

g.Do(key, fn) 对相同 key 并发调用时，只会执行一次 fn，其余调用直接拿到同一个返回值。这样热点 key 失效后，数据库只收到一次查询，其他请求在 singleflight 内部排队等结果。它不替代缓存，而是缓存和数据库之间的一道「泄洪闸」。

singleflight 不是银弹：它合并的是「同一进程内」的重复请求，多实例部署时每个实例各合并各的；fn 执行期间的所有调用者共享结果，如果 fn 很慢，所有调用者一起等。用 DoChan 还能拿到逐调用者的通知 channel，配合 context 做超时更灵活。