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

<font face="Courier New">
type Task struct {
    Priority int
    Job      func()
}

// 实现 heap.Interface 的 PriorityQueue
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
}

var taskQueue = make(chan *Task, 100)

// Worker 函数
func worker(wg *sync.WaitGroup) {
    defer wg.Done()
    for task := range taskQueue {
        task.Job()
    }
}

// 启动 worker 池
func StartWorkerPool(n int) {
    var wg sync.WaitGroup
    for i := 0; i < n; i++ {
        wg.Add(1)
        go worker(&wg)
    }
    wg.Wait()
}
</font>

<font face="Courier New">
type Scheduler struct {
    mu   sync.Mutex
    heap PriorityQueue
    cond *sync.Cond
}

func NewScheduler() *Scheduler {
    s := &Scheduler{}
    s.cond = sync.NewCond(&s.mu)
    return s
}

func (s *Scheduler) Push(task *Task) {
    s.mu.Lock()
    defer s.mu.Unlock()
    heap.Push(&s.heap, task)
    s.cond.Signal() // 唤醒等待的worker
}

func (s *Scheduler) Pop() *Task {
    s.mu.Lock()
    defer s.mu.Unlock()
    for s.heap.Len() == 0 {
        s.cond.Wait() // 阻塞等待任务
    }
    return heap.Pop(&s.heap).(*Task)
}
</font>

<font face="Courier New">
type Task struct {
    Priority int
    Ctx      context.Context
    Job      func(context.Context)
}
</font>