Go 并发模式：context 传播与链路追踪深度实践

1. context 接口设计与核心方法

context 的设计遵循了 Go 简洁有效的哲学，核心接口只有四个方法。

// context 接口定义
type Context interface {
    Deadline() (deadline time.Time, ok bool)  // 截止时间
    Done() <-chan struct{}                      // 取消信号通道
    Err() error                                 // 取消原因
    Value(key any) any                           // 获取值
}

1.1 四种基础 context

// 根 context：不可取消，没有截止时间，没有值
ctx := context.Background()

// 空的 context：通常用于测试
ctx := context.TODO()

// 可取消的 context
ctx, cancel := context.WithCancel(context.Background())
defer cancel()

// 带截止时间的 context
ctx, cancel := context.WithDeadline(parent, time.Now().Add(5*time.Second))
defer cancel()

// 带超时时间的 context（基于截止时间）
ctx, cancel := context.WithTimeout(parent, 3*time.Second)
defer cancel()

1.2 WithValue 的键值传播

// WithValue 创建链式 context
ctx := context.WithValue(parentCtx, key, value)

// 获取值（沿链路向上查找）
v := ctx.Value(key)

// 常见模式：请求 ID、用户信息、trace ID
type traceKey struct{}
type userKey struct{}

ctx = context.WithValue(ctx, traceKey{}, "trace-123")
ctx = context.WithValue(ctx, userKey{}, &User{ID: "user-456"})

// 获取时使用相同类型
traceID := ctx.Value(traceKey{}).(string)

2. 生命周期管理与取消传播

context 的取消是 Go 服务优雅关机的核心机制。正确理解取消的传播语义，可以避免大量并发相关的 bug。

2.1 取消的级联效应

// 当父 context 取消时，所有派生的子 context 都会收到信号
func parentOperation(ctx context.Context) error {
    ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
    defer cancel() // 父操作完成时清理

    // 启动子操作，它们共享同一个 ctx
    err := runSubOperations(ctx)

    return err
}

func runSubOperations(ctx context.Context) error {
    // 这个 ctx 继承了父的截止时间
    // 当父的截止时间到期或被取消，这里也会感知到
    return streamData(ctx)
}

2.2 Done 通道的监听模式

// 模式1：select + case
select {
case <-ctx.Done():
    return ctx.Err()
case result := <-ch:
    return process(result)
}

// 模式2：循环中定期检查
for {
    select {
    case <-ctx.Done():
        return ctx.Err()
    default:
    }
    // 执行一小部分工作
    doChunk()
}

// 模式3：用于长阻塞操作
func blockedCall(ctx context.Context) error {
    err := someBlockingOperation()
    // 如果 ctx 已取消，应该停止或快速返回
    if ctx.Err() != nil {
        return ctx.Err()
    }
    return err
}

2.3 Err 的语义

3. 链路传播：从 HTTP 到 gRPC

在微服务架构中，context 需要在多个层面正确传播，以确保取消信号和 trace 信息能够贯穿整个请求链路。

3.1 HTTP 服务中的 context

// HTTP 请求自带的 context
func httpHandler(w http.ResponseWriter, r *http.Request) {
    ctx := r.Context() // 请求的 context

    // 传播到下游调用
    err := callDownstream(ctx, "service-a")

    // 监听取消
    select {
    case <-ctx.Done():
        http.Error(w, "timeout", http.StatusGatewayTimeout)
        return
    default:
    }
}

// 中间件中的 context 传播
func tracingMiddleware(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        ctx := r.Context()

        // 添加 trace 信息到 context
        ctx = context.WithValue(ctx, traceKey{}, getTraceID(r))

        // 创建带 trace 的新请求（如果需要转发）
        req = req.WithContext(ctx)

        next.ServeHTTP(w, r)
    })
}

3.2 gRPC 中的 context 传播

// gRPC 客户端
func grpcClientCall(ctx context.Context, conn *grpc.ClientConn) error {
    client := pb.NewServiceClient(conn)

    // ctx 自动在 gRPC 元数据中传播
    // 包括 timeout 和 cancellation

    resp, err := client.SomeMethod(ctx, &pb.Request{
        // ...
    })

    return err
}

// gRPC 服务端
func (s *server) SomeMethod(ctx context.Context, req *pb.Request) (*pb.Response, error) {
    // 从 gRPC 元数据恢复 context
    // 包含 timeout 信息

    // 可以继续传播到其他 goroutine 或下游调用
    return s.backend.Call(ctx, req)
}

3.3 OpenTelemetry Trace 传播

// 注入 trace context 到 HTTP/gRPC metadata
func injectTraceContext(ctx context.Context, carrier *metadata.MD) {
    span := trace.SpanFromContext(ctx)
    sc := span.SpanContext()

    carrier.Set("traceparent", fmt.Sprintf(
        "00-%x-%x-%d",
        sc.TraceID(), sc.SpanID(), sc.TraceFlags(),
    ))
}

// 提取 trace context
func extractTraceContext(ctx context.Context, carrier *metadata.MD) context.Context {
    v := carrier.Get("traceparent")
    if len(v) == 0 {
        return ctx
    }

    // 解析并创建 linked span
    // ...
    return ctx
}

4. 反向传播 context value 的技巧

大多数情况下，context 是自上而下传播的。但在某些场景下，需要"反向"将值传回调用方，例如进度报告、错误聚合等。

4.1 使用 channel 回传数据

// 场景：批量处理时报告每个子任务的进度
func batchProcess(ctx context.Context, items []Item) error {
    progressCh := make(chan Progress, 10)

    // 启动进度收集 goroutine
    go collectProgress(ctx, progressCh)

    for _, item := range items {
        // 每完成一个子任务，发送进度
        result, err := processItem(ctx, item)
        if err != nil {
            // 通过 error channel 回传错误
            sendError(ctx, err)
        }
        progressCh <- Progress{Done: 1, Total: len(items)}
    }

    close(progressCh)
    return nil
}

4.2 使用 sync.ErrGroup 并行处理

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

func parallelProcessing(ctx context.Context) error {
    g, ctx := errgroup.WithContext(ctx)

    for _, item := range items {
        item := item // 捕获循环变量
        g.Go(func() error {
            return processItem(ctx, item)
        })
    }

    // 任意一个错误会导致整个 group 取消
    return g.Wait()
}

4.3 回调模式传递值

// 场景：需要将内部处理结果传回给调用方
func processWithCallback(ctx context.Context, callback func(result Result)) error {
    resultCh := make(chan Result, 1)

    go func() {
        // 执行业务逻辑
        result := compute()
        select {
        case resultCh <- result:
        case <-ctx.Done():
        }
    }()

    select {
    case result := <-resultCh:
        callback(result)
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}

5. 内存泄漏的常见场景与防范

context 是最容易导致内存泄漏的地方之一，因为它的生命周期与 Goroutine 的执行紧密耦合。

5.1 泄漏场景：未取消的长生命周期 context

// 错误：context 泄漏
func startBackgroundTask() {
    ctx := context.Background() // 根 context，永不取消
    for {
        select {
        case <-time.After(time.Minute):
            // 永远不会取消，goroutine 持续运行
            doPeriodicTask(ctx)
        }
    }
}

// 正确：使用可取消的 context
func startBackgroundTask(ctx context.Context) {
    for {
        select {
        case <-ctx.Done():
            return // 响应取消
        case <-time.After(time.Minute):
            doPeriodicTask(ctx)
        }
    }
}

5.2 泄漏场景：WithValue 链过长

// 错误：创建大量短命 context 时保留长链
for i := 0; i < 1000000; i++ {
    ctx := context.WithValue(parentCtx, key{i%10}, value{i})
    go func() {
        // ctx 引用了完整的父链
        // 如果父链很大，会保留大量无用的键值对
    }()
}

// 正确：使用轻量级的 key 结构
type keyType struct{}
var globalKey = keyType{} // 包级变量，内存稳定

// 或使用自定义 key 类型避免冲突
type stringKey struct{}
func (k stringKey) String() string { return "stringKey" }
func (k stringKey) Equal(other interface{}) bool {
    return k == other
}

5.3 泄漏场景：在 channel 关闭后继续发送

// 错误：忽略 context 取消，继续发送
func sendLoop(ch chan<- Event, ctx context.Context) {
    for {
        event := getNextEvent()
        select {
        case ch <- event:
        case <-ctx.Done():
            return
        }
    }
}

// 正确：使用 context 控制发送超时
func sendWithTimeout(ch chan<- Event, ctx context.Context, event Event) error {
    select {
    case ch <- event:
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}