本文介绍一种基于 `sync.waitgroup` 和非阻塞通道发送的 go 工作池模式,用于处理可递归生成新任务的场景(如网页爬虫),避免死锁与竞态,确保所有 worker 协同启停。
在构建并发任务系统时,一个常见但棘手的需求是:任务本身可能动态产生新任务(例如解析网页后发现更多待抓取 URL),而工作协程(worker)需在“无任务可做且无人生产新任务”时自动退出,实现整体生命周期的可控收敛。传统固定缓冲队列 + 计数器方式(如问题中所示)易引入竞态、逻辑耦合度高,且难以准确判断“全局空闲”状态。
更健壮的解法是将任务生命周期与执行权绑定:用 sync.WaitGroup 统一追踪待处理任务总数,而非依赖 worker 状态轮询;同时,通过非阻塞 select 实现“投递优先,本地兜底”的任务分发策略——既避免 channel 写入阻塞导致死锁,又消除对预设缓冲区大小的强依赖。
以下是核心实现:
package main
import (
"sync"
)
const workers = 4
type job struct {
url string
}
func (j *job) do(enqueue func(job)) {
// 模拟实际工作:下载并解析页面
// 若发现新 URL,则调用 enqueue 递归提交
// 示例:
// if j.url == "https://example.com" {
// enqueue(job{url: "https://example.com/page1"})
// enqueue(job{url: "https://example.com/page2"})
// }
}
func main() {
jobs := make(chan job, 100) // 可选小缓冲,缓解突发投递压力
wg := &sync.WaitGroup{}
var enqueue func(job)
// 启动 worker 池
for i := 0; i < workers; i++ {
go func() {
for j := range jobs {
j.do(enqueue)
wg.Done()
}
}()
}
// 安全的任务入队函数(闭包捕获 wg 和 jobs)
enqueue = func(j job) {
wg.Add(1) // 声明:即将新增一个待完成任务
select {
case jobs <- j:
// 成功投递至通道,由空闲 worker 处理
default:
// 通道满或无空闲 worker → 当前 goroutine 同步执行(递归兜底)
j.do(enqueue)
wg.Done()
}
}
// 提交初始任务
initialJobs := []job{
{url: "https://example.com"},
{url: "https://golang.org"},
}
for _, j := range initialJobs {
enqueue(j)
}
// 等待所有任务完成(包括递归生成的)
wg.Wait()
close(jobs) // 关闭通道,通知 workers 退出
}
✅ 关键设计优势: