Go errgroup 怎样收集第一个错误并取消其余任务

在一组并发任务里，调用方通常只需要两个结果：全部任务成功，或者尽快知道哪一个任务先失败，并让其余任务停止继续消耗资源。Go 的 errgroup 正好把任务计数、错误传播和 context 取消组合在一起。

官方文档：https://pkg.go.dev/golang.org/x/sync/errgroup

最常用的写法是 errgroup.WithContext：每个任务返回一个 error，第一个返回非空错误的任务会触发派生上下文取消；调用方再用 g.Wait() 等待已启动的任务结束，并拿到第一个非空错误。这里的“取消”是协作式的，正在执行的函数必须检查 ctx.Done() 或把上下文继续传给下游 API。

第 1 步：先分清 Go errgroup 的三个职责

可以把一次并发流程看成一条小型工作流：g.Go 负责启动阶段，任务函数负责执行与响应取消，g.Wait 负责收口。WithContext 额外提供一份与这组任务绑定的派生上下文。

• g.Go(func() error) 启动一个任务。任务返回 nil 表示成功，返回非空错误表示失败。
• errgroup.WithContext(ctx) 返回 Group 和派生 Context。第一个非空错误出现时，派生上下文会被取消。
• g.Wait() 会等待所有已经加入 Group 的函数返回，然后返回第一个非空错误；没有错误时返回 nil。

因此，不要把 Wait 理解成“谁先报错就立刻强制杀掉其他 goroutine”。它会等待任务函数真正返回，取消只负责发出停止信号。若任务一直阻塞在不可取消的调用里，Wait 仍会等它结束。

第 2 步：让任务真正响应取消

一个常见误区是只调用 WithContext，却不使用返回的 ctx。这样虽然错误会被 Wait 收集，但其他任务可能继续工作。可取消点至少应放在长循环、等待外部 I/O 和向结果通道发送数据的位置。

下面的示例把“任务失败”“任务取消”“调用方统一处理”放在同一条流程中。示例中的 processOne 用定时器模拟外部工作，真实项目里可以把 ctx 传给 HTTP、数据库或 RPC 客户端。

package main

import (
    "context"
    "errors"
    "fmt"
    "time"

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

type result struct {
    job   string
    value string
}

func processOne(ctx context.Context, job string) (string, error) {
    timer := time.NewTimer(120 * time.Millisecond)
    defer timer.Stop() // 任务提前取消时也要释放定时器资源

    select {
    case <-timer.C:
        if job == "报表-错误样本" {
            return "", errors.New("下游返回业务错误") // 让一个任务模拟真实失败
        }
        return "已处理:" + job, nil
    case <-ctx.Done():
        return "", ctx.Err() // 取消是协作退出，不要继续发送成功结果
    }
}

func runJobs(parent context.Context, jobs []string) ([]result, error) {
    g, ctx := errgroup.WithContext(parent)
    g.SetLimit(3) // 限制同时运行的任务数，数值应按下游容量配置
    results := make(chan result, len(jobs))

    for _, job := range jobs {
        job := job // 为闭包保存本轮循环的任务名
        g.Go(func() error {
            value, err := processOne(ctx, job)
            if err != nil {
                return err // 第一个非空错误会触发 ctx 取消
            }
            select {
            case results <- result{job: job, value: value}:
                return nil
            case <-ctx.Done():
                return ctx.Err() // 发送结果时也要避免阻塞在已取消流程上
            }
        })
    }

    err := g.Wait() // 等待所有已启动任务返回，再收集统一错误
    close(results)  // 此时所有发送者都退出，关闭通道才安全
    if err != nil {
        return nil, err
    }

    collected := make([]result, 0, len(jobs))
    for item := range results {
        collected = append(collected, item)
    }
    return collected, nil
}

func main() {
    jobs := []string{"用户-画像", "订单-汇总", "报表-错误样本", "库存-快照"}
    values, err := runJobs(context.Background(), jobs)
    if err != nil {
        fmt.Println("批处理失败:", err) // 只在入口处理最终错误
        return
    }
    fmt.Println("成功结果:", values)
}

这里的错误顺序取决于哪个任务先返回非空错误，不应依赖任务切片的排列顺序。被取消的任务可能返回 context.Canceled，但调用方只需要处理 g.Wait() 返回的第一个错误，不必把每个取消错误再重复上报。

第 3 步：把“第一个错误”理解成收口规则

errgroup 不会收集一份完整错误列表，它的默认语义是“任一任务失败，则这组任务失败”，并保留第一个非空错误。这个规则适合批量读取、并发请求、流水线阶段等场景：一旦整体结果已经不可能成功，继续启动或等待无关工作通常没有价值。

入口处应把 Wait 当成唯一的失败出口：

values, err := runJobs(ctx, jobs)
if err != nil {
    // 记录第一个真实失败原因；不要把 context.Canceled 当成新的业务根因
    return fmt.Errorf("并发批处理失败: %w", err)
}
// 只有 Wait 返回 nil，才把 values 交给下一阶段
publish(values)

如果业务确实需要“尽可能完成所有任务并报告多条错误”，就不应直接套用“第一个错误取消全部”的策略。可以让每个任务把结构化错误写入受控收集器，或使用专门的多错误聚合设计；那是另一种失败政策。

第 4 步：处理结果写入与退出顺序

并发代码最容易出问题的地方不是 g.Go，而是结果通道的生命周期。关闭通道前必须确认所有发送者已经退出，因此上例先执行 g.Wait()，再 close(results)。如果在任务仍可能发送时提前关闭，就会触发 panic。

上例使用容量为 len(jobs) 的缓冲通道，保证每个任务最多发送一次时，调用方可以先等待再统一消费。若每个任务可能产生大量结果，不能照搬这个容量，否则生产者可能在发送阶段阻塞，Wait 又在等待生产者结束，形成僵局。

大量结果更适合“独立消费者先启动”的形态：消费者持续读取通道，另一个协调 goroutine 在 g.Wait() 返回后关闭通道，主流程再等待消费者完成。无论采用哪种结构，都要同时保证三件事：取消后停止发送、发送者退出后再关闭、最终仍然调用 Wait 取得错误。

第 5 步：用 SetLimit 控制工作流压力

默认的零值 Group 不限制活动 goroutine 数量。任务来自用户输入或大批量数据时，可以先调用 g.SetLimit(n)，把活动任务数控制在下游可以承受的范围。达到上限后，后续 g.Go 会等待空闲名额。

g, ctx := errgroup.WithContext(parent)
g.SetLimit(8) // 只允许 8 个任务同时运行，按数据库或 API 配额调整

for _, item := range items {
    item := item
    g.Go(func() error {
        // 下游调用同样接收 ctx，失败或取消时才能及时返回
        return handle(ctx, item)
    })
}

if err := g.Wait(); err != nil {
    return err // Wait 仍然是整组任务的统一错误出口
}

SetLimit 不是重试器，也不是超时器；它只限制活动任务数。不要在仍有活动 goroutine 时修改限制值，也不要把一个 Group 复用于互不相关的多轮工作。

常见问题

errgroup 会强制终止其他 goroutine 吗？

不会。它取消派生上下文，其他任务需要主动检查取消信号或调用支持上下文的 API。

为什么已经有一个错误，Wait 还要等一会儿？

因为 Wait 的职责是等待已加入 Group 的函数返回。某个任务忽略取消、卡在不可取消的系统调用，都会延长收口时间。

被取消的任务返回 context.Canceled 会不会覆盖业务错误？

不会按“最后返回”覆盖。errgroup 返回第一个非空错误；取消产生的错误通常只是其他任务对停止信号的响应。

什么时候不适合一出错就取消全部任务？

当任务彼此独立、业务要求尽量完成，或需要收集全部失败原因时，不应使用默认的“一错全停”政策，应该设计独立的错误聚合与完成规则。

小结

Go errgroup 的关键不是替代 goroutine，而是把一组任务的失败政策固定下来：WithContext 连接取消信号，任务函数负责协作退出，Wait 等待收口并返回第一个非空错误。再配合发送点的取消检查、正确的通道关闭顺序和必要的 SetLimit，就能把并发流程从“启动很多 goroutine”变成可控的自动化工作流。