Go 實戰單佇列到優先順序佇列實現圖文範例

type JobQueue struct {
    mu sync.Mutex //佇列的操作需要並行安全
    jobList *list.List //List是golang庫的雙向佇列實現，每個元素都是一個job
    noticeChan chan struct{} //入隊一個job就往該channel中放入一個訊息，以供消費者消費
}

/**
 * 佇列的Push操作
 */
func (queue *JobQueue) PushJob(job Job) {
    queue.jobList.PushBack(job) //將job加到隊尾
    queue.noticeChan <- struct{}{}
}

queue.noticeChan <- struct{}{}

/**
 * 彈出佇列的第一個元素
 */
func (queue *JobQueue) PopJob() Job {
    queue.mu.Lock()
    defer queue.mu.Unlock()
    /**
     * 說明在佇列中沒有元素了
     */
    if queue.jobList.Len() == 0 {
        return nil
    }
    elements := queue.jobList.Front() //獲取隊裡的第一個元素
    return queue.jobList.Remove(elements).(Job) //將元素從佇列中移除並返回
}

func (queue *JobQueue) WaitJob() <-chan struct{} {
    return queue.noticeChan
}

type BaseJob struct {
    Err error
    DoneChan chan struct{} //當作業完成時，或者作業被取消時，通知呼叫者
    Ctx context.Context
    cancelFunc context.CancelFunc
}

/**
 * 作業執行完畢，關閉DoneChan，所有監聽DoneChan的接收者都能收到關閉的訊號
 */
func (job *BaseJob) Done() {
    close(job.DoneChan)
}

/**
 * 等待job執行完成
 */
func (job *BaseJob) WaitDone()  {
    select {
    case <-job.DoneChan:
        return
    }
}

type SquareJob struct {
    *BaseJob
    x int
}

func (s *SquareJob) Execute() error {
    result := s.x * s.x
    fmt.Println("the result is ", result)
    return nil
}

type WorkerManager struct {
    queue *JobQueue
    closeChan chan struct{}
}

 func (m *WorkerManager) StartWork() error {
    fmt.Println("Start to Work")
    for {
        select {
            case <-m.closeChan:
                return nil
            case <-m.queue.noticeChan:
                job := m.queue.PopJob()
                m.ConsumeJob(job)
        }
    }
    return nil
}
func (m *WorkerManager) ConsumeJob(job Job) {
    defer func() {
        job.Done()
    }()
    job.Execute()
}

func main() {
    //初始化一個佇列
    queue := &JobQueue{
        jobList: list.New(),
        noticeChan: make(chan struct{}, 10),
    }
    //初始化一個消費worker
    workerManger := NewWorkerManager(queue)
    // worker開始監聽佇列
    go workerManger.StartWork()
    // 構造SquareJob
    job := &SquareJob{
        BaseJob: &BaseJob{
            DoneChan: make(chan struct{}, 1),
        },
        x: 5,
    }
    //壓入佇列尾部
    queue.PushJob(job)
    //等待job執行完成
    job.WaitDone()
    print("The End")
}

type BaseJob struct {
    Err error
    DoneChan chan struct{} //當作業完成時，或者作業被取消時，通知呼叫者
    Ctx context.Context
    cancelFunc context.CancelFunc
    priority int //工作單元的優先順序
}

var queues = make([]*JobQueue, 10, 100)

var priorityIdx map[int][int]//該map的key是優先順序，value代表的是queues切片的索引

type PriorityQueue struct {
    mu sync.Mutex
    noticeChan chan struct{}
    queues []*JobQueue
    priorityIdx map[int]int
}
//原來的JobQueue會變成如下這樣：
type JobQueue struct {
    priority int //代表該佇列是哪種優先順序的佇列
    jobList *list.List //List是golang庫的雙向佇列實現，每個元素都是一個job
}

func (priorityQueue *PriorityQueue) Push(job Job) {
    priorityQueue.mu.Lock()
    defer priorityQueue.mu.Unlock()
    //先根據job的優先順序找要入隊的佇列
    var idx int
    var ok bool
    //從優先順序-切片索引的map中查詢該優先順序的佇列是否存在
    if idx, ok = priorityQueue.priorityIdx[job.Priority()]; !ok {
        //如果不存在該優先順序的佇列，則需要初始化一個佇列，並返回該佇列在切片中的索引位置
        idx = priorityQueue.addPriorityQueue(job.Priority)
    }
    //根據獲取到的切片索引idx，找到具體的佇列
    queue := priority.queues[idx]
    //將job推播到佇列的隊尾
    queue.JobList.PushBack(job)
    //佇列job個數+1
    priorityQueue.Size++
    //如果佇列job個數超過佇列的最大容量，則從優先順序最低的佇列中移除工作單元
    if priorityQueue.size > priorityQueue.capacity {
        priorityQueue.RemoveLeastPriorityJob()
    }else {
        //通知新進來一個job
        priorityQueue.noticeChan <- struct{}{}
    }
}

func (priorityQueue *PriorityQueue) addPriorityQueue(priority int) int {
    n := len(priorityQueue.queues)
    //通過二分查詢找到priority應插入的切片索引
    pos := sort.Search(n, func(i int) bool {
        return priority < priorityQueue.priority
    })
    //更新對映表中優先順序和切片索引的對應關係
    for i := pos; i < n; i++ {
        priorityQueue.priorityIdx[priorityQueue.queues[i].priority] = i + 1
    }
    tail := make([]*jobQueue, n-pos)
    copy(tail, priorityQueue.queues[pos:])
    //初始化一個新的優先順序佇列，並將該元素放到切片的pos位置中
    priorityQueue.queues = append(priorityQueue.queues[0:pos], newJobQueue(priority))
    //將高於priority優先順序的元素也拼接到切片後面
    priorityQueue.queues = append(priorityQueue.queues, tail...) 
    return pos
}

func main() {
    //初始化一個佇列
    queue := &PriorityQueue{
        noticeChan: make(chan struct{}, cap),
        capacity: cap,
        priorityIdx: make(map[int]int),
        size: 0,
    }
    //初始化一個消費worker
    workerManger := NewWorkerManager(queue)
    // worker開始監聽佇列
    go workerManger.StartWork()
    // 構造SquareJob
    job := &SquareJob{
        BaseJob: &BaseJob{
            DoneChan: make(chan struct{}, 1),
        },
        x: 5,
        priority: 10,
    }
    //壓入佇列尾部
    queue.PushJob(job)
    //等待job執行完成
    job.WaitDone()
    print("The End")
}

type WorkerManager struct {
    queue *PriorityQueue
    closeChans []chan struct{}
}

 func (m *WorkerManager) StartWork(n int) error {
    fmt.Println("Start to Work")
    for i := 0; i < n; i++ {
        m.createWorker();
    }
    return nil
}
func (m *WorkerManager) createWorker() {
    closeChan := make(chan struct{})
    //每個協程，就是一個worker
    go func(closeChan chan struct{}) {
        var job Job
        for {
                select {
                    case <-m.closeChan:
                        return nil
                    case <-m.queue.noticeChan:
                        job := m.queue.PopJob()
                        m.ConsumeJob(job)
                }   
        }
    }(closeChan)
    m.closeChanMu.Lock()
    defer m.closeChanMu.Unlock()
    m.closeChans = append(m.closeChans, closeChan)
    return nil
}
func (m *WorkerManager) ConsumeJob(job Job) {
    defer func() {
        job.Done()
    }()
    job.Execute()
}