当前位置：首页 > 文章列表 > Golang > Go教程 > Go并发批处理高效技巧分享

Go并发批处理高效技巧分享

2026-01-27 23:02:35 0浏览收藏

从现在开始，努力学习吧！本文《Go语言并发批处理实战技巧》主要讲解了等等相关知识点，我会在golang学习网中持续更新相关的系列文章，欢迎大家关注并积极留言建议。下面就先一起来看一下本篇正文内容吧，希望能帮到你！

直接用 go 启动大量 goroutine 易崩因资源失控：内存暴涨、调度压力大、下游限流被击穿；须用信号量控制并发数，并配合 errgroup 统一错误处理与等待。

Go语言如何实现并发批处理_Golang批量任务并发设计

为什么直接用 `go` 启动大量 goroutine 容易崩

不是语法错，是资源失控。每个 goroutine 默认栈约 2KB，上万并发时内存暴涨；更关键的是调度器压力、上下文切换开销、以及目标服务（如数据库、HTTP API）的连接数/限流阈值被瞬间打穿。常见现象包括：runtime: out of memory、context deadline exceeded、下游返回 429 Too Many Requests。

真正要控制的不是“启动多少 goroutine”，而是“同一时刻有多少任务在执行”。所以必须引入并发限制层，而不是靠 time.Sleep 或随意加 sync.WaitGroup 就完事。

用 `semaphore` 控制并发数最直接

Go 标准库没有内置信号量，但用 chan struct{} 实现轻量级计数信号量极简单，且语义清晰：

func NewSemaphore(n int) chan struct{} {
    return make(chan struct{}, n)
}
<p>func (s chan struct{}) Acquire() {
s <- struct{}{}
}</p><p>func (s chan struct{}) Release() {
<-s
}
</p>

使用时包裹任务逻辑：

初始化信号量：sem := NewSemaphore(10) 表示最多 10 个任务并发
每个任务开始前调用 sem.Acquire()，阻塞直到有空位
任务结束（无论成功失败）必须调用 sem.Release()，建议用 defer
别把信号量传进 goroutine 再关 channel——它要复用，不能 close

配合 `errgroup.Group` 统一收口错误和等待

errgroup 不仅能等所有 goroutine 结束，还能自动传播第一个非 nil 错误，比裸写 sync.WaitGroup + 全局 error 变量更安全：

g, ctx := errgroup.WithContext(ctx)
sem := NewSemaphore(10)
<p>for _, task := range tasks {
task := task // 避免循环变量捕获问题
g.Go(func() error {
sem.Acquire()
defer sem.Release()</p><pre class="brush:php;toolbar:false"><code>    select {
    case <-ctx.Done():
        return ctx.Err()
    default:
    }

    return doTask(task) // 实际业务逻辑
})</code>

}

if err := g.Wait(); err != nil { // 处理第一个出错的任务 }

注意点：

errgroup 的 Go 方法会继承父 ctx，任务中用 select 检查取消更可靠
不要在 Go 函数里再起 goroutine 并试图用同一个 errgroup —— 它不递归管理子 goroutine
如果需要收集全部错误（而非只第一个），得自己用 sync.Mutex + 切片存错

批量结果怎么有序返回？别依赖 goroutine 启动顺序

goroutine 执行完成时间不确定，按原始索引返回结果最稳妥。常见做法是让每个任务把自己处理的下标带上：

type Result struct {
    Index int
    Data  interface{}
    Err   error
}
<p>results := make([]Result, len(tasks))
g, _ := errgroup.WithContext(ctx)
sem := NewSemaphore(10)</p><p>for i, task := range tasks {
i, task := i, task
g.Go(func() error {
sem.Acquire()
defer sem.Release()</p><pre class="brush:php;toolbar:false"><code>    data, err := doTask(task)
    results[i] = Result{Index: i, Data: data, Err: err}
    return err
})</code>

} g.Wait() // 此时 results 已按原顺序填充，可直接用

容易踩的坑：