Golang带优先级的worker pool实现方法

可以实现带优先级的worker pool。具体方法是：1. 使用多个通道分别接收不同优先级任务，worker优先从高优先级通道取任务；2. 定义任务结构体并设置High、Medium、Low三个优先级等级；3. 创建对应优先级的channel，并在worker中按优先级顺序尝试获取任务执行；4. 任务提交后关闭通道并等待完成，确保高优先级任务优先执行。还可进一步优化如使用heap实现统一优先级队列、支持动态调整优先级、添加超时控制等。

在实际开发中，很多场景需要对并发任务进行优先级调度。比如一个服务可能既要处理用户请求，也要执行后台计算任务，这时候我们希望高优先级的任务能够更快地被处理。

Golang本身通过goroutine和channel已经提供了非常强大的并发能力，但默认情况下goroutine的调度是无优先级区分的。那我们能不能自己实现一个带优先级的worker pool呢？答案是可以的，而且不复杂。

下面我们就来演示如何用Go语言实现一个带有任务优先级调度的worker pool。

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包来自定义一个优先级队列，可以让逻辑更集中，也更容易扩展。

基本上就这些。这种方式虽然简单，但实用性强，适合一些中小型项目中的优先级任务调度需求。只要掌握了这个模式，后续可以根据具体业务做灵活调整。