当前位置：首页 > 文章列表 > Golang > Go教程 > Golang并发任务池使用教程

Golang并发任务池使用教程

2026-03-03 18:16:44 0浏览收藏

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

如何使用Golang实现并发任务池_Golang并发任务池管理与使用方法

为什么直接用 `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
}
<p>func NewTaskPool(n int) *TaskPool {
return &TaskPool{
tasks:   make(chan func(), 100), // 建议设缓冲，避免 submit 阻塞
workers: n,
}
}</p><p>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()
}
}
}()
}
}</p><p>func (p *TaskPool) Submit(task func()) {
p.tasks <- task
}</p><p>func (p *TaskPool) Stop() {
close(p.tasks)
p.wg.Wait()
}</p>

不要用 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
}
<p>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
}</p>

别在 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) // 并发上限
<p>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()
}
})
}</p><p>if err := g.Wait(); err != nil {
log.Println("at least one task failed:", err)
}</p>