在 Go 服务里做批量处理时，最常见的形态是：一个任务列表，每个元素触发一次下游调用（HTTP、RPC、数据库查询），希望并发跑得快，但又不能把下游打爆，于是需要限流；同时任何单个元素失败不应该中断整批，而是跑完后汇总所有失败项。手写 sync.WaitGroup + chan error + 信号量 channel 可以做到，但很容易踩到三个坑：错误 channel 容量不足导致 goroutine 泄漏、wg.Wait() 前没有关闭 channel 导致死锁、错误被静默丢弃。
golang.org/x/sync/errgroup 的 SetLimit 正好覆盖这个场景：限流由 group 内部管理，不需要自己维护信号量；Wait() 返回第一个错误，配合 errors.Join 或自定义聚合器就能拿到全部错误。本文给出一套可直接抄进项目的模板，包括并发上限、错误聚合、panic 兜底、context 取消与超时四部分的完整代码，以及实测中遇到的具体报错与修正方式。
三个关键 API 的行为边界
先把 errgroup 的行为说清楚，模板才有依据。以下结论基于 golang.org/x/sync v0.7.0 及以上版本，SetLimit 从 v0.3.0 引入。
g.SetLimit(n)：设置同时运行的Go调用上限为 n。它不是排队。当已有 n 个 goroutine 在跑时，第 n+1 次g.Go(f)会阻塞在Go内部，直到有槽位释放。所以调用Go的那个循环体本身会被限速，这点与 worker pool 语义一致，但要注意别在持有锁时调用Go。g.Go(f)：f 返回非 nil error 时，group 记录该错误，并调用 group 内部的cancel取消gctx。只有第一个错误会被Wait()返回，其余错误不会消失（它们仍被 f 返回给 group），只是不通过返回值暴露，因此需要自己聚合。g.Wait()：等待全部完成后返回首个错误（若使用了WithContext则由 context 派生）。它保证所有已启动的 goroutine 结束后才返回，这一点比裸WaitGroup更安全。
容易忽略的一点：
SetLimit(0)或负数等价于不限制并发，源码里if g.limit < 0 { panic }并不存在，sem为 nil 时直接放行。所以从配置读出的并发数必须自己兜底，否则线上配个 0 就会打出全量并发。
最小可用模板：限流 + errors.Join 聚合
先给一个不依赖额外库的版本。errors.Join 是 Go 1.20 引入的，能把多个 error 合成一个，errors.Is / errors.As 依然可以穿透遍历。缺点是丢失了「哪个索引失败」的上下文，所以下面用 fmt.Errorf 包一层带上下文的错误。
package batch
import (
"context"
"errors"
"fmt"
"time"
"golang.org/x/sync/errgroup"
)
// Task 表示一个待处理元素，Fn 返回 error 即视为该元素失败。
type Task struct {
Index int
Fn func(ctx context.Context) error
}
// Result 汇总执行结果。
type Result struct {
Total int
Succeeded int
Failed int
// Err 是聚合后的错误，nil 表示全部成功。
Err error
// Elapsed 为整体耗时，便于打点。
Elapsed time.Duration
}
// Run 以并发上限 limit 执行全部任务，收集所有错误而不中途退出。
// 传入的 ctx 用于整体超时控制；单个任务失败不会取消其他任务。
func Run(ctx context.Context, tasks []Task, limit int) Result {
start := time.Now()
// 关键点 1：limit 兜底，SetLimit(0) 等于不限并发。
if limit <= 0 {
limit = 1
}
if limit > len(tasks) && len(tasks) > 0 {
limit = len(tasks)
}
// 关键点 2：用不带 cancel-on-error 的 group，避免一个失败把其余任务全取消。
// WithContext 会在有任务返回 error 时取消派生 ctx，这里不需要该行为。
g := new(errgroup.Group)
g.SetLimit(limit)
var (
mu sync.Mutex
errs = make([]error, 0, len(tasks))
succeeded int
)
for _, t := range tasks {
t := t // Go 1.22 以下必须显式捕获；1.22+ 可省略
g.Go(func() error {
if err := safeCall(ctx, t.Fn); err != nil {
mu.Lock()
errs = append(errs, fmt.Errorf("task[%d]: %w", t.Index, err))
mu.Unlock()
// 返回 nil，不让 group 记录错误、也不触发任何取消。
return nil
}
mu.Lock()
succeeded++
mu.Unlock()
return nil
})
}
// g.Wait() 对所有任务返回 nil，因此这里必为 nil；保留判断以防后续改动。
_ = g.Wait()
res := Result{
Total: len(tasks),
Succeeded: succeeded,
Failed: len(errs),
Elapsed: time.Since(start),
}
if len(errs) > 0 {
// errors.Join 保留全部错误，errors.Is/As 可逐层探测。
res.Err = errors.Join(errs...)
}
return res
}
// safeCall 把 panic 转成 error，避免一个任务 panic 拖垮整个进程。
func safeCall(ctx context.Context, fn func(context.Context) error) (err error) {
defer func() {
if r := recover(); r != nil {
err = fmt.Errorf("panic: %v", r)
}
}()
return fn(ctx)
}
上面代码少了一个 import：sync。补上即可。这段代码的核心取舍是：让 g.Go 的回调永远返回 nil，把错误收集的责任从 errgroup 转移到自己的 slice 上。这样做换来两个好处：
- 不会因为第一个错误触发内部 cancel，导致后续任务拿到
context canceled的假失败。用errgroup.WithContext时这是高频困惑点，很多人看到一堆context canceled以为是自己的 bug。 Wait()的返回值不再是信息源，全部错误都在errs里，语义单一。
代价是必须加锁。如果任务量在十万级、且错误极少，锁竞争可以忽略；如果预期大量失败，可以换成 chan error 配容量 len(tasks)，或使用 atomic.Int64 计数成功数、只在错误路径加锁。
需要「快速失败」时改用 WithContext
并非所有场景都要跑完全部任务。比如上游已经超时、或者某个关键依赖挂了，继续打下去只是浪费配额。这时用 errgroup.WithContext，语义变成「任一任务失败即取消其余任务」，同时仍要收集已发生的错误。
package batch
import (
"context"
"errors"
"fmt"
"sync"
"golang.org/x/sync/errgroup"
)
// RunFailFast 在任一任务失败时取消其余任务（通过 ctx 传播），
// 但已经产生的错误仍然全部收集，不会被丢弃。
func RunFailFast(ctx context.Context, tasks []Task, limit int) (Result, error) {
if limit <= 0 {
limit = 1
}
// 这里必须用 WithContext：它的派生 ctx 在首个错误时被 cancel。
g, gctx := errgroup.WithContext(ctx)
g.SetLimit(limit)
var (
mu sync.Mutex
errs []error
)
for _, t := range tasks {
t := t
g.Go(func() error {
err := safeCall(gctx, t.Fn)
if err != nil {
mu.Lock()
errs = append(errs, fmt.Errorf("task[%d]: %w", t.Index, err))
mu.Unlock()
// 返回非 nil，触发 group 内部 cancel，后续 Go 调用会拿到 canceled ctx。
return err
}
return nil
})
}
waitErr := g.Wait()
// 注意：被取消的任务通常返回 context.Canceled，
// 这些是「连带失败」，是否计入 Failed 取决于业务口径。
joined := errors.Join(errs...)
return Result{
Total: len(tasks),
Failed: len(errs),
Err: joined,
}, waitErr
}
如果你的任务内部会把 context.Canceled 当普通错误返回，聚合结果里就会混进大量噪音。两种处理方式：一是在 safeCall 里过滤 errors.Is(err, context.Canceled) 且 gctx 已取消的情况；二是显式区分「真实失败」与「连带取消」，用两个 slice 分别收集。后者更适合要打点上报的线上服务。
为什么 SetLimit 之前要先判断任务量
上面代码里有一句 if limit > len(tasks) { limit = len(tasks) }，看起来多余，其实关系到 SetLimit 的实现方式。v0.7.0 的源码里，SetLimit 会按 n 大小创建带缓冲的 channel 作为信号量：
// 摘自 golang.org/x/sync/errgroup/errgroup.go（简化）
func (g *Group) SetLimit(n int) {
if n < 0 {
g.sem = nil
return
}
// 注意这里：按 n 分配
g.sem = make(chan token, n)
}
也就是说，SetLimit(100000) 会立刻分配一个十万容量的 channel，即使实际只有 10 个任务。更糟的是常见的写法：
g := new(errgroup.Group)
g.SetLimit(cfg.MaxConcurrency) // cfg 从配置读，可能配成 1000000
线上把并发配置写成六位数，直接分配 1e6 容量的 channel（每个 token 是 struct{}，channel 本身有额外的 per-element 开销，加上 hchan 结构），一次请求就是一次大块分配，QPS 一高内存立刻抖。压测时我们在 pprof 里看到 errgroup.(*Group).SetLimit 出现在 top alloc 里，就是这么来的。修正方式就是上面的钳制：limit = min(limit, len(tasks))。min 内置函数需要 Go 1.21+，老版本手写 if。
另一处细节：
SetLimit(-1)是合法的，语义为「不限制」，源码里sem置 nil，Go中的g.sem <- token{}就在 nil check 之外。但SetLimit(0)会创建一个容量为 0 的 channel，此时g.sem <- token{}永久阻塞——整个 group 一个任务都跑不起来，Wait()直接死锁。这是真实会踩的坑，尤其在「并发数由配置控制、默认值为 0」的代码里。所以兜底判断limit <= 0 { limit = 1 }不能省。
完整可运行示例与验证
下面是一个 main 包，直接把前面两个函数组装起来跑，模拟 20 个任务、并发 4、其中若干失败并有一个 panic。执行 go run . 即可看到输出，用来验证聚合行为是否符合预期。
package main
import (
"context"
"errors"
"fmt"
"sync"
"sync/atomic"
"time"
"golang.org/x/sync/errgroup"
)
type Task struct {
Index int
Fn func(ctx context.Context) error
}
type Result struct {
Total int
Succeeded int64
Failed int
Err error
Elapsed time.Duration
}
func safeCall(ctx context.Context, fn func(context.Context) error) (err error) {
defer func() {
if r := recover(); r != nil {
err = fmt.Errorf("panic: %v", r)
}
}()
return fn(ctx)
}
func Run(ctx context.Context, tasks []Task, limit int) Result {
start := time.Now()
if limit <= 0 {
limit = 1
}
if len(tasks) > 0 && limit > len(tasks) {
limit = len(tasks)
}
g := new(errgroup.Group)
g.SetLimit(limit)
var (
mu sync.Mutex
errs []error
ok atomic.Int64
)
for _, t := range tasks {
t := t
g.Go(func() error {
if err := safeCall(ctx, t.Fn); err != nil {
mu.Lock()
errs = append(errs, fmt.Errorf("task[%d]: %w", t.Index, err))
mu.Unlock()
return nil
}
ok.Add(1)
return nil
})
}
_ = g.Wait()
res := Result{Total: len(tasks), Succeeded: ok.Load(), Failed: len(errs), Elapsed: time.Since(start)}
if len(errs) > 0 {
res.Err = errors.Join(errs...)
}
return res
}
func main() {
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel()
var inflight, peak int32
tasks := make([]Task, 0, 20)
for i := 0; i < 20; i++ {
i := i
tasks = append(tasks, Task{Index: i, Fn: func(ctx context.Context) error {
cur := atomic.AddInt32(&inflight, 1)
defer atomic.AddInt32(&inflight, -1)
for {
p := atomic.LoadInt32(&peak)
if cur <= p || atomic.CompareAndSwapInt32(&peak, p, cur) {
break
}
}
time.Sleep(50 * time.Millisecond)
switch {
case i == 3:
return errors.New("downstream 503")
case i == 7:
panic("nil map write")
case i == 11:
return fmt.Errorf("db query: %w", context.DeadlineExceeded)
}
return nil
}})
}
res := Run(ctx, tasks, 4)
fmt.Printf("total=%d succeeded=%d failed=%d peak_concurrency=%d elapsed=%v\n",
res.Total, res.Succeeded, res.Failed, atomic.LoadInt32(&peak), res.Elapsed.Round(time.Millisecond))
if res.Err != nil {
fmt.Println("joined error:")
for i, e := range res.Err.(interface{ Unwrap() []error }).Unwrap() {
fmt.Printf(" #%d %v\n", i, e)
}
if errors.Is(res.Err, context.DeadlineExceeded) {
fmt.Println("errors.Is(context.DeadlineExceeded) = true")
}
}
}
预期输出（时间会浮动）：
total=20 succeeded=17 failed=3 peak_concurrency=4 elapsed=300ms
joined error:
#0 task[3]: downstream 503
#1 task[7]: panic: nil map write
#2 task[11]: db query: context deadline exceeded
errors.Is(context.DeadlineExceeded) = true
三点验证结论：peak_concurrency=4 证明 SetLimit 生效；failed=3 证明多错误全部保留且 panic 被兜住；errors.Is 穿透 errors.Join 与 fmt.Errorf("%w") 两层包装成功，说明聚合不会破坏错误链。注意 errors.Join 返回值的动态类型实现了 Unwrap() []error，想遍历子错误要用这个接口断言，Go 1.20 的 errors.Join 文档里叫 interface { Unwrap() []error }，不要写成 Unwrap() error。
常见报错与排查
errgroup: invalid limit
在 v0.3.0 之前没有 SetLimit，若 go.mod 里是 golang.org/x/sync v0.1.0，编译直接报：
./main.go:23:4: g.SetLimit undefined (type *errgroup.Group has no field or method SetLimit)解决：go get golang.org/x/sync@v0.7.0，并确认 go.sum 已更新。企业内网拉不到模块时，GOFLAGS=-mod=mod go mod tidy 配合私有 proxy。
fatal error: all goroutines are asleep – deadlock!
出现这个说明有 goroutine 永久阻塞。在 SetLimit 场景下，最常见原因是 SetLimit(0)：sem 容量为 0，第一次 g.Go 就卡在 g.sem <- token{}，主 goroutine 卡在循环里，永远到不了 Wait。加了 limit <= 0 { limit = 1 } 后消失。
第二种原因是回调内部又调用 g.Go（递归提交任务）。SetLimit(n) 下所有 n 个槽位都被父任务占着，子任务提交会阻塞，而父任务在等子任务完成，形成环。errgroup 本身不支持这种「任务内再提交」，需要自己不设 limit 或用独立的 worker pool。
结果里全是 context canceled
用了 errgroup.WithContext 后，第一个错误触发 cancel，其余正在跑的任务从 ctx 拿到取消信号，返回 context.Canceled，然后也被 errs 收集，最终 errors.Join 里一堆 canceled。这不是 bug，是语义。排查时先看 g.Wait() 的首个错误（那才是真因），或者在聚合时把 errors.Is(err, context.Canceled) 且父 ctx 未取消的错误单独标记为「连带」。如果业务要求「全部跑完再汇总」，就别用 WithContext，用本文第一版模板。
goroutine 数量不降反升
如果任务体内部自己又起 goroutine 做下游并发（比如每个任务里再 fan-out 10 个子请求），那么 SetLimit(4) 限制的只是外层 4 个，实际并发可能是 4 × 10。压测时用 runtime.NumGoroutine() 打点，或直接 go tool pprof http://host/debug/pprof/goroutine 看栈。正确做法是把并发预算按层分配：外层 4、内层每任务 2，总量 8，或者干脆把内层扇出改成串行。
小结
用 errgroup.SetLimit 做限流批量执行，模板要点可以收敛成五条：
SetLimit的入参必须钳制：limit <= 0兜底为 1，limit > len(tasks)收敛到任务数，避免零容量死锁与无谓大分配。- 要「跑完汇总」就用
new(errgroup.Group)，回调统一返回 nil，错误自己用sync.Mutex+ slice 收集；要「快速失败」才用WithContext。 - 所有任务入口套一层
recover，把 panic 转成 error 参与聚合，否则一个任务 panic 会带崩进程。 - 错误加
task[%d]前缀再用errors.Join聚合，保留errors.Is/errors.As的穿透能力；遍历子错误断言Unwrap() []error。 - 不要在同组内递归调用
g.Go，也不要把「限流」当成「下游配额保护」——任务内部若再扇出，需要另行限制。
FAQ
SetLimit 和固定大小的 worker pool 有什么区别？
行为上接近，都是「最多 n 个并发」。区别在于 SetLimit 把限流内嵌在 group 里，省掉任务 channel、结果 channel 和关闭逻辑；worker pool 则更适合需要复用、或者任务需要多次提交（一批接一批）的场景。单次批量执行、任务数已知，用 SetLimit 更少代码。
为什么不能在任务回调里返回错误，让 Wait() 拿全部错误？
因为 Wait() 的签名固定为 func() error，只能返回一个。errgroup 内部只保存第一个非 nil 错误。即使你返回多个，group 也只记第一个。这是 API 设计决定的，不是 bug。
errors.Join 有什么性能问题？
errors.Join 会把所有 error 放进一个 slice，Error() 被调用时用换行拼接字符串。如果一次聚合十万条错误，字符串拼接会很慢，内存也翻倍。线上建议在聚合前截断（比如只保留前 100 条 + 总数），或者改用带计数的自定义错误类型，只在需要时格式化前 N 条。
Go 1.22 之前为什么要在循环里写 t := t？
Go 1.22 之前，for _, t := range tasks 的循环变量在所有迭代间共享，闭包捕获的是同一个变量地址，不加 t := t 会导致多个 goroutine 读到同一份值（通常是最后一个）。Go 1.22 起每次迭代创建新变量，不再需要。项目若同时兼容多个版本，保留该行更安全，go vet 也不会报。
任务内部要访问数据库连接池，并发上限该设多少？
取决于下游能力而非机器核数。原则：并发上限 × 单任务连接持有时间 ≤ 连接池 MaxOpenConns 的一半，给其他业务留余量。例如 pool 设为 50，单任务平均持连接 20ms、总耗时目标 200ms，则并发 10~15 比较稳妥。压测时把 sql.DBStats 的 WaitCount 和 WaitDuration 打出来，非零就说明并发超过了池容量，正在排队。