Go 语言中实现动态任务队列的工作者池（支持递归任务分发）-Golang

本文介绍一种基于 `sync.waitgroup` 和非阻塞通道操作的优雅方案，用于构建可动态扩展、无死锁风险的 go 工作者池，特别适用于 url 抓取等可能递归生成新任务的场景。

在构建并发任务处理系统时，一个常见但棘手的需求是：任务本身可能动态产生新任务（即“递归式”或“衍生式”任务），例如网页爬虫中解析 HTML 后发现新的链接需继续抓取。此时，若简单使用固定缓冲区的通道 + 固定数量 goroutine，极易陷入死锁或资源空转——当所有 worker 都在等待新任务、而新任务又尚未被任何 worker 提交时，整个系统将停滞。

传统的“计数型协调”（如问题中用 working chan int 累加/减工作状态）不仅逻辑复杂、易出竞态，还依赖对 select 执行顺序的隐式假设，可维护性与正确性均较差。更优解是让任务生命周期本身驱动并发控制：每个任务明确关联一次 wg.Add(1) 与 wg.Done()，由 sync.WaitGroup 统一管理活跃任务总数，并通过非阻塞发送机制避免 worker 因通道满而永久阻塞。

以下是推荐实现的核心模式：

package main

import (
    "sync"
)

const workers = 4

type job struct {
    url string
    // 可扩展字段，如 headers、depth 等
}

func (j *job) do(enqueue func(job)) {
    // 模拟实际工作：抓取 URL、解析 HTML
    // ...

    // 示例：发现新 URL，递归入队
    if j.url == "https://example.com/root" {
        enqueue(job{url: "https://example.com/page1"})
        enqueue(job{url: "https://example.com/page2"})
    }
}

func main() {
    jobs := make(chan job, 1024) // 带缓冲通道，缓解突发入队压力
    var 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) // 每个新任务计入 WaitGroup
        select {
        case jobs <- j:
            // 成功入队，由某个 worker 处理
        default:
            // 通道暂满 → 当前 goroutine 直接执行（避免阻塞）
            j.do(enqueue)
            wg.Done()
        }
    }

    // 初始化任务：提交首批 URL
    initialJobs := []job{
        {url: "https://example.com/root"},
        {url: "https://example.com/start"},
    }
    for _, j := range initialJobs {
        enqueue(j)
    }

    // 等待所有任务（含递归生成的）完成
    wg.Wait()
    close(jobs) // 关闭通道，通知 workers 退出
}

关键设计要点说明

• sync.WaitGroup 作为唯一权威状态源：wg.Add(1) 在任务创建时立即调用，wg.Done() 在任务结束时调用。wg.Wait() 自然阻塞至所有任务（包括动态生成的子任务）完成，无需手动跟踪“空闲 worker 数量”，彻底规避竞态和逻辑漏洞。

• 非阻塞入队（select + default）是防死锁核心：当 jobs 通道已满时，不阻塞当前 goroutine，而是降级为同步执行该任务（j.do(enqueue)）。这确保了：

  • 即使所有 worker 全在 range jobs 中等待，只要还有未提交的任务（例如在 enqueue 调用栈中），系统仍能推进；
  • 不存在“全部 worker 等待，新任务无法入队”的僵局。

• 缓冲通道大小是性能调优项，非正确性依赖：make(chan job, 1024) 的缓冲区仅用于吞吐优化，不影响逻辑正确性。即使设为 0（无缓冲），default 分支仍能兜底执行，只是并发度略低。