Go worker pool 的优雅实现：动态扩缩容与任务优先级队列

写 Go 并发程序的人，几乎都会在某个阶段写出一个 worker pool。任务塞进 channel，一组 goroutine 在另一端消费，简单直接，也够用。但当这个 pool 被放到真实生产环境，问题就来了：任务量不再是平稳的，任务的重要程度也不再一致。固定数量的 worker 池，在流量高峰时任务排队堆积，在低谷时又白白占着资源；更尴尬的是，突然插入的少量紧急任务，只能和普通任务一起排在队尾。

这篇文章想聊的是 worker pool 的进阶形态：如何做到动态扩缩容，以及如何引入任务优先级队列。这里没有能直接复制到生产环境的完整库，但有更重要的东西——把关键机制和取舍讲清楚。

一、固定 worker pool：简单，但不总是够用

先看一个最常见的基础版本。任务是一个结构体，worker 从 jobs channel 里取任务执行：

func WorkerPool(jobs <-chan Job, wg *sync.WaitGroup) {
    defer wg.Done()
    for job := range jobs {
        job.Run()
    }
}

这个模式的价值在于，用有限数量的 goroutine 处理无限量的任务，避免为每个任务单独开一个 goroutine。channel 本身是并发安全的队列，worker 之间天然互斥，实现成本极低。不过，它背后有三个隐藏假设：任务到达率均匀、任务执行时间波动不大、所有任务优先级相同。这三个条件一旦不成立，固定池就开始捉襟见肘。

比如一个支付通知回调服务，白天每秒几千次回调，晚上可能只有几十次。把池子固定为 100 个 worker，白天依然会有几万任务积压，晚上大多数时间空转。你当然可以把 worker 数调大，但调大后内存和栈开销也会跟着上来。这时候，你会希望 worker 数量能随负载自动变化。

再比如同一个服务里，既有点击日志这种低实时性任务，也有用户触发的即时通知。如果都放在一个队列里，即时通知只能等前面几千个日志任务处理完。这种场景下，任务优先级队列就变成了硬需求。

二、动态扩缩容：关键在于安全地缩容

动态扩缩容，听起来是按负载调整 worker 数量，但 Go 里面真正的难点不在扩容，而在缩容。扩容可以加一个新 worker 就完事，缩容却没法直接终止 goroutine。Go 没有提供 kill 机制，你也不能中断正在执行的业务逻辑。安全缩容的前提是，让 worker 在处理完当前任务后，自己退出。

最常见的做法是用 context。调度器持有每个 worker 的 cancel 函数，需要缩容时调用 cancel，worker 在下一次 select 时收到 ctx.Done()，然后退出。代码可以写成这样：

func startWorker(ctx context.Context, jobs <-chan Job, wg *sync.WaitGroup) {
    defer wg.Done()
    for {
        select {
        case job := <-jobs:
            job.Run() // 当前任务不会因为 ctx 取消而中断
        case <-ctx.Done():
            return
        }
    }
}

注意：这里的 ctx 只用于控制 worker 的循环，不应该直接传给业务函数，否则任务可能被取消到一半。

扩缩容的时机与阈值

扩缩容的时机，常见的策略是基于队列长度，或者基于 worker 利用率。基于队列长度比较直观：当待处理任务数超过阈值时扩容，空闲时间超过阈值时缩容。基于利用率则需要为 worker 维护状态，统计忙闲比，要求更高。

这里有一个很常见的问题：阈值设置不好，worker 数量会震荡。任务量在临界点附近波动时，队列一会儿超过扩容线，一会儿低于缩容线，每次都会触发调整。goroutine 创建虽然便宜，但频繁创建销毁会放大系统的不稳定性。

建议引入滞回区，例如队列长度超过 2000 时扩容，低于 300 时才缩容；每次调整后至少等 10 秒再观察。同时为 worker 数量设置明确的上下界。

扩容条件：队列长度超过 maxQueueSize，或 worker 平均利用率持续超过 80%
缩容条件：队列长度低于 minQueueSize，且 worker 空闲时间持续超过 idleTimeout
边界约束：始终保留 minWorkers，且不超过 maxWorkers

三、任务优先级队列：select 不是银弹

动态扩缩容解决了资源伸缩问题，但没有解决任务乱序问题。紧急任务插不进来，依然排在队尾。很多团队的第一反应是改用 select 同时监听多个 channel：

select {
case job := <-highCh:
    job.Run()
case job := <-lowCh:
    job.Run()
}

表面看高优先级 channel 好像会被优先处理，但 select 在多个 case 同时就绪时是随机选择，低优先级任务有接近 50% 的概率被优先执行。而且，如果你有十个优先级级别，就需要十个 channel，代码完全不可维护。

更合理的方式是把任务存入优先级队列。Go 标准库 container/heap 提供了最小堆接口，通过自定义 Less 可以让优先级高的先出队。下面是一个常见的 Job 结构定义和堆接口实现。

type Job struct {
    Priority int
    Payload  any
}

type PriorityQueue []*Job

func (pq PriorityQueue) Len() int { return len(pq) }
func (pq PriorityQueue) Less(i, j int) bool {
    return pq[i].Priority > pq[j].Priority
}
func (pq PriorityQueue) Swap(i, j int) { pq[i], pq[j] = pq[j], pq[i] }

func (pq *PriorityQueue) Push(x any) { *pq = append(*pq, x.(*Job)) }
func (pq *PriorityQueue) Pop() any {
    old := *pq
    n := len(old)
    item := old[n-1]
    old[n-1] = nil
    *pq = old[:n-1]
    return item
}

这段代码是 container/heap 的标准实现，入队出队复杂度都是 O(log n)，性能足够。但要记住，heap 本身不是并发安全的，所以在你的 pool 里，Push 和 Pop 都要用互斥锁保护。

优先级队列的“饿死”问题

当高优先级任务不断涌入时，低优先级任务可能永远得不到执行，这就是“饿死”。两种常见解法：让步策略，每执行 N 个高优先级任务后强制执行一个低优先级任务；老化策略，任务等待超过一定时间后自动提升优先级。前者会增加高优先级任务的尾部延迟，后者需要维护时间戳和扫描逻辑。没有银弹，只能结合业务取舍。

四、把动态 worker 和优先级队列组合起来

现在把两部分拼到一起。一个可用的结构大概是：

type Pool struct {
    mu     sync.Mutex
    queue  *PriorityQueue
    min    int
    max    int
    idle   time.Duration
    ctx    context.Context
    cancel context.CancelFunc
    wg     sync.WaitGroup
}

worker 循环不再是简单读 channel，而是从优先级队列中弹出一个任务执行。由于队列可能为空，还需要条件变量或 channel 通知 worker 继续等待。核心逻辑类似：

func (p *Pool) worker() {
    defer p.wg.Done()
    for {
        p.mu.Lock()
        if p.queue.Len() == 0 {
            p.mu.Unlock()
            // 等待任务或关闭信号
            return
        }
        job := heap.Pop(p.queue).(*Job)
        p.mu.Unlock()
        job.Run()
    }
}

这个版本的 worker 在队列为空时会直接退出，真实实现需要结合条件变量让 worker 阻塞等待。这里只表达结构。

这种组合带来的收益是任务存储、任务调度和 worker 生命周期之间的解耦。你可以替换队列实现、调整优先级策略，而 worker 代码基本不用动。代价是锁、条件变量、context 关闭路径的组合复杂度明显上升，需要谨慎设计优雅关闭流程。

五、动态 worker pool 的几个常见坑

这里写三个我在实现和 code review 中经常遇到的坑。

用 close(channel) 来退出 worker。channel 关闭后写入会 panic，而且你很难保证任务 producer 已经全部停止。更安全的方式是使用独立的 quit channel 或 context，并先停止任务投递。
把 cancel context 传给了任务函数。worker 的 context 只用于控制 worker 生命周期，如果任务函数接收了它并响应取消，业务状态可能被中途打断。除非任务显式支持取消，否则不要传。
缩容判断只看队列长度。任务执行时间波动大时，队列短不代表 worker 空闲。需要结合 worker 繁忙状态，否则会出现缩容后马上又扩容的震荡。

六、方案选型与落地建议

不同业务阶段适合不同复杂度。我整理了一个对比表，方便评估。

方案	扩缩容	优先级	实现成本	适用场景
固定 worker + channel	无	无	低	任务均匀、负载稳定
固定 worker + 多 channel	无	两级	低	只需要区分高低优先级
动态 worker + channel	支持	无	中	负载波动明显
动态 worker + 优先级堆	支持	多级	高	负载波动 + 任务分级