Go channel 怎么实现有界并发的工作池

Go channel 做工作池时，真正控制“同时跑多少个任务”的不是缓冲区，而是固定启动的 worker 数量。一个可靠的有界并发结构通常是：生产者把任务送进 jobs，固定数量的 worker 从中取任务，生产者投递完后关闭 jobs，最后等所有 worker 完成，再关闭 results。

把 worker 数量当作并发上限，把 channel 容量当作等待区；关闭 channel 只由知道发送已经结束的一方负责。

一、先把并发上限和队列容量分开

假设任务是读取文件、调用接口或处理消息。直接为每个任务启动 goroutine，任务一多就会把内存、连接和下游服务一起推高。工作池的第一个取舍是固定 worker 数量，例如 3 个 worker；它决定同时执行的任务数。

make(chan Job, 4) 的 4 不是并发数，而是等待区容量。3 个 worker 都忙时，生产者最多还能先放入 4 个任务；再继续发送就会阻塞，形成反压。这个阻塞通常比无限创建 goroutine 更容易观察和治理。

二、让一个发送方负责关闭任务 channel

关闭责任要跟发送责任绑定。最简单的约定是：生产者是 jobs 的唯一发送方，所有任务发送完后执行 close(jobs)；worker 只接收，不关闭它。这样 worker 的退出条件很直观：range 读完队列，发现 channel 已关闭，就结束循环。

type Job struct {
	ID int
}

func produce(jobs chan

多个生产者同时发送时，不要让其中任意一个随意 close(jobs)，否则另一个生产者仍在发送就会触发 send on closed channel。这时应增加一个协调者，等所有生产者结束后再关闭任务 channel。

三、用固定 worker 和 WaitGroup 收口

worker 的数量是显式参数，而不是跟任务数一起增长。下面的 worker 只负责从 jobs 取任务并发送结果；它不关闭共享的 results，因为共享发送者不止一个。

type Result struct {
	ID  int
	Err error
}

func worker(jobs 

启动前先 wg.Add(workerCount)，再逐个启动 worker。不要在 worker 已经可能退出之后才补 Add；WaitGroup 的计数建立顺序必须早于对应的 Done。

四、结果 channel 只在全部 worker 退出后关闭

results 的关闭要晚于最后一次发送。一个专门的收口 goroutine 等待 WaitGroup 归零，再关闭 results；消费方就可以安全地使用 for result := range results。

func run(jobs 

五、用反例和清单判断是否真的有界

常见反例是“每来一个任务就 go process(job)，再额外放一个带缓冲的 channel”。这只能缓冲通知，不能限制 process goroutine 的数量。另一个反例是 worker 结束时关闭 results：第一个结束的 worker 关闭后，其他 worker 仍可能发送，程序就会崩溃。

如果业务还需要超时或主动取消，应让 worker 在取任务和发结果时同时监听 context.Done；取消只解决“何时停止”，并不会改变 channel 的关闭所有权。先把这条边界守住，工作池才不会从有界并发变成隐藏的 goroutine 堆积。

相关问题

jobs 一定要关闭吗？

如果 worker 用 range 读取，生产者完成后就必须关闭，否则 worker 会一直等待下一个任务。若改用 context 或显式停止信号，也可以用另一种退出协议，但责任仍要明确。

results 可以由消费者关闭吗？

不建议。消费者不知道还有多少 worker 会发送，关闭时机应由发送完成的一方协调。

缓冲区越大越好吗？

不是。过大的缓冲区会把压力藏起来，延迟和内存会一起增长；容量应根据任务成本、突发量和下游承受能力设定。