Go · context 取消传播

💡 一句话理解

context 是每个请求的生命线: 从入口创建, 沿调用树一层层往下传, 谁派生谁负责 cancel。超时是预算, 子节点只会继承到更紧的 deadline; 取消是广播, 父节点一声 close, 全树所有监听 Done() 的 goroutine 同时亮灯退出。它管的是"什么时候不干了" —— 用户断开、上游超时、优雅停机, 都靠这一根线把散落各处的等待点一起叫醒。它不该管业务参数: value 里只放请求元数据, 越界的用法都是事故的开始。

🧠 必知必会 必考 & 必会

Context 接口

四个方法: Deadline() 返回截止时刻、Done() 返回一个 channel、Err() 返回取消原因、Value(key) 沿链取值。前三个管生死, 第四个管元数据。

type Context interface {
    Deadline() (time.Time, bool)
    Done() <-chan struct{} // 关闭 = 广播取消
    Err() error            // → Canceled / DeadlineExceeded
    Value(key any) any
}

Background / TODO

context.Background() 是根, 永不取消; context.TODO() 是"还没想好传谁"的占位 —— 出现在业务代码里通常说明调用链设计欠账。

ctx := context.Background()
fmt.Println(ctx.Err()) // → <nil> (根永不取消)
// TODO() 只是占位: 业务代码里出现 = 调用链设计欠账

WithCancel

手动停止: 返回 (ctx, cancel)。cancel 是幂等的, 多次调用安全; 不调也不崩, 但计时器与父节点引用会等到父取消才释放。

ctx, cancel := context.WithCancel(parent)
cancel() // 关键: cancel 幂等, 多次调用安全
cancel()
fmt.Println(ctx.Err()) // → context canceled

WithTimeout / WithDeadline

相对时长与绝对时刻, 到点自动 cancel, 内部就是一个 timer。绝大多数业务用 WithTimeout; 跨时区的定时任务用 WithDeadline。

ctx, cancel := context.WithTimeout(parent, 3*time.Second)
defer cancel() // 相对时长: 到点自动 cancel
// 绝对时刻版: WithDeadline(parent, time.Now().Add(3*time.Second))

defer cancel()

派生之后立刻 defer: 即使函数提前返回, 计时器和父节点里的子表项也被即时清掉 —— 这是官方文档用粗体写的规矩。

ctx, cancel := context.WithTimeout(parent, time.Second)
defer cancel() // 关键: 派生的下一行就 defer
// 不 defer: 计时器+父节点子表项拖到父取消才释放

树形广播原理

cancel 的本质是 close(done chan): channel 关闭对所有接收方同时可见, 所以一次 cancel 能瞬间唤醒整棵子树 —— 广播而不是逐个通知。

c1, cancel1 := context.WithCancel(ctx)
c2, _ := context.WithCancel(c1) // c2 挂在 c1 子树
cancel1() // 关键: ≈ close(done), 一次唤醒全子树
<-c2.Done() // → 立即可读, 广播非逐个通知

第一参数惯例

func(ctx context.Context, ...) 永远排第一且不放进 struct —— lint 会盯; 例外是长生命周期对象在 Start/Stop 成对持有。

func Load(ctx context.Context, id int64) {} // ctx 永远第一
// 反例: type S struct{ ctx context.Context } — lint 必盯
// 例外: Start/Stop 成对的长生命周期对象才存 struct

r.Context()

服务端集成: net/http 为每个请求自动创建, 客户端断开时自动取消 —— 这条白送的生命线, 不接等于放弃快速失败。

func h(w http.ResponseWriter, r *http.Request) {
    ctx := r.Context() // 白送的生命线
    <-ctx.Done() // → 客户端一断开立即醒来
}

NewRequestWithContext

客户端集成: 把 ctx 绑进请求, 超时/取消时传输被立即中断、连接归还池 —— fd 不再缓慢泄漏。

req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
resp, err := http.DefaultClient.Do(req) // 超时/取消在此生效
defer resp.Body.Close()
// → DeadlineExceeded 一到, 传输中断连接归还池

穿透驱动层

db.QueryContext 会一路传到驱动甚至数据库服务端: 查询真被 cancel; 只在应用层 select 挡超时是"假超时"。

rows, err := db.QueryContext(ctx, sql, args...)
defer rows.Close()
// 关键: 取消一路送到驱动/DB 服务端, 查询真被中断
// 只在应用层 select 挡 = 假超时, 连接池照样被拖垮

ctx.Err()

Done 亮灯后取原因: context.Canceled(被主动取消) / context.DeadlineExceeded(超时)。判定一律 errors.Is, 它们是哨兵。

<-ctx.Done()
fmt.Println(ctx.Err())
// → context.DeadlineExceeded (超时) / context.Canceled (取消)
errors.Is(err, context.Canceled) // 判定一律 Is, 是哨兵

WithValue 纪律

只放请求级元数据(reqID/trace/租户), key 用私有类型防撞名; 查找是沿链 O(n), 链越深越慢 —— 它不是参数传递通道。

type ctxKey int // 私有类型: 别包撞不了这个 key
ctx = context.WithValue(ctx, ctxKey(0), "req-42")
id, _ := ctx.Value(ctxKey(0)).(string)
fmt.Println(id) // → req-42; 只放请求元数据, 非参数通道

Shutdown(ctx)

优雅停机: 停接新连接 + 等存量请求排干, ctx 给出强断窗口(如 15s); 注意必须用 Background 新根, 详见场景 5。

ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
defer cancel()
if err := srv.Shutdown(ctx); err != nil {
    log.Print("forced close") // 15s 没排干才强断
}
// 关键: 必须用 Background 新根, 不能复用已取消的

🏭 生产实战 real world

func (h *Handler) GetFeed(w http.ResponseWriter, r *http.Request) {
    ctx, cancel := context.WithTimeout(r.Context(), 3*time.Second)
    defer cancel() // r.Context() 做父: 客户端断开也一起停
    feed, err := h.svc.Feed(ctx, r.URL.Query().Get("uid"))
    switch {
    case errors.Is(err, context.DeadlineExceeded):
        http.Error(w, "upstream timeout", http.StatusGatewayTimeout)
    case errors.Is(err, context.Canceled):
        return // 用户已断开, 写响应无意义, 静默结束
    case err != nil:
        slog.Error("feed", "err", err) // 只有边界层打日志
        http.Error(w, "internal", 500)
    default:
        json.NewEncoder(w).Encode(feed)
    }
}

func (r *UserRepo) ByID(ctx context.Context, id int64) (*User, error) {
    ctx, cancel := context.WithTimeout(ctx, 800*time.Millisecond)
    defer cancel() // 不 cancel: 计时器与父引用一直被持有

    rows, err := r.db.QueryContext(ctx,
        `SELECT id, name FROM users WHERE id = $1`, id)
    if err != nil {
        return nil, fmt.Errorf("query user %d: %w", id, err)
    }
    defer rows.Close() // 取消后 Rows 自动关闭, 这里双保险

    u := &User{}
    for rows.Next() {
        err = rows.Scan(&u.ID, &u.Name)
    }
    return u, rows.Err()
}

func callDownstream(ctx context.Context, url string) ([]byte, error) {
    req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
    if err != nil {
        return nil, fmt.Errorf("build req: %w", err)
    }
    resp, err := http.DefaultClient.Do(req)
    if err != nil {
        return nil, fmt.Errorf("do %s: %w", url, err) // ctx 超时在此生效
    }
    defer resp.Body.Close()
    return io.ReadAll(io.LimitReader(resp.Body, 4<<20)) // 顺手防大响应
}
// ctx 版: DeadlineExceeded 一到, 传输立即中断, 连接归还池;
// 别用全局 http.Client{Timeout} 一刀切 — 它不区分调用方预算

func aggregate(ctx context.Context, deps []string) (*Page, error) {
    g, ctx := errgroup.WithContext(ctx) // 同一父: 一处失败全体取消
    parts := make([]Part, len(deps))
    for i, d := range deps {
        g.Go(func() error { // 首个 return err 触发 ctx 取消
            p, err := fetchPart(ctx, d) // 这里的 ctx 已是"首错即取消"版
            if err != nil {
                return fmt.Errorf("part %s: %w", d, err)
            }
            parts[i] = p // 各写各下标, 无竞争
            return nil
        })
    }
    if err := g.Wait(); err != nil {
        return nil, err // 首错返回时, 其余分支已被 ctx 叫停
    }
    return compose(parts), nil
}

srv := &http.Server{Addr: ":8080", Handler: h}
go func() {
    if err := srv.ListenAndServe(); err != http.ErrServerClosed {
        log.Fatalf("listen: %v", err)
    }
}()

quit := make(chan os.Signal, 1)
signal.Notify(quit, syscall.SIGTERM)
<-quit // 等 SIGTERM, 不杀存量连接

ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
defer cancel()
// 必须 Background 新根: 复用已取消的根 ctx, Shutdown 立即强断
if err := srv.Shutdown(ctx); err != nil {
    log.Printf("forced close: %v", err) // 15s 没退完的请求强制断
}

// 后台订阅: 手动停止信号(与请求无关, 用 Background 做根)
func (s *Subscriber) Start() {
    s.ctx, s.cancel = context.WithCancel(context.Background())
    go func() {
        for {
            msg, err := s.pull(s.ctx) // 阻塞读也要吃 ctx
            if s.ctx.Err() != nil {
                s.cleanup() // 停机收尾: ack/断连/刷缓冲
                return
            }
            if err != nil {
                backoff(s.ctx, err) // 退避也感知取消, 不假死
                continue
            }
            s.handle(msg)
        }
    }()
}

func (s *Subscriber) Stop() { s.cancel() } // 广播一发, 干净退出

type ctxKey int // 私有类型: 任何包都不可能撞这个 key

const requestIDKey ctxKey = 0

func WithRequestID(ctx context.Context, id string) context.Context {
    return context.WithValue(ctx, requestIDKey, id) // 只放元数据
}

func Log(ctx context.Context, msg string, kv ...any) {
    id, _ := ctx.Value(requestIDKey).(string) // 断言失败得零值
    slog.Info(msg, append(kv, "req_id", id)...)
}
// 中间件注入一次, 任何深度的日志都自带 req_id;
// 反例: reqID 当参数层层手工传 — 签名膨胀且必有人漏传

const (
    totalBudget   = 3 * time.Second    // 入口总预算 = SLA
    dbBudget      = 800 * time.Millisecond // 依赖子预算, 加起来 < 总
    downstreamBud = 1500 * time.Millisecond
)

func serve(ctx context.Context) error {
    ctx, cancel := context.WithTimeout(ctx, totalBudget)
    defer cancel() // 总闸: 内部怎么派生, 3s 必须收口

    dbCtx, dbCancel := context.WithTimeout(ctx, dbBudget)
    defer dbCancel() // 每个子预算独立 cancel, 用完即弃
    if err := loadDB(dbCtx); err != nil {
        return err
    }
    return fetchDownstream(ctx) // 用总预算剩余, 天然不超总盘
}

func handle(msg broker.Message) error {
    // 新根 Background: 处理时长与消费连接的 ctx 解耦
    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()

    if err := process(ctx, msg); err != nil {
        if errors.Is(err, context.DeadlineExceeded) {
            return broker.NackRequeue // 30s 没处理完: 重投再试
        }
        return broker.NackDiscard // 业务错: 重投也没用, 进死信
    }
    return broker.Ack
}
// 错误示范: ctx 复用消费连接的 — 连接一抖动等于消息失败,
// 配合重投策略 = 无限重投风暴

// 排查: curl :6060/debug/pprof/goroutine?debug=1 | head -50
// 大量 stuck 在 chan receive, 创建栈指向 pollLoop

for { // 旧代码: 永不退出, 每个 G 持 ~2KB 栈 + 闭包引用
    msg := s.ch.Recv() // 不感知任何取消信号
    s.dispatch(msg)
}

for { // 修复: 每个长循环都要有 ctx 出口
    select {
    case msg := <-s.ch:
        s.dispatch(msg)
    case <-s.ctx.Done(): // 停机/重连时整批退出
        return
    }
}
// 修复后 goroutine 稳定在 ~200, 泄漏增长率归零

⚠️ 编码注意与常见坑 pitfalls

// 错: ctx, cancel := context.WithCancel(p) 之后没 defer — G 数缓涨
ctx, cancel := context.WithTimeout(p, time.Second) // 对: 下一行就 defer
defer cancel() // go vet 的 lostcancel 能抓漏网

// 错: type S struct{ ctx context.Context } — 依赖与生命周期全藏住
func (s *S) Load(ctx context.Context, id int64) {}
// 对: ctx 一路当第一参数传, Start/Stop 对象是唯一例外

// 错: ctx = WithValue(ctx, amountKey, 100) — 金额藏进 value
func Pay(ctx context.Context, amount int64) error {}
// 对: 业务输入进签名, value 只放 reqID/trace 请求属性

// 错: select 里加 default: — 永不阻塞, CPU 打满且不感知取消
select {
case msg := <-ch: // 对: 去掉 default, 业务与取消同时等
    dispatch(msg)
case <-ctx.Done():
    return ctx.Err()
}

// 错: WithTimeout(context.Background(), 3s) — 断开链断, 请求傻跑
ctx, cancel := context.WithTimeout(r.Context(), 3*time.Second)
defer cancel() // 对: 全链统一从 r.Context() 派生

// 错: 只用 select+time.After 挡应用层 — SQL 在 DB 跑满 30s
rows, err := db.QueryContext(ctx, q, args...) // 对: 穿透驱动
req, _ := http.NewRequestWithContext(ctx, http.MethodGet, u, nil)
// → 取消送进传输层, 查询真被中断, 连接立刻归还

case <-ctx.Done():
    flush(hugeBuf) // 错: 取消分支还干重活 = 没取消
    return ctx.Err() // 对: 立即返回, 清理放轻量 defer

// 错: 每层 WithTimeout(ctx, 3s) — 子 3s 从未兑现, 父先收走
dbCtx, cancel := context.WithTimeout(ctx, 800*time.Millisecond)
defer cancel() // 对: 子 deadline=min(自身,父剩余), 表驱动分配

_, err := w.Write(buf)
log.Error("write", err) // 错: 用户关页面刷爆 error 告警
if errors.Is(err, context.Canceled) {
    return // 对: 降 info 或静默
}

ctx = context.WithValue(ctx, "id", x) // 错: 裸 string 必撞名
type ctxKey int // 对: 私有类型 key, 编译器保证唯一
ctx = context.WithValue(ctx, ctxKey(1), x)

// 错: WithTimeout(p, 0) — 非正=已过期, 首个查询即超时
ctx, cancel := context.WithTimeout(p, time.Second)
defer cancel()
// 对: 想表达"不设限"就别派生, 直接用父

// 错: var globalCancel context.CancelFunc 存全局跨请求复用
//     → 上个请求收尾一调, 后到请求直接夭折
// 对: cancel 与 ctx 同请求作用域, 用完即弃, 绝不放全局

for { work(<-ch) } // 错: 无出口, G 只增不减
for {
    select {
    case m := <-ch: work(m)
    case <-ctx.Done(): // 对: 每个阻塞点都能被叫醒
        return
    }
}

slog.Info("ctx", "ctx", ctx) // 错: 打出整条 value 链
slog.Info("ctx", "err", ctx.Err(),
    "dl", ctx.Deadline()) // 对: 只打摘要

// 错: 一律 5xx 告警 — 半夜被"用户关页面"叫醒
if errors.Is(err, context.Canceled) {
    return 499 // 对: 取消→499/静默, 真错→5xx
}

// 错: for { select { case <-time.After(time.Second): … } }
//     1.23 前每轮新 timer, 高频循环内存堆积
t := time.NewTimer(time.Second) // 对: 循环外建 + Reset
defer t.Stop()

// 错: 嵌套十几层 WithValue — Value() 沿链 O(n) 越查越慢
type meta struct { reqID, traceID string }
ctx = context.WithValue(ctx, metaKey, m) // 对: 单 key 收拢

// 错: ctx := context.Background() — 下游 hang 则 CI 卡死
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel() // 对: 必带兜底超时; 1.24+ 直接 t.Context()

// 错: srv.Shutdown(reqCtx) — 已取消的根 → 立即强断 502
ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
defer cancel()
return srv.Shutdown(ctx) // 对: 新根 + 停机窗口

f, _ := os.CreateTemp("", "job-") // 取消后不会自己消失
defer f.Close()
tx, _ := db.BeginTx(ctx, nil)
defer tx.Rollback() // 对: Done 分支 + defer 双保险, 事务显式回滚