Go · 并发模式实战

场景 1 · 站点爬取: 8 并发 worker pool + 首错全体收摊

50 万 URL 抓取, 目标站单 IP 限并发, 且任一 worker 拿到封禁错误就该全队撤退:

func crawl(ctx context.Context, urls []string) error {
    g, ctx := errgroup.WithContext(ctx)          // 控制线: 首错即取消
    jobs := make(chan string, 200)              // 缓冲吸收突发, 不打爆内存
    worker := func() error {
        for u := range jobs {
            if err := fetch(ctx, u); err != nil { return err } // 级联取消
        }
        return nil
    }
    for i := 0; i < 8; i++ { g.Go(worker) }  // 8 = 目标站单 IP 上限
feed:
    for _, u := range urls {
        select {
        case jobs <- u:
        case <-ctx.Done(): break feed     // 某 worker 已挂: 停止投喂
        }
    }
    close(jobs)                              // 发送方负责关闭
    return g.Wait()                             // 只返回第一个错误
}

对比裸开 goroutine-per-URL: 峰值并发可控, 被封禁时 0.1s 内全队停止, 不再浪费请求配额。

场景 2 · 批量调下游: fan-out 8 并发, 结果保序归还

批量查 2000 个订单状态, 下游只扛 8 并发, 但调用方要求返回顺序与入参一致:

func fetchAll(ctx context.Context, ids []string) []Result {
    sem := make(chan struct{}, 8)           // 下游接口只扛 8 并发
    out := make(chan indexed, len(ids))     // 容量=任务数: 写端永不阻塞
    var wg sync.WaitGroup
    for i, id := range ids {
        wg.Add(1)
        go func(i int, id string) {        // 1.22 前必须传参捕获
            defer wg.Done()
            sem <- struct{}{}                  // 拿到令牌才出发
            defer func() { <-sem }()          // panic 路径也要归还
            out <- indexed{i, callAPI(ctx, id)}  // 带下标: 保序关键
        }(i, id)
    }
    go func() { wg.Wait(); close(out) }()  // 等全部写端退出才能关
    res := make([]Result, len(ids))
    for it := range out { res[it.i] = it.r }     // 按下标归位还原顺序
    return res
}

2000 单从串行 4 分钟压到 8 并发 32 秒, 且各写各的下标 —— 无锁也无数据竞争。

场景 3 · 10GB 日志流式 ETL: 读→解析→写库三阶段 pipeline

全量读进内存必 OOM, 三阶段各一个 goroutine, 数据像水流过去, 任意时刻只在内存里驻留一小段:

func etl(ctx context.Context, path string) error {
    return writeStage(ctx, parseStage(ctx, readStage(ctx, path)))
}                                                // parse/write 同构: range 上游
func readStage(ctx context.Context, p string) <-chan string {
    out := make(chan string, 512)               // 小缓冲: 背压自然形成
    go func() {
        defer close(out)                         // 谁开 chan 谁收尾
        f, err := os.Open(p)
        if err != nil { log.Print(err); return }
        defer f.Close()
        sc := bufio.NewScanner(f)
        for sc.Scan() {
            select {
            case out <- sc.Text():               // 满则阻塞: 反压上游停读
            case <-ctx.Done(): return            // 下游不要了就停读
            }
        }
    }()
    return out
}

512MB 容器跑通 10GB 文件, RSS 稳定在 90MB: 库写得慢, 读端就自动慢下来, 这就是背压。

场景 4 · Go 1.20+ 用 errgroup.SetLimit 替掉手搓信号量

老代码里 sem chan + defer 归去归来容易漏, SetLimit 一行等价且语义更直白:

func batchProcess(ctx context.Context, items []Task) error {
    g, ctx := errgroup.WithContext(ctx)
    g.SetLimit(16)                            // 内置信号量: 最多 16 并发
    for _, t := range items {
        g.Go(func() error {                    // 拿不到令牌时阻塞在这
            if err := process(ctx, t); err != nil {
                return fmt.Errorf("task %d: %w", t.ID, err)
            }
            return nil
        })
    }
    return g.Wait()  // 一行替掉手搓 sem chan + defer 归还
}

注意 g.Go 在满员时会阻塞提交循环 —— 这正是限流语义, 不是 bug。

场景 5 · 夜间对账跑批: 首错策略与部分结果报表

金融对账 10 万账户: 单户失败不该连坐取消全局, 但要记进报表第二天人工复核:

func nightlyReconcile(ctx context.Context, accts []string) (Report, error) {
    var mu sync.Mutex
    ok, failed := 0, 0                        // 部分结果也要记账
    g, ctx := errgroup.WithContext(ctx)
    g.SetLimit(32)
    for _, acc := range accts {
        acc := acc
        g.Go(func() error {
            if err := reconcile(ctx, acc); err != nil {
                mu.Lock(); failed++; mu.Unlock()
                return nil                 // 不连坐: 记录后继续
            }
            mu.Lock(); ok++; mu.Unlock()
            return nil
        })
    }
    err := g.Wait()                              // 全部跑完才出报表
    return Report{OK: ok, Failed: failed}, err
}

若业务要求 fail-fast (一个错全停), 把 return nil 换成 return err 即可 —— 两种策略只差这一行。

场景 6 · 手写 done channel: 读懂标准库取消模式的祖宗

面试与源码都绕不开: close(done) 广播 vs send 只唤醒一个, 这是 context 的思想原型:

// done 是 ctx 出现前的取消原语, 标准库 ctx 同款思想
func worker(done <-chan struct{}, jobs <-chan Job) {
    for {
        select {
        case j, ok := <-jobs:                  // ok=false: 上游 close
            if !ok { return }
            process(j)
        case <-done:                         // close(done) 是广播
            return                            // 所有监听者同时收到
        }
    }
}
quit := make(chan struct{})
close(quit)   // 广播取消: 绝不是 quit <- struct{}{} (只唤醒一个)

场景 7 · 图像缩略服务: 信号量控内存, 10×30MB 封顶

不限并发时 500 张图同时解码, 300MB×500 直接 OOMKilled; 信号量把"在制品数量"钉死:

func processImages(ctx context.Context, keys []string) error {
    sem := make(chan struct{}, 10)          // 10×30MB ≈ 内存封顶
    var wg sync.WaitGroup
    for _, k := range keys {
        wg.Add(1)
        go func(k string) {
            defer wg.Done()
            select {
            case sem <- struct{}{}:             // 先拿内存额度
            case <-ctx.Done(): return
            }
            defer func() { <-sem }()        // defer: panic 也归还
            img, err := loadFromOSS(ctx, k)      // 约 30MB 解码缓冲
            if err != nil { return }
            uploadThumbnail(ctx, img, k)         // 加工完即可被 GC
        }(k)
    }
    wg.Wait()
    return nil
}

控并发就是控内存: 波峰 RSS 从无界的 15GB 变成稳定 300MB+基础开销。

场景 8 · for-select 优雅退出模板: ctx.Done 先到也要排空队列

K8s 发 SIGTERM 时正在排队的任务不能丢, 但也不能无限等 —— 排空 + 上限两件事都做:

func serve(ctx context.Context, tasks <-chan Task) error {
    for {
        select {
        case <-ctx.Done():                    // 收到停止信号
            return drainAll(tasks)              // 先排空存量: 不丢任务
        case t, ok := <-tasks:
            if !ok { return nil }              // 上游 close: 正常收工
            if err := handle(ctx, t); err != nil { return err }
        }
    }
}
func drainAll(tasks <-chan Task) error {
    bg := context.Background()                   // 收尾用新 ctx: 原已取消
    for t := range tasks { handle(bg, t) }
    return nil
}

排空也要配 deadline (外层 WithTimeout 包一层), 防止上游永不 close 把优雅退出拖成永久挂起。

场景 9 · 万级任务进度上报: atomic 计数 + Ticker 定时打印

跑批 40 分钟没任何输出会被值班当成挂了; worker 里直接 log 又太吵 —— 独立协程定时打快照:

var done int64                                // 原子计数: int++ 是竞态
total := int64(len(tasks))
var wg sync.WaitGroup
for _, t := range tasks {
    wg.Add(1)
    go func(t Task) {
        defer wg.Done()
        process(t); atomic.AddInt64(&done, 1)
    }(t)
}
go func() {                                 // 独立进度协程: 只读快照
    tk := time.NewTicker(5 * time.Second)
    defer tk.Stop()                           // Ticker 不 Stop 就泄漏
    for range tk.C {
        p := atomic.LoadInt64(&done)
        log.Printf("progress %d/%d", p, total)
        if p == total { return }
    }
}()
wg.Wait()

场景 10 · goroutine 三天涨到 18 万: 一次泄漏排查实录

RSS 缓涨 + goroutine 只增不减, 先抓全量栈再对代码, 定位到"发送端退出却没人 close":

// 症状: 上线 3 天 goroutine 2k → 180k, RSS 缓涨
// curl :6060/debug/pprof/goroutine?debug=1 抓栈定位:
//   180000 @ ... main.(*Client).Watch  watch.go:42
// 42:  for msg := range c.stream  // 发送端退出却没人 close
func (c *Client) Watch(ctx context.Context) {
    defer c.wg.Done()
    for {
        select {                            // 修复: 阻塞点全挂 ctx
        case msg, ok := <-c.stream:
            if !ok { return }
            c.handle(ctx, msg)
        case <-ctx.Done():                  // 断连/超时也能走人
            return
        }
    }
}

修复后 goroutine 曲线变成锯齿 (涨了就落), RSS 回落到基线; 原则是每个阻塞点都有退出分支。