Golang并发任务池使用教程

在Go开发中，盲目使用大量goroutine虽看似高效，实则暗藏内存浪费、调度过载、下游服务雪崩等严重风险，而GOMAXPROCS仅控制OS线程数，无法替代真正的并发节流；本文直击痛点，详解如何用sync.WaitGroup与带缓冲channel手写轻量、可控、健壮的任务池——从防goroutine泄漏、panic恢复到任务提交非阻塞设计，提供零依赖、易理解、可落地的最小可行实现方案。

为什么直接用 go 启动大量 goroutine 容易出问题

因为 goroutine 虽轻量，但不是免费的：内存开销（默认 2KB 栈）、调度压力、上下文切换成本都会在任务量突增时暴露。更关键的是，没有节流机制会导致下游服务被压垮（比如并发调用数据库或 HTTP 接口时触发限流或超时）。runtime.GOMAXPROCS 控制的是 OS 线程数，不是并发任务上限，它不解决任务排队、拒绝、超时等实际问题。

所以你需要一个显式控制并发数的池子，而不是依赖 runtime 自行调度。

• goroutine 泄漏常见于未关闭的 channel 或无退出条件的 for-select 循环
• 不带缓冲的 channel 在任务提交时可能阻塞调用方，需根据场景决定是否设置缓冲
• 任务 panic 若未 recover，会杀死整个 worker，必须在每个 worker 内部做 defer/recover

用 sync.WaitGroup + channel 实现最小可行任务池

核心是固定数量的 worker 从任务 channel 中取任务执行，主协程通过 WaitGroup 等待全部完成。这是最可控、无第三方依赖的写法。

type TaskPool struct {
    tasks   chan func()
    workers int
    wg      sync.WaitGroup
}
func NewTaskPool(n int) *TaskPool {
return &TaskPool{
tasks:   make(chan func(), 100), // 建议设缓冲，避免 submit 阻塞
workers: n,
}
}
func (p *TaskPool) Start() {
for i := 0; i < p.workers; i++ {
p.wg.Add(1)
go func() {
defer p.wg.Done()
for task := range p.tasks {
if task != nil {
defer func() {
if r := recover(); r != nil {
log.Printf("task panic: %v", r)
}
}()
task()
}
}
}()
}
}
func (p *TaskPool) Submit(task func()) {
p.tasks <- task
}
func (p *TaskPool) Stop() {
close(p.tasks)
p.wg.Wait()
}

• 不要用 for i := 0; i 直接启动，i 变量会被闭包捕获成最后值，要用 go func(i int) 显式传参
• make(chan func(), N) 的缓冲大小不是并发数，而是待处理任务队列长度；并发数由 worker 数量决定
• Stop 时先 close channel，再 Wait()，顺序不能反，否则可能死锁

如何给任务加超时和返回结果

原始 channel 只支持“投递”，无法感知执行结果或耗时。要支持结果和超时，得把任务包装成带 done channel 和 ctx 的结构。

type Result struct {
    Value interface{}
    Err   error
}
func (p *TaskPool) SubmitWithTimeout(ctx context.Context, task func() (interface{}, error)) <-chan Result {
ch := make(chan Result, 1)
p.tasks <- func() {
select {
case <-ctx.Done():
ch <- Result{Err: ctx.Err()}
default:
val, err := task()
ch <- Result{Value: val, Err: err}
}
}
return ch
}

• 别在 worker 内部用 context.WithTimeout，超时控制应由调用方决定，worker 只响应 cancel
• 返回 而不是 Result，避免阻塞 submit 调用方
• 接收方必须消费这个 channel，否则 sender 会卡在 ch ，造成 goroutine 泄漏

生产环境建议用 golang.org/x/sync/errgroup 替代手写池

如果你只需要“限制并发数 + 汇总错误”，errgroup.Group 更简洁安全。它内置了 context 传播、panic 捕获、错误聚合，且经过长期验证。

g, ctx := errgroup.WithContext(context.Background())
g.SetLimit(5) // 并发上限
for i := 0; i < 100; i++ {
i := i
g.Go(func() error {
select {
case <-time.After(time.Second):
return fmt.Errorf("task %d done", i)
case <-ctx.Done():
return ctx.Err()
}
})
}
if err := g.Wait(); err != nil {
log.Println("at least one task failed:", err)
}

• SetLimit 是唯一需要关注的配置，其他行为（如 cancel 传播）自动生效
• 它不提供任务排队、重试、指标统计等能力，复杂调度仍需自研或引入 panjf2000/ants 这类库
• 注意 Go 方法接收的是 func() error，不支持返回多值或无错误场景，需自行包装

真正难的不是实现池子，而是定义清楚「任务失败算谁的责任」：是立即重试、丢弃、还是降级？这些逻辑不在池子本身，而在你往池子里塞的 func() 里面。

文中关于的知识介绍，希望对你的学习有所帮助！若是受益匪浅，那就动动鼠标收藏这篇《Golang并发任务池使用教程》文章吧，也可关注golang学习网公众号了解相关技术文章。