Go并发模式：Context、Channel与Pipeline架构

一、Context包设计与取消机制

1.1 Context的核心设计

Go的context包是管理goroutine生命周期、传递请求作用域值的标准方式。其设计遵循"通过显式传递而非隐式共享"的哲学，解决了goroutine取消、超时、截止时间等问题。

Context接口定义了四个方法：Deadline()返回截止时间、Done()返回一个只读通道用于监听取消信号、Err()返回取消原因、Value(key)获取上下文中的值。

// Context接口定义
type Context interface {
    Deadline() (deadline time.Time, ok bool)
    Done() <-chan struct{}
    Err() error
    Value(key any) any
}

// 重点理解：Done()返回的channel被关闭 vs 有数据可读
// 当channel被关闭时，读取该channel会立即返回零值
// 这是Go中广播机制的标准实现——关闭channel让所有接收者都收到信号

1.2 Context的继承与取消

Context通过派生（Derived Context）形成树状结构。父Context被取消时，所有子Context也会被级联取消。这是实现请求级取消的核心机制。

package main

import (
    "context"
    "fmt"
    "time"
)

func main() {
    // 1. 创建根Context
    ctx := context.Background()

    // 2. 派生带超时的Context（3秒后自动取消）
    ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
    defer cancel()

    // 3. 启动goroutine监听取消信号
    go func() {
        <-ctx.Done() // 阻塞直到Context被取消
        fmt.Println("Context cancelled:", ctx.Err())
    }()

    // 4. 模拟工作
    select {
    case <-time.After(5 * time.Second):
        fmt.Println("Work completed normally")
    case <-ctx.Done():
        fmt.Println("Work cancelled by timeout:", ctx.Err())
    }

    // 5. 手动取消示例
    ctx2, cancel2 := context.WithCancel(context.Background())
    go func() {
        time.Sleep(100 * time.Millisecond)
        cancel2() // 手动取消
    }()
    
    <-ctx2.Done()
    fmt.Println("Manual cancel:", ctx2.Err()) // context canceled
}

关键点：cancel 函数必须被调用（通常在defer中），即使Context已经超时。这是因为Context内部会关联资源（如定时器），不调用cancel会导致资源泄漏。

二、Channel关闭原则与口头协议

2.1 Channel关闭的经典问题

Go的Channel关闭有一个著名的规则：不要从接收端关闭Channel，不要关闭已经关闭的Channel。违反这些规则会导致panic。

这导致了"谁来关闭Channel"的设计难题。在简单场景下，通常是发送者关闭Channel；但在复杂场景下（多个发送者、多个接收者），需要特殊的协调机制。

// 错误1：从接收端关闭Channel
func badReceive() {
    ch := make(chan int, 1)
    go func() {
        v := <-ch
        close(ch) // 错误！接收端不应该关闭Channel
        fmt.Println(v)
    }()
    ch <- 1
}

// 错误2：重复关闭
func badDoubleClose() {
    ch := make(chan int)
    close(ch)
    close(ch) // panic: close of closed channel
}

// 正确做法：由发送者关闭，且只关闭一次
func goodClose() {
    ch := make(chan int, 1)
    go func() {
        ch <- 1
        close(ch) // 发送者关闭
    }()
    v, ok := <-ch
    if !ok {
        fmt.Println("channel closed")
        return
    }
    fmt.Println(v)
}

2.2 多发送者场景的关闭协议

当有多个goroutine向同一个Channel发送数据时，应该怎么关闭？Go的口头协议是：如果你不是Channel的创建者，就不要关闭它。

对于多发送者场景，可以引入一个额外的"停止信号"Channel（使用空结构体chan struct{}），通知所有发送者停止发送，然后由创建者关闭数据Channel。

// 多发送者、单接收者场景的关闭协议
func multiSender() {
    dataCh := make(chan int, 10)
    stopCh := make(chan struct{}) // 由接收者控制停止信号
    
    for i := 0; i < 3; i++ {
        go func(id int) {
            for {
                select {
                case <-stopCh:
                    fmt.Printf("Sender %d stopping\n", id)
                    return
                default:
                    select {
                    case dataCh <- id:
                    case <-stopCh:
                        return
                    }
                }
            }
        }(i)
    }
    
    go func() {
        for i := 0; i < 5; i++ {
            v := <-dataCh
            fmt.Println("Received:", v)
        }
        close(stopCh) // 广播停止信号给所有发送者
    }()
}

// 更好的模式：使用context作为停止信号
func multiSenderWithContext() {
    ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
    defer cancel()
    
    dataCh := make(chan int, 10)
    var wg sync.WaitGroup
    
    for i := 0; i < 3; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            for {
                select {
                case <-ctx.Done():
                    return
                case dataCh <- id:
                }
            }
        }(i)
    }
    
    go func() {
        for v := range dataCh {
            fmt.Println("Received:", v)
        }
    }()
    
    wg.Wait()
    close(dataCh) // 所有发送者已退出，安全关闭
}

三、Pipeline模式与Fan-out/Fan-in

3.1 Pipeline模式基础

Pipeline（管道）是Go并发编程中的经典模式，它将复杂任务分解为多个阶段（stage），每个阶段通过Channel连接，形成数据流处理管道。

Pipeline模式的核心优势：各阶段解耦、易于测试、可灵活组合、支持并发执行。每个阶段本质上是一个函数，接收输入Channel，返回输出Channel。

// Pipeline架构示意
// 生成器 → 阶段1 → 阶段2 → 阶段3 → 消费者
//   │         │        │        │        │
//   └──ch1───┘        └──ch2───┘        └──ch3───┘

// 阶段函数签名模式
// type Stage func(ctx context.Context, in <-chan T) <-chan R

// 示例：数字处理Pipeline
func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
    
    numbers := generate(ctx, 1, 2, 3, 4, 5)
    squares := square(ctx, numbers)
    evens := filterEven(ctx, squares)
    
    for n := range evens {
        fmt.Println(n) // 输出：4, 16
    }
}

func generate(ctx context.Context, nums ...int) <-chan int {
    out := make(chan int, len(nums))
    go func() {
        defer close(out)
        for _, n := range nums {
            select {
            case out <- n:
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

func square(ctx context.Context, in <-chan int) <-chan int {
    out := make(chan int, cap(in))
    go func() {
        defer close(out)
        for n := range in {
            select {
            case out <- n * n:
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

func filterEven(ctx context.Context, in <-chan int) <-chan int {
    out := make(chan int, cap(in))
    go func() {
        defer close(out)
        for n := range in {
            if n%2 == 0 {
                select {
                case out <- n:
                case <-ctx.Done():
                    return
                }
            }
        }
    }()
    return out
}

3.2 Fan-out与Fan-in模式

Fan-out指一个输入被多个goroutine处理（提高并行度）；Fan-in指多个输入合并到一个输出（结果汇聚）。这两种模式是Pipeline的高级形式。

// Fan-out：启动多个worker处理同一个输入Channel
func fanOut(ctx context.Context, in <-chan int, n int) []<-chan int {
    workers := make([]<-chan int, n)
    for i := 0; i < n; i++ {
        workers[i] = worker(ctx, in)
    }
    return workers
}

func worker(ctx context.Context, in <-chan int) <-chan int {
    out := make(chan int, 10)
    go func() {
        defer close(out)
        for n := range in {
            select {
            case out <- n * 2:
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

// Fan-in：将多个Channel合并为一个
func fanIn(ctx context.Context, channels ...<-chan int) <-chan int {
    out := make(chan int, 10)
    var wg sync.WaitGroup
    wg.Add(len(channels))
    
    for _, ch := range channels {
        go func(c <-chan int) {
            defer wg.Done()
            for n := range c {
                select {
                case out <- n:
                case <-ctx.Done():
                    return
                }
            }
        }(ch)
    }
    
    go func() {
        wg.Wait()
        close(out)
    }()
    return out
}

// 完整使用
gen := generate(ctx, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
workers := fanOut(ctx, gen, 3)
result := fanIn(ctx, workers...)
for n := range result {
    fmt.Println("Result:", n)
}

四、错误处理与goroutine泄漏防护

4.1 goroutine泄漏的常见原因

goroutine泄漏是指goroutine在完成任务后无法正常退出，持续占用资源。这是Go并发编程中最常见的bug之一，特别是在长时间运行的服务中。

goroutine泄漏的根本原因：阻塞在channel操作上、阻塞在锁上、阻塞在I/O上、无限循环没有退出条件。通过使用Context和select，可以有效避免这些问题。

// goroutine泄漏示例

// 泄漏1：接收者退出，但发送者还在阻塞发送
func leakSender() {
    ch := make(chan int)
    go func() {
        for i := 0; ; i++ {
            ch <- i // 如果没人接收，这里永远阻塞
        }
    }()
    <-ch // 只接收一次就退出，发送者泄漏了
}

// 正确做法：使用Context控制goroutine生命周期
func noLeak(ctx context.Context) {
    ch := make(chan int, 10)
    
    go func() {
        for i := 0; ; i++ {
            select {
            case ch <- i:
            case <-ctx.Done():
                return // Context取消时退出
            }
        }
    }()
    
    go func() {
        for {
            select {
            case v := <-ch:
                fmt.Println(v)
            case <-ctx.Done():
                return
            }
        }
    }()
}

4.2 优雅退出的模式

优雅退出（Graceful Shutdown）是指程序收到中断信号时，能够完成正在处理的请求，释放资源，然后干净地退出。对于Go服务，这涉及监听OS信号、取消Context、等待goroutine退出。

// 优雅退出模式
package main

import (
    "context"
    "fmt"
    "net/http"
    "os"
    "os/signal"
    "syscall"
    "time"
)

func main() {
    srv := &http.Server{Addr: ":8080"}
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    // 启动HTTP服务
    go func() {
        http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
            time.Sleep(2 * time.Second)
            fmt.Fprintln(w, "Hello")
        })
        if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
            fmt.Println("HTTP server error:", err)
        }
    }()

    // 启动后台worker
    go backgroundWorker(ctx)

    // 监听中断信号
    sigCh := make(chan os.Signal, 1)
    signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
    <-sigCh

    fmt.Println("Shutting down gracefully...")
    cancel() // 通知所有goroutine退出

    // 给HTTP服务5秒时间完成现有请求
    shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer shutdownCancel()
    srv.Shutdown(shutdownCtx)

    fmt.Println("Server gracefully stopped")
}

func backgroundWorker(ctx context.Context) {
    ticker := time.NewTicker(1 * time.Second)
    defer ticker.Stop()
    for {
        select {
        case <-ticker.C:
            fmt.Println("Background work...")
        case <-ctx.Done():
            fmt.Println("Background worker exiting")
            return
        }
    }
}

五、实战：构建一个可取消的Pipeline

5.1 完整Pipeline实现

以下是一个生产级的Pipeline实现，包含错误处理、取消传播、结果收集等完整功能：

package pipeline

import (
    "context"
    "fmt"
    "sync"
)

type Result struct {
    Value interface{}
    Err   error
}

type StageFunc func(ctx context.Context, in <-chan Result) <-chan Result

// Pipeline 可取消的Pipeline
type Pipeline struct {
    ctx    context.Context
    cancel context.CancelFunc
    stages []StageFunc
    first  <-chan Result
}

func NewPipeline(ctx context.Context) *Pipeline {
    ctx, cancel := context.WithCancel(ctx)
    return &Pipeline{ctx: ctx, cancel: cancel}
}

func (p *Pipeline) AddStage(stage StageFunc) *Pipeline {
    p.stages = append(p.stages, stage)
    return p
}

func (p *Pipeline) From(generator func(ctx context.Context) <-chan Result) *Pipeline {
    p.first = generator(p.ctx)
    return p
}

func (p *Pipeline) Run() <-chan Result {
    current := p.first
    for _, stage := range p.stages {
        current = stage(p.ctx, current)
    }
    return current
}

func (p *Pipeline) Cancel() { p.cancel() }

// 使用示例
pipe := NewPipeline(context.Background())
pipe.From(func(ctx context.Context) <-chan Result {
    out := make(chan Result, 10)
    go func() {
        defer close(out)
        for i := 1; i <= 10; i++ {
            select {
            case out <- Result{Value: i}:
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
})

// 添加阶段：乘以2
pipe.AddStage(func(ctx context.Context, in <-chan Result) <-chan Result {
    out := make(chan Result, cap(in))
    go func() {
        defer close(out)
        for r := range in {
            if r.Err != nil { out <- r; continue }
            select {
            case out <- Result{Value: r.Value.(int) * 2}:
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
})

// 过滤大于10的
pipe.AddStage(func(ctx context.Context, in <-chan Result) <-chan Result {
    out := make(chan Result, cap(in))
    go func() {
        defer close(out)
        for r := range in {
            if r.Err != nil { out <- r; continue }
            v := r.Value.(int)
            if v <= 10 {
                select {
                case out <- Result{Value: v}:
                case <-ctx.Done():
                    return
                }
            }
        }
    }()
    return out
})

results := pipe.Run()
for r := range results {
    if r.Err != nil { fmt.Println("Error:", r.Err); continue }
    fmt.Println("Result:", r.Value)
}

5.2 Pipeline模式的最佳实践

在构建生产级Pipeline时，需要遵循以下最佳实践：

六、Context在微服务中的传播

6.1 gRPC中的Context传播

在微服务架构中，Context需要在服务间传播。gRPC原生支持Context——每个gRPC调用都接收一个Context参数，并且会自动将Context中的元数据（Metadata）通过HTTP/2头部传播到下游服务。

通过gRPC Metadata，我们可以将traceID、userID等请求作用域数据附加到Context中，并在服务端提取出来，实现全链路追踪。

// gRPC客户端：将Context数据通过Metadata传播
func callService(ctx context.Context) {
    md := metadata.Pairs(
        "x-trace-id", "trace-123",
        "x-user-id", "user-456",
    )
    ctx = metadata.NewOutgoingContext(ctx, md)
    
    conn, _ := grpc.Dial("localhost:50051", grpc.WithInsecure())
    defer conn.Close()
    
    client := pb.NewMyServiceClient(conn)
    resp, err := client.MyMethod(ctx, &pb.MyRequest{})
}

// gRPC服务端：从Metadata中提取Context数据
func (s *Server) MyMethod(ctx context.Context, req *pb.MyRequest) (*pb.MyResponse, error) {
    md, ok := metadata.FromIncomingContext(ctx)
    if ok {
        if traceIDs := md.Get("x-trace-id"); len(traceIDs) > 0 {
            ctx = context.WithValue(ctx, "trace_id", traceIDs[0])
        }
    }
    return &pb.MyResponse{Message: "ok"}, nil
}

6.2 HTTP服务中的Context传播

对于HTTP服务，Context通常通过请求作用域传递。Go 1.7+的net/http包原生支持Request.Context()，可以用于传递取消信号和超时。

// 中间件：从HTTP Header提取traceID并注入Context
func traceMiddleware(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        traceID := r.Header.Get("X-Trace-ID")
        if traceID == "" { traceID = "auto-generated" }
        ctx := context.WithValue(r.Context(), "trace_id", traceID)
        r = r.WithContext(ctx)
        next.ServeHTTP(w, r)
    })
}

// 业务处理：传递Context到下游
func handler(w http.ResponseWriter, r *http.Request) {
    ctx := r.Context()
    ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
    defer cancel()
    
    result, err := callDownstreamService(ctx)
    if err != nil {
        http.Error(w, err.Error(), http.StatusGatewayTimeout)
        return
    }
    fmt.Fprintf(w, "Result: %s", result)
}

// 调用下游并传播traceID
func callDownstreamService(ctx context.Context) (string, error) {
    req, _ := http.NewRequestWithContext(ctx, "GET", "http://downstream/api", nil)
    if traceID := ctx.Value("trace_id"); traceID != nil {
        req.Header.Set("X-Trace-ID", fmt.Sprintf("%v", traceID))
    }
    resp, err := http.DefaultClient.Do(req)
    if err != nil { return "", err }
    defer resp.Body.Close()
    return "ok", nil
}

七、并发模式对比（ErrGroup/Worker Pool）

7.1 ErrGroup：错误传播的并发控制

ErrGroup（来自golang.org/x/sync/errgroup）是一个简化版的goroutine组管理工具。它的核心特性：当其中一个goroutine返回错误时，整个组会被取消（通过Context）。

ErrGroup特别适合"多任务并发执行，任一失败即取消"的场景，如同时调用多个下游服务

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

func main() {
    g, ctx := errgroup.WithContext(context.Background())
    results := make([]string, 3)
    
    g.Go(func() error {
        results[0] = "task1 done"
        return nil
    })
    g.Go(func() error {
        return fmt.Errorf("task2 failed")
    })
    g.Go(func() error {
        <-ctx.Done()
        return ctx.Err()
    })
    
    if err := g.Wait(); err != nil {
        fmt.Println("Error:", err) // task2 failed
    }
}

7.2 Worker Pool：限流并发

Worker Pool是另一种经典并发模式，通过固定数量的worker处理任务队列，限制并发度，防止资源耗尽。与ErrGroup不同，Worker Pool追求的是吞吐量而不是错误传播。

// Worker Pool实现
func workerPool(ctx context.Context, tasks <-chan Task, workerCount int) <-chan Result {
    results := make(chan Result, cap(tasks))
    var wg sync.WaitGroup
    
    for i := 0; i < workerCount; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            for task := range tasks {
                select {
                case results <- processTask(id, task):
                case <-ctx.Done():
                    return
                }
            }
        }(i)
    }
    
    go func() {
        wg.Wait()
        close(results)
    }()
    return results
}

7.3 并发模式选择指南

不同的并发模式适用于不同场景。以下决策树可以帮助你快速选择：