Go 结构化并发怎样收敛后台任务：errgroup、退出信号与错误回收

订单导入接口经常要同时拉取商品、库存和价格三份数据。最初的写法是启动三个 goroutine，再用一个 channel 收错误；一旦库存请求先失败，另外两个任务还在后台跑，接口虽然已经返回，日志里却继续出现超时。这个问题的关键不是“能不能并发”，而是让一组任务共享同一个生命周期：任意一个任务失败，其他任务收到退出信号，最后由调用方统一拿到第一个错误。

先把“并发完成”改成“并发收敛”

结构化并发可以先用一句工程规则理解：父任务创建的子任务，必须在父任务返回前结束，并且共享父任务的取消边界。这样一来，调用关系和资源关系是一致的，排查时不需要猜某个 goroutine 是否还活着。

下面的示例模拟订单导入：三个任务并行读取不同数据，库存任务故意返回错误。真实项目中，loadProducts、loadStock 和 loadPrices 可以替换成 HTTP、数据库或消息系统调用。

package main

import (
    "context"
    "errors"
    "fmt"
    "golang.org/x/sync/errgroup"
    "time"
)

func importOrder(ctx context.Context) error {
    group, taskCtx := errgroup.WithContext(ctx)

    group.Go(func() error { return loadProducts(taskCtx) })
    group.Go(func() error { return loadStock(taskCtx) })
    group.Go(func() error { return loadPrices(taskCtx) })

    return group.Wait()
}

func loadProducts(ctx context.Context) error { return waitOrStop(ctx, 80*time.Millisecond) }

func loadStock(ctx context.Context) error {
    if err := waitOrStop(ctx, 30*time.Millisecond); err != nil { return err }
    return errors.New("stock service: version conflict")
}

func loadPrices(ctx context.Context) error { return waitOrStop(ctx, 200*time.Millisecond) }

func waitOrStop(ctx context.Context, d time.Duration) error {
    timer := time.NewTimer(d)
    defer timer.Stop()
    select {
    case 

这里没有单独创建错误 channel。errgroup 会记录第一个非 nil 错误，并在该错误出现后取消 taskCtx。价格任务如果还在等待计时器，就会从 ctx.Done() 分支返回；Wait() 则等它完成后才把错误交还给调用方。

为什么只调用 cancel 还不够

手写并发代码时，常见做法是创建一个 cancel，某个任务出错后调用它。这个动作只能发出信号，不能替调用方等待其他 goroutine 退出。如果函数随后直接 return，后台任务可能仍在占用连接、读响应体或写共享缓存。

errgroup.WithContext 的价值在于把两个动作绑在一起：错误会触发取消，Wait() 会等待所有已启动的函数。注意，任务内部仍要主动配合取消；一个不支持 context 的第三方 SDK，即使外层传入了 taskCtx，也可能继续阻塞。

让阻塞任务真的响应退出信号

取消检查要放在最深的阻塞边界附近，而不是只写在外层调度函数。HTTP 调用应该把 context 绑定到请求；批量处理则要在每批开始前和每次写入前检查一次。这样既不会因为频繁检查拖慢正常路径，也能避免取消后继续处理大批数据。

func fetch(ctx context.Context, client *http.Client, url string) ([]byte, error) {
    req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
    if err != nil { return nil, err }

    resp, err := client.Do(req)
    if err != nil { return nil, err }
    defer resp.Body.Close()

    if resp.StatusCode != http.StatusOK {
        return nil, fmt.Errorf("upstream status: %s", resp.Status)
    }
    return io.ReadAll(resp.Body)
}

如果上游客户端只提供不可取消的调用，可以把它放在单独的适配层，并明确记录“取消只对等待结果生效”。不要把这种不完整的语义伪装成完全可取消，否则出现连接泄漏时很难定位。

用最小验收证明任务已经结束

测试不应只断言返回了 stock service: version conflict。还要记录取消分支是否被其他任务走到，并观察测试前后的 goroutine 数量。对 HTTP 或数据库任务，可以在 fake client 中记录 context 是否已取消。

func TestImportOrderStopsSiblings(t *testing.T) {
    before := runtime.NumGoroutine()
    err := importOrder(context.Background())
    if err == nil || !strings.Contains(err.Error(), "version conflict") {
        t.Fatalf("unexpected error: %v", err)
    }

    time.Sleep(20 * time.Millisecond) // 仅用于测试观察窗口，不属于业务实现
    after := runtime.NumGoroutine()
    if after-before > 1 {
        t.Fatalf("possible leaked goroutine: before=%d after=%d", before, after)
    }
}

生产环境不要依赖固定等待时间判断收敛。更可靠的方式是在 fake 任务里用 channel 发出“收到取消”的事件，在测试中等待这个事件并设置测试级超时；监控侧则关注请求结束后的 goroutine、连接和队列指标是否持续增长。

几个容易把生命周期写断的细节

• 不要在 group.Go 之前把耗时工作做完，否则它根本不受任务组管理。
• 不要只返回最后一个错误；多个任务同时失败时，最后一个错误通常丢失了最早的根因。
• 不要复用已取消的 context 处理下一批独立请求；每个请求都应从新的父 context 派生。
• 不要把业务成功写入放在 Wait() 之前，避免部分结果已经落库而整体任务最终失败。

相关问题

errgroup 会限制并发数量吗？

不会。它负责错误传播和等待；需要限制并发时，再使用 SetLimit 或信号量，并保留同一个派生 context。

任务返回 context.Canceled 算不算真正错误？

通常它是兄弟任务失败后的结果，不应覆盖首个业务错误。最终错误以 Wait() 返回值为准，再根据日志区分主动取消和上游失败。

没有 errgroup 时能手写吗？

可以，但必须同时维护任务计数、首错保护、取消函数和等待逻辑。项目已经依赖 golang.org/x/sync 时，使用 errgroup 通常更容易让这些边界被复查。

小结：把退出路径当成主流程

后台并发的完成标准不是“几个 goroutine 都启动了”，而是父函数返回时，子任务、连接和错误都已经有明确归宿。用 errgroup.WithContext 建立父子生命周期，把 ctx.Done() 放到真实阻塞点，再用 Wait() 和可观测指标验收，任务失败时才能快速停在可控边界内。