日本免费高清视频-国产福利视频导航-黄色在线播放国产-天天操天天操天天操天天操|www.shdianci.com

學無先后，達者為師

網(wǎng)站首頁 編程語言 正文

golang優(yōu)先級隊列的實現(xiàn)全過程_Golang

作者：游魚的編程旅行 ? 更新時間： 2023-01-31 編程語言

前言

在數(shù)據(jù)結(jié)構中，隊列遵循著FIFO（先進先出）的規(guī)則。在此基礎上，人們引申出了“優(yōu)先級隊列”的概念。

優(yōu)先級隊列，是帶有優(yōu)先級屬性的隊列，所有的隊列元素按照優(yōu)先級進行排序，消費者會先對優(yōu)先級高的隊列元素進行處理。

優(yōu)先級隊列的使用場景也是非常多的。比如，作業(yè)調(diào)度系統(tǒng)，當一個作業(yè)完成后，需要從剩下的作業(yè)中取出優(yōu)先級最高的作業(yè)進行處理。又比如，一個商城的用戶分為普通用戶和vip用戶，vip用戶更容易搶到那些秒殺商品。

在本文中，我將和大家一起探討，golang優(yōu)先級隊列的一種實現(xiàn)方案。

你可以收獲

  • golang切片特性
  • golang map特性
  • golang并發(fā)場景下的解決方案
  • golang優(yōu)先級隊列的實現(xiàn)思路

正文

內(nèi)容脈絡

為了讓大家腦海里有個大致的輪廓，我先把正文的大綱展示出來。

基礎知識

在正式開始“優(yōu)先級隊列”這個話題之前，我們首先要明確以下的一些golang特性。

  • 切片的特性

    • 元素的有序性
    • 非線程安全

  • map的特性

    • 元素的無序性
    • 非線程安全

  • 并發(fā)場景下的解決方案

    • 互斥鎖：可以對非線程安全的數(shù)據(jù)結(jié)構創(chuàng)建臨界區(qū)，一般用于同步場景；
    • 管道：可以對非線程安全的數(shù)據(jù)結(jié)構進行異步處理

實現(xiàn)思路

既然，我們了解了golang的一些特性，那么，我們接下來就要明確，如何去實現(xiàn)優(yōu)先級隊列了。

我們都知道，無論是哪一種隊列，必然是存在生產(chǎn)者和消費者兩個部分，對于優(yōu)先級隊列來說，更是如此。因此，咱們的實現(xiàn)思路，也將從這兩個部分來談。

1、生產(chǎn)者

對于生產(chǎn)者來說，他只需要推送一個任務及其優(yōu)先級過來，咱們就得根據(jù)優(yōu)先級處理他的任務。

由于，我們不大好判斷，到底會有多少種不同的優(yōu)先級傳過來，也無法確定，每種優(yōu)先級下有多少個任務要處理，所以，我們可以考慮使用map來存儲優(yōu)先級隊列。其中key為優(yōu)先級，value為屬于該優(yōu)先級下的任務隊列（即管道） 。

2、消費者

對于消費者來說，他需要獲取優(yōu)先級最高的任務進行消費。

但是，如果只按照上面所說的map來存儲優(yōu)先級隊列的話，我們是沒法找到優(yōu)先級最高的任務隊列的，因為map的元素是無序的。那么，我們怎么處理這個問題呢？

我們都知道，在golang的數(shù)據(jù)結(jié)構里，切片的元素是具有有序性的。那么，我們只需要將所有的優(yōu)先級按從小到大的方式，存儲在一個切片里，就可以了。等到消費的時候，我們可以先從切片中，取出最大的優(yōu)先級，然后再根據(jù)這個key去優(yōu)先級隊列的map中查詢，是不是就可以了？

目標規(guī)劃

想好了實現(xiàn)思路之后，我們就得對接下來的代碼實現(xiàn)做一個規(guī)劃了。

  • 數(shù)據(jù)結(jié)構

    • 存儲優(yōu)先級隊列的map
    • 存儲優(yōu)先級的切片
    • 互斥鎖
    • 其他......

  • 生產(chǎn)者

    • 添加任務到優(yōu)先級隊列

  • 消費者

    • 從優(yōu)先級隊列獲取任務

步步為營

1、數(shù)據(jù)流

（1）調(diào)用NewPriorityQueue() ，初始化優(yōu)先級隊列對象。

（2）初始化優(yōu)先級隊列map。

（3）開啟協(xié)程，監(jiān)聽一個接收推送任務的全局管道pushChan。

（4）用戶調(diào)用Push() ，推送的任務進入pushChan。

（5）推送的任務被加到優(yōu)先級隊列中。

（6）消費者從優(yōu)先級隊列中獲取優(yōu)先級最高的一個任務。

（7）消費者執(zhí)行任務。

2、數(shù)據(jù)結(jié)構

（1）優(yōu)先級隊列對象

type PriorityQueue struct {
   mLock      sync.Mutex         // 互斥鎖，queues和priorities并發(fā)操作時使用
   queues     map[int]chan *task // 優(yōu)先級隊列map
   pushChan   chan *task         // 推送任務管道
   priorities []int              // 記錄優(yōu)先級的切片（優(yōu)先級從小到大排列）
}

（2）任務對象

type task struct {
   priority int    // 任務的優(yōu)先級
   f        func() // 任務的執(zhí)行函數(shù)
}

3、初始化優(yōu)先級隊列對象 

func NewPriorityQueue() *PriorityQueue {
   pq := &PriorityQueue{
      queues:   make(map[int]chan *task), // 初始化優(yōu)先級隊列map
      pushChan: make(chan *task, 100),
   }

   return pq
}

當然，在這個過程中，我們需要對pushChan進行監(jiān)聽。如果有任務推送過來，咱們得處理。

func (pq *PriorityQueue) listenPushChan() {
   for {
      select {
      case taskEle := <-pq.pushChan:
         // TODO 這里接收到推送的任務，并且準備處理
      }
   }
}

將這個監(jiān)聽函數(shù)放到NewPriorityQueue()中：

func NewPriorityQueue() *PriorityQueue {
   pq := &PriorityQueue{
      queues:   make(map[int]chan *task),
      pushChan: make(chan *task, 100),
   }

   // 監(jiān)聽pushChan
   go pq.listenPushChan()
   return pq
}

4、生產(chǎn)者推送任務

生產(chǎn)者推送任務的時候，我們只需要將任務放到pushChan中：

func (pq *PriorityQueue) Push(f func(), priority int) {
   pq.pushChan <- &task{
      f:        f,
      priority: priority,
   }
}

5、將推送任務加到優(yōu)先級隊列中

這一步就比較關鍵了。我們前面談到，優(yōu)先級隊列最核心的數(shù)據(jù)結(jié)構有兩個：優(yōu)先級隊列map和優(yōu)先級切片。因此，推送任務添加到優(yōu)先級隊列的操作，咱們得分兩種情況來看：

（1）之前已經(jīng)推過相同優(yōu)先級的任務

這種情況非常簡單，咱們其實只要操作優(yōu)先級隊列map就可以了。

func (pq *PriorityQueue) listenPushChan() {
   for {
      select {
      case taskEle := <-pq.pushChan:
         priority := taskEle.priority
         pq.mLock.Lock()
         if v, ok := pq.queues[priority]; ok {
            pq.mLock.Unlock()
            // 之前推送過相同優(yōu)先級的任務
            // 將推送的任務塞到對應優(yōu)先級的隊列中
            v <- taskEle
            continue
         }

         // todo 之前未推過相同優(yōu)先級任務的處理...
      }
   }
}

（2）之前未推過相同優(yōu)先級的任務

這種情況會稍微復雜一些。我們不僅要將新的優(yōu)先級插入到優(yōu)先級切片正確的位置，而且要將任務添加到對應優(yōu)先級的隊列。

1）將新的優(yōu)先級插入到優(yōu)先級切片中

a. 首先，咱們得尋找新優(yōu)先級在切片中的插入位置。這里，咱們用了二分法。

// 通過二分法尋找新優(yōu)先級的切片插入位置
func (pq *PriorityQueue) getNewPriorityInsertIndex(priority int, leftIndex, rightIndex int) (index int) {
   if len(pq.priorities) == 0 {
      // 如果當前優(yōu)先級切片沒有元素，則插入的index就是0
      return 0
   }

   length := rightIndex - leftIndex
   if pq.priorities[leftIndex] >= priority {
      // 如果當前切片中最小的元素都超過了插入的優(yōu)先級，則插入位置應該是最左邊
      return leftIndex
   }

   if pq.priorities[rightIndex] <= priority {
      // 如果當前切片中最大的元素都沒超過插入的優(yōu)先級，則插入位置應該是最右邊
      return rightIndex + 1
   }

   if length == 1 && pq.priorities[leftIndex] < priority && pq.priorities[rightIndex] >= priority {
      // 如果插入的優(yōu)先級剛好在僅有的兩個優(yōu)先級之間，則中間的位置就是插入位置
      return leftIndex + 1
   }

   middleVal := pq.priorities[leftIndex+length/2]

   // 這里用二分法遞歸的方式，一直尋找正確的插入位置
   if priority <= middleVal {
      return pq.getNewPriorityInsertIndex(priority, leftIndex, leftIndex+length/2)
   } else {
      return pq.getNewPriorityInsertIndex(priority, leftIndex+length/2, rightIndex)
   }
}

b. 找到插入位置之后，我們才要插入。在這個過程中，插入位置右側(cè)的元素全部都要向右邊移動一位。

// index右側(cè)元素均需要向后移動一個單位
func (pq *PriorityQueue) moveNextPriorities(index, priority int) {
   pq.priorities = append(pq.priorities, 0)
   copy(pq.priorities[index+1:], pq.priorities[index:])

   pq.priorities[index] = priority
}

這樣，我們就成功地將新的優(yōu)先級插入了切片。

2）將推送任務放入優(yōu)先級隊列map也就順理成章。

// 創(chuàng)建一個新優(yōu)先級管道
pq.queues[priority] = make(chan *task, 10000)

// 將任務塞到新的優(yōu)先級管道中
pq.queues[priority] <- taskEle

因此，listenPushChan()的代碼如下：

func (pq *PriorityQueue) listenPushChan() {
   for {
      select {
      case taskEle := <-pq.pushChan:
         priority := taskEle.priority
         pq.mLock.Lock()
         if v, ok := pq.queues[priority]; ok {
            pq.mLock.Unlock()
            // 將推送的任務塞到對應優(yōu)先級的隊列中
            v <- taskEle
            continue
         }

         // 如果這是一個新的優(yōu)先級，則需要插入優(yōu)先級切片，并且新建一個優(yōu)先級的queue
         // 通過二分法尋找新優(yōu)先級的切片插入位置
         index := pq.getNewPriorityInsertIndex(priority, 0, len(pq.priorities)-1)

         // index右側(cè)元素均需要向后移動一個單位
         pq.moveNextPriorities(index, priority)

         // 創(chuàng)建一個新優(yōu)先級隊列
         pq.queues[priority] = make(chan *task, 10000)

         // 將任務塞到新的優(yōu)先級隊列中
         pq.queues[priority] <- taskEle
         pq.mLock.Unlock()
      }
   }
}

完成了生產(chǎn)者部分之后，接下來我們看看消費者。

6、消費者消費隊列

這里分成兩個步驟，首先咱們得拿到最高優(yōu)先級隊列的任務，然后再去執(zhí)行任務。代碼如下：

// 消費者輪詢獲取最高優(yōu)先級的任務
func (pq *PriorityQueue) Consume() {
   for {
      task := pq.Pop()
      if task == nil {
         // 未獲取到任務，則繼續(xù)輪詢
         continue
      }

      // 獲取到了任務，就執(zhí)行任務
      task.f()
   }
}

// 取出最高優(yōu)先級隊列中的一個任務
func (pq *PriorityQueue) Pop() *task {
   pq.mLock.Lock()
   defer pq.mLock.Unlock()

   for i := len(pq.priorities) - 1; i >= 0; i-- {
      if len(pq.queues[pq.priorities[i]]) == 0 {
         // 如果當前優(yōu)先級的隊列沒有任務，則看低一級優(yōu)先級的隊列中有沒有任務
         continue
      }

      // 如果當前優(yōu)先級的隊列里有任務，則取出一個任務。
      return <-pq.queues[pq.priorities[i]]
   }

   // 如果所有隊列都沒有任務，則返回null
   return nil
}

7、完整代碼

這樣，咱們的優(yōu)先級隊列就實現(xiàn)了。下面，我們將完整代碼展示。

pq.go

package priority_queue

import (
   "sync"
)

type PriorityQueue struct {
   mLock      sync.Mutex         // 互斥鎖，queues和priorities并發(fā)操作時使用
   queues     map[int]chan *task // 優(yōu)先級隊列map
   pushChan   chan *task         // 推送任務管道
   priorities []int              // 記錄優(yōu)先級的切片（優(yōu)先級從小到大排列）
}

type task struct {
   priority int    // 任務的優(yōu)先級
   f        func() // 任務的執(zhí)行函數(shù)
}

func NewPriorityQueue() *PriorityQueue {
   pq := &PriorityQueue{
      queues:   make(map[int]chan *task),
      pushChan: make(chan *task, 100),
   }

   go pq.listenPushChan()
   return pq
}

func (pq *PriorityQueue) listenPushChan() {
   for {
      select {
      case taskEle := <-pq.pushChan:
         priority := taskEle.priority
         pq.mLock.Lock()
         if v, ok := pq.queues[priority]; ok {
            pq.mLock.Unlock()
            // 將推送的任務塞到對應優(yōu)先級的隊列中
            v <- taskEle
            continue
         }

         // 如果這是一個新的優(yōu)先級，則需要插入優(yōu)先級切片，并且新建一個優(yōu)先級的queue
         // 通過二分法尋找新優(yōu)先級的切片插入位置
         index := pq.getNewPriorityInsertIndex(priority, 0, len(pq.priorities)-1)

         // index右側(cè)元素均需要向后移動一個單位
         pq.moveNextPriorities(index, priority)

         // 創(chuàng)建一個新優(yōu)先級隊列
         pq.queues[priority] = make(chan *task, 10000)

         // 將任務塞到新的優(yōu)先級隊列中
         pq.queues[priority] <- taskEle
         pq.mLock.Unlock()
      }
   }
}

// 插入work
func (pq *PriorityQueue) Push(f func(), priority int) {
   pq.pushChan <- &task{
      f:        f,
      priority: priority,
   }
}

// index右側(cè)元素均需要向后移動一個單位
func (pq *PriorityQueue) moveNextPriorities(index, priority int) {
   pq.priorities = append(pq.priorities, 0)
   copy(pq.priorities[index+1:], pq.priorities[index:])

   pq.priorities[index] = priority
}

// 通過二分法尋找新優(yōu)先級的切片插入位置
func (pq *PriorityQueue) getNewPriorityInsertIndex(priority int, leftIndex, rightIndex int) (index int) {
   if len(pq.priorities) == 0 {
      // 如果當前優(yōu)先級切片沒有元素，則插入的index就是0
      return 0
   }

   length := rightIndex - leftIndex
   if pq.priorities[leftIndex] >= priority {
      // 如果當前切片中最小的元素都超過了插入的優(yōu)先級，則插入位置應該是最左邊
      return leftIndex
   }

   if pq.priorities[rightIndex] <= priority {
      // 如果當前切片中最大的元素都沒超過插入的優(yōu)先級，則插入位置應該是最右邊
      return rightIndex + 1
   }

   if length == 1 && pq.priorities[leftIndex] < priority && pq.priorities[rightIndex] >= priority {
      // 如果插入的優(yōu)先級剛好在僅有的兩個優(yōu)先級之間，則中間的位置就是插入位置
      return leftIndex + 1
   }

   middleVal := pq.priorities[leftIndex+length/2]

   // 這里用二分法遞歸的方式，一直尋找正確的插入位置
   if priority <= middleVal {
      return pq.getNewPriorityInsertIndex(priority, leftIndex, leftIndex+length/2)
   } else {
      return pq.getNewPriorityInsertIndex(priority, leftIndex+length/2, rightIndex)
   }
}

// 取出最高優(yōu)先級隊列中的一個任務
func (pq *PriorityQueue) Pop() *task {
   pq.mLock.Lock()
   defer pq.mLock.Unlock()

   for i := len(pq.priorities) - 1; i >= 0; i-- {
      if len(pq.queues[pq.priorities[i]]) == 0 {
         // 如果當前優(yōu)先級的隊列沒有任務，則看低一級優(yōu)先級的隊列中有沒有任務
         continue
      }

      // 如果當前優(yōu)先級的隊列里有任務，則取出一個任務。
      return <-pq.queues[pq.priorities[i]]
   }

   // 如果所有隊列都沒有任務，則返回null
   return nil
}

// 消費者輪詢獲取最高優(yōu)先級的任務
func (pq *PriorityQueue) Consume() {
   for {
      task := pq.Pop()
      if task == nil {
         // 未獲取到任務，則繼續(xù)輪詢
         continue
      }

      // 獲取到了任務，就執(zhí)行任務
      task.f()
   }
}

測試代碼pq_test.go：

package priority_queue

import (
   "fmt"
   "math/rand"
   "testing"
   "time"
)

func TestQueue(t *testing.T) {
   defer func() {
      if err := recover(); err != nil {
         fmt.Println(err)
      }
   }()
   pq := NewPriorityQueue()
   rand.Seed(time.Now().Unix())

   // 我們在這里，隨機生成一些優(yōu)先級任務
   for i := 0; i < 100; i++ {
      a := rand.Intn(10)
      go func(i int) {
         pq.Push(func() {
            fmt.Println("推送任務的編號為：", i)
            fmt.Println("推送的任務優(yōu)先級為：", a)
            fmt.Println("============")
         }, a)
      }(i)
   }

   // 這里會阻塞，消費者會輪詢查詢?nèi)蝿贞犃?
   pq.Consume()
}

發(fā)散思維

上面的方案的確是實現(xiàn)了優(yōu)先級隊列，但是，有一種極端情況：如果消費者的消費速度遠遠小于生產(chǎn)者的生產(chǎn)速度，并且高優(yōu)先級的任務被不斷插入，這樣，低優(yōu)先級的任務就會有“餓死”的風險。

對于這種情況，我們在消費的時候，可以考慮給每一個優(yōu)先級隊列分配一個權重，高優(yōu)先級的隊列有更大的概率被消費，低優(yōu)先級的概率相對較小。感興趣的朋友們，可以自己去實現(xiàn)一下。

小結(jié)

本文和大家討論了優(yōu)先級隊列在golang中的一種實現(xiàn)方案，里面應用到了切片、map、互斥鎖、管道等諸多golang特性，可以說是一個非常典型的案例。其實，優(yōu)先級隊列在實際的業(yè)務場景中使用廣泛，其實現(xiàn)方式也不止一種，我們需要根據(jù)實際的需求，選擇最優(yōu)解。

原文鏈接：https://juejin.cn/post/7180632889489358906

欄目分類
最近更新