不能直接用 sync.waitgroup 控制并发度,因为它仅用于等待 goroutine 完成,不提供并发数量限制;若启动 1000 个 goroutine,就会真正创建 1000 个,导致内存暴涨和调度压力过大。

为什么不能直接用 sync.WaitGroup 控制并发度
因为 sync.WaitGroup 只负责等待,不提供并发数限制能力。你往里塞 1000 个 goroutine,它就真开 1000 个——哪怕你只想同时跑 5 个。常见错误是先 wg.Add(n) 再启动所有 goroutine,结果内存暴涨、调度压力大、甚至触发 runtime: failed to create new OS thread 错误。
semaphore 是最轻量可靠的限流手段
Go 官方没内置信号量,但用 chan struct{} 实现一个非常简单且无锁(相比 sync.Mutex):
func newSemaphore(n int) chan struct{} {
sem := make(chan struct{}, n)
for i := 0; i <p>使用时:<code>sem 获取许可,<code> 归还。注意必须配对,漏归还会导致后续任务永久阻塞。</code></code></p>
- 并发度为 0?别传 0,
make(chan struct{}, 0)是无缓冲 channel,会立刻阻塞——应提前校验n > 0 - 并发度设得过大(比如 >500)?可能触发 goroutine 调度延迟,实测在 CPU 密集型任务中,超过 128 后吞吐提升极小
- 要不要用
context.Context?如果批处理可能被取消或超时,必须传入并在每个任务中检查ctx.Done(),否则sem可能永远卡住
批处理函数要支持动态调整并发度
静态的 sem 创建后无法扩容缩容,所以得把信号量封装进结构体,并暴露修改接口:
type BatchExecutor struct {
sem chan struct{}
mu sync.RWMutex
}
<p>func (b *BatchExecutor) SetConcurrency(n int) {
b.mu.Lock()
defer b.mu.Unlock()
// 关闭旧 channel,新建并预填充
close(b.sem)
b.sem = make(chan struct{}, n)
for i := 0; i </p><p>关键点:</p><div class="aritcle_card flexRow artxards">
<div class="artcardd flexRow">
<a class="aritcle_card_img" rel="nofollow" href="/ai/3625" title="Humva"><img
src="https://img.php.cn/upload/ai_manual/001/246/273/178599559248874.png" alt="Humva" onerror="this.onerror='';this.src='/static/lhimages/moren/morentu.png'" ></a>
<div class="aritcle_card_info flexColumn">
<a rel="nofollow" href="/ai/3625" title="Humva" class="overflowclass">Humva</a>
<p class="overflowclass">Humva是一款AI文本写作工具,AI数字人生成工具,自定义创建专属数字人。</p>
</div>
<a rel="nofollow" href="/ai/3625" title="Humva" class="aritcle_card_btn flexRow flexcenter"><b></b><span>下载</span>
</a>
</div>
</div>
- 必须用
sync.RWMutex保护sem字段,因为SetConcurrency和任务执行会并发访问 - 不能只改缓冲区大小——
cap(b.sem)是只读的,必须重建 channel - 旧 channel 关闭后,正在等待的 goroutine 会 panic:「send on closed channel」,所以要在
SetConcurrency前确保没有新任务进入,或加额外协调逻辑(比如用原子计数器 wait all in-flight)
实际调用时如何避免 goroutine 泄漏和 panic
典型错误写法:go func() { ... }() 没捕获 panic,或忘记归还 sem。正确姿势是封装一层带恢复和清理的执行器:
func (b *BatchExecutor) Do(ctx context.Context, f func() error) error {
select {
case <code>defer func() {
if r := recover(); r != nil {
// 记录 panic,但不传播
}
}()
return f()</code>
}
注意:
defer 必须在获取成功后立即注册,否则 panic 会导致信号量泄露- 不要在
f()里直接调用cancel()或关闭ctx,这会影响其他并行任务 - 如果
f()本身耗时很长,建议内部也定期检查ctx.Err(),而不是等整个函数返回才响应取消
动态并发控制真正难的不是实现,而是当多个协程同时调用 SetConcurrency 并发修改时,如何保证语义一致——比如缩容时正在运行的任务是否该被中断,这个边界必须由业务层明确约定,库本身没法替你决定。










