Go 项目中用 errgroup 并发控制处理批量请求

起因

前阵子接了个小需求：要把数据库里几千个用户的信息，挨个调一遍第三方的接口同步过去。接口单次响应大概 200ms，我第一版图省事直接 for 循环串行调，跑完一遍得七八分钟，测试的同学在旁边等着，脸都绿了。

那还不简单，起 goroutine 呗。然后我就写出了这种东西：

for _, u := range users {
    go syncUser(u)
}

跑起来确实快，快到第三方直接给我限流了，返回一堆 429，中间还夹着几个连接被重置的。而且这样写错误也没法收集，哪个用户同步失败了完全不知道，反正就是跑完了，成没成功另说。为什么当初写代码的时候没人提醒我呢，可能是我瞎没搜到吧...

去翻了翻别人是怎么处理的，才发现 golang.org/x/sync 里早就给好了现成的工具：errgroup。这玩意儿用起来是真顺手，下面记录一下实测可用的写法和几个踩过的坑。

errgroup 是个啥

errgroup 你可以理解成一个带错误传递的 WaitGroup。WaitGroup 只管等所有 goroutine 跑完，出错的事它一概不管；errgroup 会把其中第一个非 nil 的错误记下来，Wait 的时候还给你。

先安装：

go get golang.org/x/sync

最简版本长这样：

g := new(errgroup.Group)

for _, u := range users {
    u := u
    g.Go(func() error {
        return syncUser(u)
    })
}

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

这样就能并发跑了，还能拿到错误。注意 u := u 这行，Go 1.22 之前循环变量是共享的，不拷贝一份的话闭包里拿到的可能全是最后一个值。虽然 1.22 之后不用这么写了，但很多老项目还没升上去，写上总没坏处。

控制并发数

光并发还不够，上面那种写法有多少任务就起多少 goroutine，几十万个任务再叠上第三方的限流，服务器当场就给你表演一个雪崩。errgroup 的 SetLimit 就是干这个的：

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

for _, u := range users {
    u := u
    g.Go(func() error {
        return syncUser(ctx, u)
    })
}

err := g.Wait()

SetLimit(10) 表示同一时间最多 10 个 goroutine 在跑，满了之后 g.Go 会阻塞住，等有空位了再放新的进去。效果类似拿带缓冲的 channel 当信号量，只是不用自己收发那么麻烦。这个方法是比较新的版本才加进去的，如果你的 go.mod 里 errgroup 太老，升级一下就行。

用 WithContext 及时止损

上面例子里我用了 errgroup.WithContext，这个值得单独说。它的行为是：任何一个 goroutine 返回了非 nil 的错误，ctx 就会被取消。也就是说只要你把 ctx 传进每个请求里，一旦有一个任务挂了，剩下的请求会立刻收到取消信号，不用等全部跑完过后才知道失败了。

func syncUser(ctx context.Context, u User) error {
    req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, body)
    if err != nil {
        return err
    }
    // ...
}

要是你的场景是“失败了也没关系，能同步几个是几个”，那就别用 WithContext，直接 new(errgroup.Group) 就行，一个任务失败不影响其他人。

还有一个点：g.Wait() 返回的是第一个非 nil 的错误，不是所有错误。想把每个失败的都记下来的话，得自己在 goroutine 里收集。另外 WithContext 只是发取消信号，Wait 还是会等所有 goroutine 退出来，所以函数里面得老老实实响应 ctx，别把信号忽略了。

整理成完整例子

最后贴一段我在项目里实际用的结构，按索引写回结果，连锁都不用加：

results := make([]Result, len(users))

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

for i, u := range users {
    i, u := i, u
    g.Go(func() error {
        res, err := fetchUser(ctx, u)
        if err != nil {
            log.Printf("sync user %d failed: %v", u.ID, err)
            return err
        }
        results[i] = res
        return nil
    })
}

if err := g.Wait(); err != nil {
    // 有失败的处理，比如把失败的那部分捞出来重试
}

每个 goroutine 只写 results 里属于自己的那个下标，互不干扰，所以不用 mutex。要是任务数量在跑之前不确定，那就老实用锁，或者用 channel 收结果。

小结

总结一下，批量请求这个事，用 errgroup 基本就三步：SetLimit 控并发，WithContext 让失败能及时取消，Wait 收错误。比自己手搓 WaitGroup 加 channel 信号量省事多了，错误也不会漏。

唯一要注意的就是别一激动又写出裸起 goroutine 的代码，虽然跑起来是真快，但快过头了总归是要还的。