1. 从一段被评审打回的并发代码说起前阵子帮一个团队做代码走查翻到一个“批量同步用户资料”的函数主干大概 twenty 行一个for循环循环体里直接go func()外面挂一个sync.WaitGroup。逻辑上没毛病跑起来也能用但只要网络抖一下、某个下游接口返回了异常数据、里面某个解析函数抛了panic整个进程就直接躺了而且是那种连堆栈都不好定位的躺——因为panic发生在子 goroutine 里主 goroutine 已经不知道该等谁了。这类代码在业务里遍地都是写的时候顺手出问题的时候抓瞎。并发库 conc 就是冲着这个痛点来的。它不是一门新语言也不是一个新框架它是一套建立在 Go 原生goroutine之上的结构化并发工具箱把WaitGroup、并发池、切片并行遍历、流式管道、panic捕获这几件最常用的活儿用一套统一的、带类型参数的 API 包起来。用上它之后同样一段批量拉取逻辑代码量不一定变少但可读性和容错性会明显上一个台阶——至少panic不会再把进程带走错误不会再散落在十几个匿名函数里没人收。这篇文章适合两类人看一类是已经能熟练写go关键字和sync.WaitGroup但写完之后心里总有点不踏实的同学另一类是在做服务端编排、数据批处理、爬取调度这类“天然扇出”场景想找一套比自己手搓更稳的抽象的老手。我会重点讲清楚每个组件解决的是什么问题、内部大概是怎么实现的、迁移的时候哪几行必须改、以及我自己踩过的几个坑。2. 裸写 goroutine 到底哪里不友好2.1 panic 逃逸是第一个致命问题go关键字启动的函数运行在独立的 goroutine 上它没有调用者也就没有一个天然的recover边界。原生sync.WaitGroup只负责计数不负责兜底所以只要子 goroutine 里出现空指针、数组越界、类型断言失败这类运行时异常整个进程就会按默认行为终止。很多团队的应对方式是在每个并发函数的第一行手写defer func() { recover() }()但这东西写多了容易变形——有的只打日志不上报有的 recover 了却当成成功返回调用方拿到的是 nil 结果加 nil 错误排查起来比直接崩还难。我自己总结的判断标准很简单如果一个并发单元里可能调用外部不可控代码HTTP 客户端、数据库驱动、第三方解析器、反射、JSON 反序列化那它就必须有panic边界。手写recover的问题是它没有统一语义——recover之后这次任务算成功还是失败错误以什么形式返回给调用方这些问题每个作者都会给出不同答案最后整个仓库风格分裂。2.2 无边界扇出是第二个问题for循环里起 goroutine 最爽的地方在于不用管上限但这也正是它最危险的地方。假设你从一个表里捞了五万条待刷新记录每条起一个 goroutine 去打下游接口瞬间就是五万个活跃协程加五万个连接请求。Go 的 goroutine 确实轻但它后面的连接池、文件描述符、下游服务的承载能力不是无上限的。我见过因为这类写法把下游打挂、然后触发级联超时的案例最后查出来就是一个没有任何限制的for go。原生的做法一般是chan struct{}当信号量代码长得像这样进循环前塞 N 个令牌进循环后取一个函数里defer还一个。逻辑是对的但这段“令牌管道”代码会在每个需要限流的地方重复出现一遍而且极容易写错defer的位置导致令牌不归还、整个流程卡死。2.3 错误收集是第三个问题sync.WaitGroup完全没有错误通道。于是大家只能搞一个sync.Mutex保护的[]error或者用一个带缓冲的 channel 收错误然后Wait之后再统一读取。写法大同小异但细节各异有人忘记加锁有人 channel 缓冲设小了导致 goroutine 泄漏有人在Wait之前就把 channel 关了。这些都不是难问题但它们属于重复的、无价值的、容易出错的劳动本来不应该由每个业务开发者重新发明一遍。conc 的价值就在于它把这三件事都固化成标准答案并且是用泛型写出来的、类型安全的答案。3. conc 的四个构件各管一段路3.1 conc.WaitGroup把 panic 装进口袋它长得跟sync.WaitGroup几乎一样区别是Go方法只接受func()一个签名并且内部包了一层panics.Catcher。用法var wg conc.WaitGroup for _, id : range ids { id : id wg.Go(func() { refreshUser(id) }) } wg.Wait() // 如果某个任务 panic 了这里会重新抛出关键点有两个。第一wg.Go里发生的panic会被捕获并暂存不会立即炸进程等到Wait返回时再在主 goroutine 上重新抛出——这样堆栈信息会完整地指向原始panic的位置日志里一眼就能看出来是谁炸的比在子 goroutine 里recover然后自己打日志强得多。第二如果你不想让Wait直接炸可以用WaitAndRecover它返回(panicked bool, recovered panics.Recovered)你可以自己决定是上报监控还是降级处理。这个设计的好处在于语义统一并发单元里的panic等价于不可恢复错误它会在聚合点暴露出来而不是被某个匿名recover悄悄吞掉。我的建议是对于“这批任务里只要有一个炸了就应该整体失败”的场景直接用Wait对于“部分失败可容忍、需要继续跑完并收集”的场景用WaitAndRecover。3.2 pool带限流和错误聚合的任务分发器pool是我用得最多的一个。它解决的是“我要并发干 N 件事但最多只能同时干 M 件并且要把所有错误收回来”这个复合需求。基础用法p : pool.New().WithErrors().WithMaxGoroutines(8) for _, url : range urls { url : url p.Go(func() error { return fetch(url) }) } if err : p.Wait(); err ! nil { // err 是一个聚合错误包含所有失败任务的原因 }三个With方法值得展开说。WithMaxGoroutines(n)设置的是并发上限。这里有个容易误解的地方p.Go在达到上限时会阻塞当前调用方它不是把任务丢进队列就返回。这意味着你可以在主 goroutine 里安全地循环提交十万个任务而不用担心任务队列无限膨胀因为提交过程本身就被限速了。代价是提交循环和消费是耦合同步的如果你的任务提交速度本身也需要异步那就得再套一层。我第一次用的时候以为它像errgroup那样只是限制活跃数结果发现在某些复合场景下主流程被卡住了才反应过来这个语义。WithErrors()让Wait返回一个聚合错误。默认是multierror里面包含所有任务的错误如果只关心第一条加WithFirstError()会快一些在“第一个失败就该停”的场景下更合适。WithContext(ctx)会给池子挂一个上下文取消时所有还没启动的任务会被跳过Wait会返回上下文相关的错误。这个在服务请求链路里非常有用尤其是上游客户端已经断开、没必要继续算下去的时候。还有一个带结果收集的变体pool.NewWithResults[T]()它的Wait直接返回[]T顺序和任务提交顺序一致不需要自己搞带锁的切片。这个在“并发查 N 个 ID、结果要按原顺序拼装”的场景里特别顺手p : pool.NewWithResults[*User]().WithMaxGoroutines(16) for _, id : range ids { id : id p.Go(func() *User { return loadUser(id) }) } users : p.Wait() // 顺序与 ids 一致3.3 iter切片并行的最短路径如果你的数据源本来就是一个切片而且任务是 CPU 密集型或者纯内存操作iter比pool更直接iter.ForEach(rows, func(row *Row) { row.Score compute(row) })没有Wait没有创建池子一行搞定。同类 API 还有ForEachIdx带下标、Map一对一变换并返回新切片、MapErr变换可能失败。Map的实现思路很朴素预分配一个等长结果切片按下标并行写入各写各的位置所以不需要锁。这也是它比pool.NewWithResults更轻的原因——省掉了结果收集的中间结构。有个细节必须记住iter系列传进来的函数参数是元素本身。如果元素是结构体而不是指针函数里对参数的修改不会反映回原切片。想要原地改就必须用指针切片或者下标访问。这个坑我在处理一坨[]Item的时候踩过一次跑完发现数据一个都没变还以为是并发写冲突查了半天才反应过来是值拷贝。3.4 stream当数据源是 channelstream和iter的 API 长得几乎一样但语义完全不同iter是对一个已知长度的切片做并行分片stream是从一个 channel 里持续消费适合数据源本身就是流式的场景——比如从消息队列读、从文件逐行扫描、从上游管道接数据。in : make(chan *Job, 64) go produce(in) stream.ForEach(in, func(j *Job) { handle(j) })这里的选择逻辑很清楚数据已经全在内存里、长度已知用iter数据是边产生边消费、长度不可知用stream。stream内部会把输入 channel 拆给多个工作协程你不必自己写分发逻辑。要注意的是它的内存占用受输入 channel 的缓冲大小影响如果生产端远快于消费端缓冲会堆满然后生产端阻塞——这其实是好事是一种天然背压但如果你把缓冲开得特别大就等于把背压机制废掉了。4. 从原生写法迁移过来的逐行对照4.1 sync.WaitGroup 换成 conc.WaitGroup原来可能是这样var wg sync.WaitGroup errCh : make(chan error, len(items)) for _, it : range items { wg.Add(1) go func(it Item) { defer wg.Done() if err : process(it); err ! nil { errCh - err } }(it) } wg.Wait() close(errCh) for err : range errCh { log.Println(err) }改成 conc 之后p : pool.New().WithErrors() for _, it : range items { it : it p.Go(func() error { return process(it) }) } if err : p.Wait(); err ! nil { log.Println(err) }改动点在于错误通道、Add/Done、闭包传参这三个容易出错的地方全部消失panic也顺带被兜住了。我这里直接用pool而不是conc.WaitGroup是因为原代码已经在收错误了用池子更贴合原意。如果原代码根本不需要错误收集那换成conc.WaitGroup就够了一行wg.Go替代Add go defer Done三件事。4.2 手写信号量换成 WithMaxGoroutines原来那种“进循环塞令牌、defer还令牌”的写法直接换成p.WithMaxGoroutines(n)。这里唯一的决策点是n 取多少。我的经验公式是先看下游的承受能力比如对方限流是 200 QPS、单次请求平均 100ms那么理论并发是 20取 15 到 20 比较稳如果任务本身是 CPU 密集的比如图片解码、文本分析就取runtime.NumCPU()左右超了只会增加调度开销如果是混合型取两者中较小的那个再打个八折。别上来就写 100那基本等于没限流。4.3 管道式处理链换成 stream原来用两三个 channel 串起来的流水线很多时候可以直接用stream.ForEach简化掉分发那一层。但要注意stream只做“消费并发”不做“阶段串联”。如果你的处理逻辑本身是多阶段的解析 → 清洗 → 落库每个阶段速率不同那你还是需要显式的多级 channelstream只适合用在最后一跳或者单级扇出。这个边界我一开始没分清试图用stream拼多级流水线结果写出来的代码比原来还绕。5. panic 与取消传播最容易被误解的两块5.1 panic 在哪一层被捕获conc 的pool和iter都会在每个并发单元外面套panics.Catcher。这意味着子任务里的panic不会立刻终止进程而是被记录下来等到Wait时统一重抛。但这里有个非常隐蔽的坑如果一个任务panic了池子里其它已经启动的任务并不会被取消。Wait会等到所有任务都跑完然后把panic抛出来。所以如果你指望“一个炸了立刻停”pool默认不是这个行为你得自己配合WithContext和一个取消函数来实现。我的做法通常是这样ctx, cancel : context.WithCancel(context.Background()) defer cancel() p : pool.New().WithErrors().WithContext(ctx).WithMaxGoroutines(8) for _, it : range items { it : it p.Go(func() error { if err : handle(ctx, it); err ! nil { cancel() // 主动打断其它任务 return err } return nil }) } err : p.Wait()注意cancel()之后其它任务是否真的会退出取决于你的handle有没有认真检查ctx.Done()。如果handle里全是不带上下文的阻塞调用比如某个老 SDK 的同步方法那取消就是空话。这一点跟 conc 无关是上下文机制本身的限制但很多人把WithContext当成万能开关这是误解。5.2 取消之后 Wait 返回什么WithContext生效后Wait返回的错误里会包含上下文错误。这里有一个语义细节需要注意当一个任务因为上下文取消而被跳过时它不会产生业务错误但整体Wait依然会返回非 nil。所以你不能简单地用err ! nil判断“业务失败”得配合errors.Is(err, context.Canceled)之类的判断区分“被取消”和“真出错”。这个在链路超时场景里尤其重要否则你的监控面板上会全是“错误”实际上都是正常的超时取消。5.3 iter 里的 panic 会怎样iter.ForEach内部同样有panics.Catcherpanic会在ForEach返回时重抛。但因为它是同步 API没有Wait抛出点就在调用ForEach的那一行堆栈相对清晰。这里不建议在传给iter的函数内部自己再写recover会打破统一的语义让上层以为一切正常。6. 几个真会踩到的坑6.1 循环变量的共享问题虽然较新的语言版本已经改了for range的变量捕获语义但如果你维护的代码库还停留在旧版本或者在for里用的是range之外的索引变量就还得手动做一次拷贝for _, it : range items { it : it // 必须否则所有 goroutine 看到的是同一个 it p.Go(func() error { return handle(it) }) }这个坑的可怕之处在于它不是必现的。任务跑得快的时候你可能碰巧拿到正确值压测一上来、调度一乱就会发现结果全错但错误为零。我的习惯是不管什么版本只要在for里起闭包一律加一行拷贝多打一个字符总比半夜查数据错乱强。6.2 pool 的结果切片和错误索引不对应用NewWithResults的时候Wait返回的结果切片长度是任务提交数顺序对齐提交顺序。但如果你中途用WithFirstError或者上下文跳过了一些任务某些位置的值会是零值。这个零值到底是“任务真的返回了零值”还是“任务没跑”从结果切片上区分不出来。所以只要涉及可能跳过的场景我就不会用纯结果切片而是让结果本身带一个状态字段或者干脆用pool.New().WithErrors()加自己的结构体收集。6.3 嵌套使用导致并发数翻倍很常见的一种写法是外层一个iter.ForEach遍历批次内层每个元素又用pool去并发调下游。两层的并发数相乘实际打出去的量是 N×M而你在WithMaxGoroutines里写的那个 M 只约束了内层。要控制总量外层就不能用无上限的iter得改成带限流的pool或者在内层共用一个全局池子。这个问题在做多层聚合统计的时候特别容易撞上。6.4 把 pool 当成长期运行的任务队列pool的设计目标是“一批任务跑完就结束”Wait之后这个池子就该丢掉了。拿它当常驻 worker 池反复提交、从不Wait会导致内部状态一直累积。长期运行的后台任务队列还是老老实实用显式的 worker goroutine 加一个 channel 更合适。7. 和手写方案、其它方案的对照7.1 什么情况下不该用如果只是两三个 goroutine、一两个错误分支、生命周期短到一眼能看完那手写sync.WaitGroup反而更直观引一个额外依赖换来的收益不大。conc 真正开始产生价值的临界点大概是并发数超过一屏、错误路径超过两条、或者有panic风险的时候。另外如果团队已经有了一套自己的并发编排封装并且对panic语义有自己的约定比如统一转成业务错误码那硬塞 conc 进去会造成两套语义并存反而更难维护。这种情况下的正确做法不是引入新库而是把现有封装的panic边界补上。7.2 和 errgroup 的取舍很多人会拿它和官方扩展库里的errgroup对比。errgroup的优势是生态成熟、和context的集成非常自然、出错即取消的语义开箱即用。conc 的优势则在于更细的并发上限控制errgroup.SetLimit也有但pool.Go的阻塞式提交语义更不容易堆积、类型安全的结果收集不用自己搞带锁切片、以及默认的panic兜底。我的实际选择是纯请求编排、错误即取消这种场景用errgroup更省心批量数据处理、需要结果按序收集、任务里可能触发不可控代码的场景用 conc 更舒服。两者并不互斥同一个项目里各用各的完全正常。7.3 开销到底有多少有人担心多套一层会不会有性能损失。实测下来的结论是单次任务的额外开销在几十到一两百纳秒这个量级主要由接口包装和panics.Catcher的defer组成。对于有网络 IO 或者复杂计算的任务这点开销完全可以忽略但如果你是在对一个百万级的小切片做纯内存的加一操作确实会比裸写for慢一截这种情况下iter本身的分片开销才是主要成本而不是 conc 的抽象层。粗略经验是单个任务耗时低于 1 微秒的时候先别急着并行串行可能更快。7.4 我在项目里固定下来的几条规矩跑了一段时间之后团队里约定了几条简单规则执行下来效果不错。第一凡是go关键字出现在业务代码里必须同时出现在一个 conc 的Go方法内不允许裸写第二WithMaxGoroutines的数值必须写成一个有注释的常量说明它对应哪个下游的限流值禁止直接写字面量第三任何带上下文的池子必须有单独的日志区分“取消”和“失败”否则监控会失真。这三条落地之后跟并发相关的线上问题基本归零了剩下的都是业务逻辑本身的问题。最后分享一个我自己用顺手的小技巧给pool外面包一层极薄的辅助函数把“最大并发、是否收集全部错误、是否带上下文”这三个选项固化成一个内部标准调用业务侧只传任务函数和数据。这样做的好处是团队里不用每个人都去记这些With方法的组合也避免了有人漏写WithErrors导致错误被静默丢弃。抽象层越薄越好只包一层不做二次封装改起来也不费劲。