Go goroutine 怎样用 semaphore 限制并发

我第一次给批量任务加 goroutine 并发时，最先看到的不是速度变快，而是下游接口被一口气打满。后来我把任务拆成“拿到令牌才能工作”，再把令牌数量固定下来，问题就从“到底启动了多少 goroutine”变成了“同时拿到几个执行名额”。这正是 Go 里用带缓冲 channel 做轻量 semaphore 的价值。

官方地址：https://go.dev/doc/

结论是：用容量为 N 的带缓冲 channel 表示 N 个令牌，任务开始前发送一个令牌，任务结束时用 defer 接收并归还。再用 WaitGroup 等待已经启动的任务，便能同时约束并发数和收尾时机。

先把并发上限变成令牌容量

我会先把一个小项目的边界定死：假设批量任务最多允许 3 个工作同时执行，任务本身可以是抓取、文件处理或外部接口调用，但本文只关心并发上限和令牌回收。Go 规范把 channel 定义为并发函数之间发送和接收指定类型值的机制；带缓冲 channel 有容量，发送会在有空间时继续，因此它很适合承载有限数量的空令牌。

// Semaphore 用带缓冲 channel 保存有限数量的执行令牌。
type Semaphore chan struct{}

// NewSemaphore 创建最多允许 limit 个任务同时进入的令牌池。
func NewSemaphore(limit int) Semaphore {
	// limit 必须为正数，否则调用方无法得到有意义的并发上限。
	if limit 

这里不需要给每个任务分配一个复杂对象，空结构体只表达“占用一个名额”。当 channel 已经有 3 个令牌时，第四个任务的发送会等待；前面的任务释放后，等待者才有机会继续。这个等待是阻塞式的，适合“必须排队，不能丢任务”的批处理。

把获取和释放放进同一套生命周期

真正容易出问题的地方不是创建 channel，而是任务中途返回。比如工作函数发现输入为空直接 return，或者调用下游失败后提前结束；如果令牌只在正常路径释放，运行几轮以后就会出现“还有任务，但所有 goroutine 都卡在 Acquire”的假象。我自己的习惯是：Acquire 成功的下一行立刻安排 defer Release。

// RunOne 处理一个任务，并保证所有退出路径都归还令牌。
func RunOne(s Semaphore, job func() error) error {
	s.Acquire()
	// 只要获取成功，后续 return 也必须触发归还。
	defer s.Release()

	if job == nil {
		// 空任务属于调用错误，但不能带走一个令牌。
		return fmt.Errorf("nil job")
	}
	return job()
}

这个写法有一个很重要的边界：不能在 Acquire 之前就 defer Release，因为那样会释放一个从未拿到的令牌，最终把 semaphore 的状态弄乱。也不要把 Release 藏在某个只会执行一次的成功回调里；任务函数的所有返回路径都应该经过同一个 defer。

把等待收尾接到任务计数

semaphore 只限制“同时有多少任务进入工作区”，它不负责告诉主流程“所有任务是否已经结束”。这件事交给 sync.WaitGroup 更清楚：提交 goroutine 前增加计数，goroutine 结束时 Done，最后 Wait。官方文档也强调，计数归零时 Wait 才会解除；因此 Add 的位置必须和启动任务的边界一致。

// RunBatch 提交一批任务，限制同时运行的任务数并等待全部收尾。
func RunBatch(jobs []func() error, limit int) error {
	sem := NewSemaphore(limit)
	var wg sync.WaitGroup
	var firstErr error
	var errMu sync.Mutex

	for _, job := range jobs {
		// Add 放在 go 之前，避免主 goroutine 过早进入 Wait。
		wg.Add(1)
		go func(job func() error) {
			// Done 和 Release 都必须覆盖正常返回与错误返回。
			defer wg.Done()
			if err := RunOne(sem, job); err != nil {
				// 多个 goroutine 可能同时写错误，使用锁保护共享状态。
				errMu.Lock()
				if firstErr == nil {
					firstErr = err
				}
				errMu.Unlock()
			}
		}(job)
	}

	// Wait 只负责等待已经登记的任务，令牌仍由每个任务自己归还。
	wg.Wait()
	return firstErr
}

这段结构里有两个独立的不变量：WaitGroup 计数最终必须归零，semaphore 的每次成功 Acquire 必须对应一次 Release。只看其中一个是不够的；Wait 可能正常返回，但如果令牌泄漏，下一批任务仍会永久等待。

处理无法获取令牌的两种策略

阻塞 Acquire 很简单，但它意味着调用方愿意一直排队。如果任务来自请求处理、定时器或可以取消的上游，我通常会让获取动作也接受 context：拿不到令牌时，先判断取消信号；这样不会因为下游已经放弃而继续占着 goroutine。

// AcquireContext 在令牌可用或 context 取消时返回。
func (s Semaphore) AcquireContext(ctx context.Context) error {
	select {
	case s 

这里要分清两条路径：select 走到发送分支，才算成功获取令牌；走到 ctx.Done 分支时，任务根本没有进入 semaphore，不能释放。若业务选择“排队但不取消”，就使用阻塞 Acquire；若业务有超时和取消，就使用可取消版本，并把错误交给上层决定是否重试。

上线前我会检查哪些泄漏边界

我会按下面的清单复查，而不是只看 limit 写成了多少：

特别是 nil channel：Go 规范明确说明，对 nil channel 的接收和发送都会永久阻塞。所以 NewSemaphore 不应该静默返回 nil；如果并发上限是配置项，启动阶段就应拒绝非正数，把错误留在可读的配置边界上。

相关问题

semaphore 会减少 goroutine 的创建数量吗？

不会。上面的 RunBatch 仍然可以为每个 job 启动 goroutine，semaphore 只限制同时进入工作区的数量。如果任务总量很大，还应进一步考虑固定 worker 数量或有界任务队列。

为什么不用 time.Sleep 控制并发？

Sleep 只能延迟下一次动作，不能表达“某个任务已经释放了一个名额”。带缓冲 channel 直接把名额变成可发送和接收的资源，边界更明确。

WaitGroup 能替代 semaphore 吗？

不能。WaitGroup 记录任务是否完成，semaphore 约束同时占用的执行名额；一个负责收尾，一个负责限流，通常需要配合使用。

对我来说，最稳的实现不是把 semaphore 封装得多复杂，而是坚持三件事：令牌容量就是并发上限，成功获取后立刻 defer 释放，所有已启动任务都登记到 WaitGroup。这样读代码时，限制、释放和收尾各自有清楚的落点。