4. Go 并发实战：模式、限流与排查

4. Go 并发实战：模式、限流与排查

掌握 goroutine、channel、select、sync 之后，还需要把它们组合成稳定的工程模式。Go 并发实战的关键是：限制并发、传播取消、收集错误、保证收尾。

Worker Pool：固定数量处理任务

worker pool 用固定数量的 goroutine 消费任务，避免”每个任务一个 goroutine”导致资源失控。

type Job struct {
    ID int
}

type Result struct {
    JobID int
    Err   error
}

func workerPool(ctx context.Context, jobs <-chan Job, workerCount int) <-chan Result {
    results := make(chan Result)
    var wg sync.WaitGroup

    for i := 0; i < workerCount; i++ {
        wg.Add(1)
        go func(workerID int) {
            defer wg.Done()
            for {
                select {
                case <-ctx.Done():
                    return
                case job, ok := <-jobs:
                    if !ok {
                        return
                    }
                    results <- handle(ctx, workerID, job)
                }
            }
        }(i)
    }

    go func() {
        wg.Wait()
        close(results)
    }()

    return results
}

这个版本还有一个隐藏风险：results <- ... 如果下游不再接收，worker 会阻塞。更完整的写法应该让发送结果也监听取消：

result := handle(ctx, workerID, job)
select {
case results <- result:
case <-ctx.Done():
    return
}

worker pool 的检查清单：

  1. worker 数量是否有上限？
  2. jobs 由谁关闭？
  3. results 由谁关闭？
  4. 下游提前退出时，worker 能否退出？
  5. 任务处理函数是否也接收 context？

Semaphore：限制并发数量

如果不想维护完整 worker pool，只想限制同时执行的任务数量，可以用带 buffer 的 channel 做 semaphore：

func runAll(ctx context.Context, items []Item, limit int) error {
    sem := make(chan struct{}, limit)
    errCh := make(chan error, len(items))
    var wg sync.WaitGroup

    for _, item := range items {
        select {
        case sem <- struct{}{}:
        case <-ctx.Done():
            return ctx.Err()
        }

        wg.Add(1)
        go func(item Item) {
            defer wg.Done()
            defer func() { <-sem }()

            if err := process(ctx, item); err != nil {
                errCh <- err
            }
        }(item)
    }

    wg.Wait()
    close(errCh)

    for err := range errCh {
        if err != nil {
            return err
        }
    }
    return nil
}

sem <- struct{}{} 表示获取许可，<-sem 表示释放许可。buffer 大小就是最大并发数。

这种模式适合批量请求外部 API、批量处理文件、批量生成缩略图等任务。

Fan-out / Fan-in：并行处理再合并

Fan-out 是把一条输入流分给多个 worker，Fan-in 是把多个 worker 的输出合并。

func fanOut(ctx context.Context, in <-chan int, workers int) []<-chan int {
    outs := make([]<-chan int, 0, workers)
    for i := 0; i < workers; i++ {
        out := make(chan int)
        outs = append(outs, out)

        go func() {
            defer close(out)
            for n := range in {
                select {
                case out <- n * n:
                case <-ctx.Done():
                    return
                }
            }
        }()
    }
    return outs
}

func fanIn(ctx context.Context, inputs ...<-chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup

    for _, input := range inputs {
        wg.Add(1)
        go func(ch <-chan int) {
            defer wg.Done()
            for v := range ch {
                select {
                case out <- v:
                case <-ctx.Done():
                    return
                }
            }
        }(input)
    }

    go func() {
        wg.Wait()
        close(out)
    }()

    return out
}

两个要点：

  • 多个 goroutine 从同一个 input channel 读，天然会分摊任务
  • 合并输出时，必须等所有输入都结束后再关闭 out

Pipeline：分阶段处理

pipeline 把任务拆成多个阶段，每个阶段只关心自己的输入和输出：

read files → parse → validate → write database

每个阶段的标准形状：

func stage(ctx context.Context, in <-chan Item) <-chan Item {
    out := make(chan Item)
    go func() {
        defer close(out)
        for item := range in {
            result := transform(item)
            select {
            case out <- result:
            case <-ctx.Done():
                return
            }
        }
    }()
    return out
}

pipeline 最常见的问题是下游提前退出，上游还在发送，导致 goroutine 泄漏。所以每个发送点都应该带上取消分支：

select {
case out <- result:
case <-ctx.Done():
    return
}

如果某个阶段发现错误，应当取消整个 pipeline，而不是只让自己退出。

errgroup：错误收集与取消

并发任务需要”一个失败，全部取消”时，errgroup 比手写 WaitGroup 更合适：

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

    for _, url := range urls {
        url := url
        g.Go(func() error {
            req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
            if err != nil {
                return err
            }

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

            if resp.StatusCode >= 400 {
                return fmt.Errorf("%s: %s", url, resp.Status)
            }
            return nil
        })
    }

    return g.Wait()
}

errgroup.WithContext 的语义是：任意一个 goroutine 返回非 nil error，派生出的 ctx 会被取消。其他任务如果正确监听 ctx，就会尽快退出。

如果项目不想引入 golang.org/x/sync/errgroup，也可以继续用 WaitGroup + error channel，只是样板代码会更多。

常见问题一：死锁

典型死锁：没有 goroutine 能继续推进。

func main() {
    ch := make(chan int)
    ch <- 1 // 没有接收者，main 阻塞
}

运行时可能报：

fatal error: all goroutines are asleep - deadlock!

常见原因：

  • 无缓冲 channel 只有发送没有接收
  • range ch 等待关闭，但发送方忘记 close
  • WaitGroup 计数没有归零
  • 锁没有释放
  • 多把锁顺序不一致

排查时先看 goroutine dump，找每个 goroutine 卡在哪里。

常见问题二：goroutine 泄漏

泄漏是 goroutine 没有退出，但程序还在运行：

func search(query string) <-chan Result {
    ch := make(chan Result)
    go func() {
        ch <- doSearch(query)
    }()
    return ch
}

func handler(ctx context.Context) {
    ch := search("go")
    select {
    case result := <-ch:
        _ = result
    case <-ctx.Done():
        return // search goroutine 可能还在发送，泄漏
    }
}

修复方式一：给结果 channel 加 buffer，保证发送方能结束：

ch := make(chan Result, 1)

修复方式二：让发送方也监听 context：

func search(ctx context.Context, query string) <-chan Result {
    ch := make(chan Result, 1)
    go func() {
        defer close(ch)
        result := doSearch(query)
        select {
        case ch <- result:
        case <-ctx.Done():
            return
        }
    }()
    return ch
}

常见问题三：过度并发

下面的代码在数据量小时没问题，数据量大时会把服务打爆：

for _, id := range ids {
    go fetchUser(id)
}

过度并发的症状：

  • CPU 飙高
  • 内存上涨
  • goroutine 数量持续增长
  • 外部 API 返回限流错误
  • 数据库连接池耗尽

修复方向：

  • 给并发数设置上限
  • 给每个请求设置超时
  • 使用连接池和队列
  • 做批处理而不是无限 fan-out

常见问题四：忽略 backpressure

如果生产速度长期大于消费速度，buffer 只会延迟问题，不会解决问题：

jobs := make(chan Job, 100000)

大 buffer 可能让生产者看起来很顺畅，但真实代价是内存堆积和处理延迟。更好的方式是让 channel 小一些，让生产者在消费者跟不上时自然阻塞，形成 backpressure。

jobs := make(chan Job, workerCount*2)

队列长度应该是系统设计的一部分，而不是随手写一个很大的数字。

排查工具

race detector

检查数据竞争：

go test -race ./...

pprof

查看 goroutine 数量和阻塞位置：

import _ "net/http/pprof"

go func() {
    log.Println(http.ListenAndServe("localhost:6060", nil))
}()

常用地址：

http://localhost:6060/debug/pprof/goroutine?debug=2
http://localhost:6060/debug/pprof/profile
http://localhost:6060/debug/pprof/heap

runtime.NumGoroutine

简单监控 goroutine 数量：

log.Println("goroutines:", runtime.NumGoroutine())

如果某个接口压测后 goroutine 数量只涨不跌，大概率有泄漏。

trace

go test -trace 可以看到 goroutine 调度、阻塞、网络、系统调用等时间线：

go test -trace trace.out ./pkg/...
go tool trace trace.out

trace 适合排查”为什么并发了但不快”、“时间到底卡在哪里”这类问题。

并发设计检查表

写并发代码前，可以先问自己：

  1. 最大并发数是多少？
  2. 每个 goroutine 的退出条件是什么？
  3. 上游失败时，下游会不会退出？
  4. 下游提前退出时，上游会不会阻塞？
  5. 错误由谁收集？
  6. channel 由谁关闭？
  7. 共享状态是否有锁、atomic 或 channel 保护？
  8. 是否能用 go test -race 覆盖关键路径？

小结

Go 并发实战不是把所有东西都丢进 goroutine，而是建立一套清楚的生命周期：

  • 用 worker pool 或 semaphore 限制并发
  • 用 context 传播取消
  • 用 errgroup 或 WaitGroup 收尾
  • 用 channel 表达数据流
  • 用 mutex 或 atomic 保护共享状态
  • 用 race、pprof、trace 排查问题

稳定的并发代码通常看起来并不花哨：入口有限流，中间有取消，出口能关闭，错误能返回。