可以实现带优先级的worker pool。具体方法是：1. 使用多个通道分别接收不同优先级任务，worker优先从高优先级通道取任务；2. 定义任务结构体并设置high、medium、low三个优先级等级；3. 创建对应优先级的channel，并在worker中按优先级顺序尝试获取任务执行；4. 任务提交后关闭通道并等待完成，确保高优先级任务优先执行。还可进一步优化如使用heap实现统一优先级队列、支持动态调整优先级、添加超时控制等。
在实际开发中，很多场景需要对并发任务进行优先级调度。比如一个服务可能既要处理用户请求，也要执行后台计算任务，这时候我们希望高优先级的任务能够更快地被处理。
Golang本身通过goroutine和channel已经提供了非常强大的并发能力，但默认情况下goroutine的调度是无优先级区分的。那我们能不能自己实现一个带优先级的worker pool呢？答案是可以的，而且不复杂。
下面我们就来演示如何用Go语言实现一个带有任务优先级调度的worker pool。
立即学习“go语言免费学习笔记（深入）”；
1. 基本思路：优先级队列 + 多个工作池
要实现优先级调度，核心在于任务队列的排序机制。我们可以为每个任务设置一个优先级（比如整数，数值越小或越大代表优先级越高），然后让worker从队列中取出当前优先级最高的任务来执行。
常见做法：
- 使用多个通道（channel）分别接收不同优先级的任务
- 每个worker会按照优先级顺序去尝试获取任务
- 或者使用优先级队列结构（如heap.Interface实现的最小堆）
这里我们先采用第一种方式，便于理解与实现。
2. 定义任务结构和优先级等级
我们可以先定义一个任务结构体，并给它加上优先级字段：
type Task struct {
Priority int
Fn func()
}为了简化处理，我们设定三个优先级等级：High、Medium、Low，对应数字0、1、2。
const (
HighPriority = iota
MediumPriority
LowPriority
)然后为每个优先级创建一个独立的channel，worker优先从高优先级的channel中取任务。
3. 实现带优先级的Worker Pool
下面是具体的实现代码：
package main
import (
"fmt"
"sync"
)
const (
HighPriority = iota
MediumPriority
LowPriority
)
type Task struct {
Priority int
Fn func()
}
func NewWorkerPool(numWorkers int, taskChanSize int) {
highChan := make(chan Task, taskChanSize)
medChan := make(chan Task, taskChanSize)
lowChan := make(chan Task, taskChanSize)
var wg sync.WaitGroup
// 启动worker
for i := 0; i < numWorkers; i++ {
wg.Add(1)
go func() {
defer wg.Done()
for {
select {
case task, ok := <-highChan:
if !ok {
return
}
task.Fn()
case task, ok := <-medChan:
if !ok {
return
}
task.Fn()
case task, ok := <-lowChan:
if !ok {
return
}
task.Fn()
}
}
}()
}
// 模拟添加任务
tasks := []Task{
{Priority: HighPriority, Fn: func() { fmt.Println("High priority task") }},
{Priority: MediumPriority, Fn: func() { fmt.Println("Medium priority task") }},
{Priority: LowPriority, Fn: func() { fmt.Println("Low priority task") }},
}
for _, task := range tasks {
switch task.Priority {
case HighPriority:
highChan <- task
case MediumPriority:
medChan <- task
case LowPriority:
lowChan <- task
}
}
close(highChan)
close(medChan)
close(lowChan)
wg.Wait()
}
func main() {
NewWorkerPool(3, 10)
}这段代码做了以下几件事：
- 定义了三种优先级的任务
- 创建了三个任务通道，分别对应高、中、低优先级
- worker会按顺序尝试从这三个通道中取任务执行（先high，再medium，最后low）
- 所有任务提交完后关闭通道并等待完成
这样就能保证高优先级任务尽可能先执行。
4. 进阶优化建议
上面的实现是一个基本版本，如果想进一步增强功能，可以考虑以下几点：
- 使用heap实现统一的优先级队列，避免多个channel带来的管理成本
- 增加动态调整优先级的能力
- 添加任务超时控制、限流机制等
- 支持更多优先级等级
比如，使用
container/heap包来自定义一个优先级队列，可以让逻辑更集中，也更容易扩展。
基本上就这些。这种方式虽然简单，但实用性强，适合一些中小型项目中的优先级任务调度需求。只要掌握了这个模式，后续可以根据具体业务做灵活调整。
