Golang如何实现并发任务优先级调度演示带优先级的worker pool-Golang-PHP中文网

Golang如何实现并发任务优先级调度演示带优先级的worker pool

P粉602998670

发布： 2025-08-22 11:30:02

原创

825人浏览过

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

Golang如何实现并发任务优先级调度演示带优先级的worker pool

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

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

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

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

1. 基本思路：优先级队列 + 多个工作池

要实现优先级调度，核心在于任务队列的排序机制。我们可以为每个任务设置一个优先级（比如整数，数值越小或越大代表优先级越高），然后让worker从队列中取出当前优先级最高的任务来执行。

常见做法：

使用多个通道（channel）分别接收不同优先级的任务
每个worker会按照优先级顺序去尝试获取任务
或者使用优先级队列结构（如heap.Interface实现的最小堆）

这里我们先采用第一种方式，便于理解与实现。

2. 定义任务结构和优先级等级

我们可以先定义一个任务结构体，并给它加上优先级字段：

MindShow

MindShow官网 | AI生成PPT，快速演示你的想法

1492

查看详情

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)
}

登录后复制

这段代码做了以下几件事：