如何在 Golang 中构建任务优先级调度器_Golang 并发优先队列实现方案-Golang-PHP中文网

如何在 Golang 中构建任务优先级调度器_Golang 并发优先队列实现方案

P粉602998670

发布： 2025-11-16 19:11:02

原创

879人浏览过

答案：Golang中通过container/heap实现优先队列，结合sync.Mutex和sync.Cond保证并发安全，使用goroutine执行任务，构建高效优先级调度器，适用于消息系统等场景。

如何在 golang 中构建任务优先级调度器_golang 并发优先队列实现方案

在 Golang 中构建任务优先级调度器，核心是实现一个并发安全的优先队列，并结合 goroutine 和 channel 进行任务调度。优先级调度器适用于需要按紧急程度或重要性处理任务的场景，比如消息系统、作业调度平台、后台任务处理等。

优先队列的数据结构设计

优先队列通常基于最小堆或最大堆实现。Golang 标准库 container/heap 提供了堆操作接口，我们可以结合它来构建带优先级的任务队列。

定义任务结构体，包含执行函数和优先级字段：

type Task struct {
    Priority int
    ExecFunc func()
}

// 任务切片，实现 heap.Interface
type PriorityQueue []*Task

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 interface{}) {
    *pq = append(*pq, x.(*Task))
}

func (pq *PriorityQueue) Pop() interface{} {
    old := *pq
    n := len(old)
    item := old[n-1]
    *pq = old[0 : n-1]
    return item
}

登录后复制

并发安全的调度器封装

多个 goroutine 可能同时添加任务或调度执行，因此需要用互斥锁保护优先队列。调度器启动固定数量的工作协程，从队列中取出最高优先级任务执行。

立即学习“go语言免费学习笔记（深入）”；

先见AI

数据为基，先见未见

查看详情

type Scheduler struct {
    tasks   *PriorityQueue
    lock    sync.Mutex
    cond    *sync.Cond
    running bool
}

func NewScheduler() *Scheduler {
    pq := &PriorityQueue{}
    heap.Init(pq)
    s := &Scheduler{
        tasks:   pq,
        running: false,
    }
    s.cond = sync.NewCond(&s.lock)
    return s
}

func (s *Scheduler) AddTask(priority int, exec func()) {
    s.lock.Lock()
    defer s.lock.Unlock()
    heap.Push(s.tasks, &Task{Priority: priority, ExecFunc: exec})
    s.cond.Signal() // 唤醒等待的 worker
}

登录后复制

调度器启动后，worker 循环等待新任务：

func (s *Scheduler) Start(workers int) {
    s.running = true
    for i := 0; i < workers; i++ {
        go s.worker()
    }
}

func (s *Scheduler) worker() {
    for {
        s.lock.Lock()
        for s.tasks.Len() == 0 && s.running {
            s.cond.Wait() // 阻塞等待新任务
        }
        
        if !s.running && s.tasks.Len() == 0 {
            s.lock.Unlock()
            return
        }

        task := heap.Pop(s.tasks).(*Task)
        s.lock.Unlock()

        task.ExecFunc() // 执行任务
    }
}

func (s *Scheduler) Stop() {
    s.lock.Lock()
    s.running = false
    s.cond.Broadcast() // 唤醒所有 worker 退出
    s.lock.Unlock()
}

登录后复制

使用示例与性能考量

以下是一个简单使用示例，提交不同优先级的任务并观察执行顺序：

func main() {
    scheduler := NewScheduler()
    defer scheduler.Stop()

    scheduler.Start(2)

    // 提交任务：优先级越小，越早执行
    scheduler.AddTask(3, func() { fmt.Println("Low priority task") })
    scheduler.AddTask(1, func() { fmt.Println("High priority task") })
    scheduler.AddTask(2, func() { fmt.Println("Medium priority task") })

    time.Sleep(time.Second)
}

登录后复制

输出预期：

High priority task
Medium priority task
Low priority task

性能优化建议：