Golang如何实现并发任务优先级_Golang任务优先级调度实践-Golang-PHP中文网

Golang如何实现并发任务优先级_Golang任务优先级调度实践

P粉602998670

发布： 2025-12-02 10:58:41

原创

930人浏览过

Go语言原生不支持goroutine优先级调度，需在应用层通过container/heap实现优先级队列。定义Task结构体包含Priority字段，数值越小优先级越高，利用最大堆或最小堆维护任务顺序，配合工作协程按优先级取出并执行任务，从而实现高优先级任务先执行的调度机制。

golang如何实现并发任务优先级_golang任务优先级调度实践

在Go语言中，原生的goroutine调度器并不直接支持任务优先级的概念。所有goroutine由调度器统一管理，按协作式调度机制运行，无法像操作系统线程那样设置高、低优先级执行。但在实际开发中，我们常常需要实现带优先级的任务调度，比如高优先级的消息通知要早于日志写入处理。这时候就需要在应用层设计优先级队列与调度逻辑。

使用带优先级的队列实现任务调度

最常见的方式是结合优先级队列和一个或多个工作协程（worker goroutines），通过比较任务的优先级来决定执行顺序。

可以使用最小堆或最大堆结构来维护任务队列。Go标准库container/heap提供了堆的操作接口，我们可以基于它构建一个优先级任务队列。

示例：定义带优先级的任务

type Task struct {
    Priority int // 数值越小，优先级越高
    Data     string
    Fn       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{}) {
    task := x.(*Task)
    *pq = append(*pq, task)
}

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

登录后复制

调度器控制任务分发

有了优先级队列后，需要一个调度器来安全地向队列中添加任务，并由工作协程取出执行。

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

Replit Ghostwrite

一种基于 ML 的工具，可提供代码完成、生成、转换和编辑器内搜索功能。

查看详情

通常使用互斥锁 + 条件变量（sync.Mutex 和 sync.Cond）或channel来协调生产者和消费者。

示例：启动调度器与工作协程

func NewScheduler() *Scheduler {
    pq := make(PriorityQueue, 0)
    heap.Init(&pq)

    return &Scheduler{
        tasks: &pq,
        mu:    new(sync.Mutex),
        cond:  sync.NewCond(&sync.Mutex{}),
    }
}

type Scheduler struct {
    tasks *PriorityQueue
    mu    *sync.Mutex
    cond  *sync.Cond
}

func (s *Scheduler) Submit(priority int, f func()) {
    s.mu.Lock()
    heap.Push(s.tasks, &Task{Priority: priority, Fn: f})
    s.mu.Unlock()
    s.cond.Signal() // 唤醒等待的工作协程
}

func (s *Scheduler) Run() {
    for {
        s.cond.L.Lock()
        for s.tasks.Len() == 0 {
            s.cond.Wait() // 队列为空时等待
        }

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

        s.cond.L.Unlock()

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

登录后复制

实际使用示例

启动调度器并提交不同优先级的任务：

func main() {
    scheduler := NewScheduler()

    // 启动调度器
    go scheduler.Run()

    // 提交任务
    scheduler.Submit(2, func() {
        fmt.Println("低优先级任务")
    })
    scheduler.Submit(1, func() {
        fmt.Println("高优先级任务")
    })
    scheduler.Submit(3, func() {
        fmt.Println("最低优先级任务")
    })

    // 简单延时让任务执行
    time.Sleep(time.Second)
}

登录后复制

输出结果会优先执行Priority=1的任务，然后是2，最后是3。