Go语言并发模式实战与goroutine泄漏排查方法

goroutine泄漏的常见场景

Go语言的goroutine轻量级特性让开发者容易忽略其生命周期管理。每个goroutine初始栈仅2KB，但泄漏的goroutine会持续占用内存和CPU资源，最终导致OOM。生产环境中最常见的泄漏场景有三类：

1. channel无人接收：goroutine向channel发送数据，但没有接收方，发送方永远阻塞。

2. select缺少default或退出条件：select中所有case都阻塞，goroutine永远挂起。

3. HTTP请求未设超时：http.Get默认无超时，远端无响应时goroutine永远等待。

使用context控制goroutine生命周期

context.Context是Go并发安全的取消信号传播机制，是解决goroutine泄漏的核心工具：

func worker(ctx context.Context, jobs <-chan int, results chan<- int) {
    for {
        select {
        case <-ctx.Done():
            log.Println("worker shutdown")
            return
        case job, ok := <-jobs:
            if !ok {
                return
            }
            results <- process(job)
        }
    }
}

func main() {
    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()

    jobs := make(chan int, 100)
    results := make(chan int, 100)

    for i := 0; i < 10; i++ {
        go worker(ctx, jobs, results)
    }
    // 超时后所有worker收到ctx.Done()信号自动退出
}

WithTimeout设定30秒超时，超时后context自动取消，所有监听ctx.Done()的goroutine收到信号退出。

fan-out/fan-in并发模式

fan-out将一个任务分发给多个goroutine并行处理，fan-in将多个结果汇聚到同一个channel。这是Go并发最实用的模式之一：

func fanOut(ctx context.Context, input <-chan Task, workerCount int) []<-chan Result {
    channels := make([]<-chan Result, workerCount)
    for i := 0; i < workerCount; i++ {
        channels[i] = processWorker(ctx, input)
    }
    return channels
}

func fanIn(ctx context.Context, channels ...<-chan Result) <-chan Result {
    merged := make(chan Result)
    var wg sync.WaitGroup
    wg.Add(len(channels))

    for _, ch := range channels {
        go func(c <-chan Result) {
            defer wg.Done()
            for result := range c {
                select {
                case merged <- result:
                case <-ctx.Done():
                    return
                }
            }
        }(ch)
    }

    go func() {
        wg.Wait()
        close(merged)
    }()
    return merged
}

func processWorker(ctx context.Context, input <-chan Task) <-chan Result {
    out := make(chan Result)
    go func() {
        defer close(out)
        for task := range input {
            select {
            case out <- process(task):
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

fan-in中的sync.WaitGroup确保所有worker完成后才关闭merged channel，避免接收方读到零值。

使用runtime监控goroutine泄漏

Go标准库提供了goroutine监控手段：

func monitorGoroutines() {
    ticker := time.NewTicker(10 * time.Second)
    for range ticker.C {
        count := runtime.NumGoroutine()
        log.Printf("active goroutines: %d", count)
        if count > 500 {
            log.Println("WARNING: goroutine count exceeds 500")
            buf := make([]byte, 64*1024)
            n := runtime.Stack(buf, true)
            log.Printf("goroutine stack:\n%s", buf[:n])
        }
    }
}

goroutine数量持续增长且不回落，基本可以确认存在泄漏。runtime.Stack(true)打印所有goroutine的调用栈。

pprof定位泄漏goroutine

集成net/http/pprof后，通过HTTP端点获取运行时信息：

import _ "net/http/pprof"

func main() {
    go http.ListenAndServe(":6060", nil)
    // 业务代码...
}

访问goroutine profile：

curl http://localhost:6060/debug/pprof/goroutine?debug=1
go tool pprof http://localhost:6060/debug/pprof/goroutine

pprof的top命令显示哪些函数创建了最多的goroutine，list命令显示具体代码行。

errgroup：带错误处理的并发控制

sync.WaitGroup只负责等待，不处理错误。golang.org/x/sync/errgroup增加了错误传播和取消机制：

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

func fetchAll(ctx context.Context, urls []string) ([]string, error) {
    g, ctx := errgroup.WithContext(ctx)
    results := make([]string, len(urls))

    for i, url := range urls {
        i, url := i, url
        g.Go(func() error {
            data, err := httpGet(ctx, url)
            if err != nil {
                return err
            }
            results[i] = data
            return nil
        })
    }

    if err := g.Wait(); err != nil {
        return nil, err
    }
    return results, nil
}

errgroup.WithContext将context与group绑定。任一goroutine返回error，context自动取消，其余goroutine通过ctx.Done()感知并退出。