golang優(yōu)先級隊列的實現(xiàn)全過程
前言
在數(shù)據(jù)結(jié)構(gòu)中,隊列遵循著FIFO(先進先出)的規(guī)則。在此基礎(chǔ)上,人們引申出了“優(yōu)先級隊列”的概念。
優(yōu)先級隊列,是帶有優(yōu)先級屬性的隊列,所有的隊列元素按照優(yōu)先級進行排序,消費者會先對優(yōu)先級高的隊列元素進行處理。
優(yōu)先級隊列的使用場景也是非常多的。比如,作業(yè)調(diào)度系統(tǒng),當(dā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)容脈絡(luò)
為了讓大家腦海里有個大致的輪廓,我先把正文的大綱展示出來。
基礎(chǔ)知識
在正式開始“優(yōu)先級隊列”這個話題之前,我們首先要明確以下的一些golang特性。
切片的特性
- 元素的有序性
- 非線程安全
map的特性
- 元素的無序性
- 非線程安全
并發(fā)場景下的解決方案
- 互斥鎖:可以對非線程安全的數(shù)據(jù)結(jié)構(gòu)創(chuàng)建臨界區(qū),一般用于同步場景;
- 管道:可以對非線程安全的數(shù)據(jù)結(jié)構(gòu)進行異步處理
實現(xiàn)思路
既然,我們了解了golang的一些特性,那么,我們接下來就要明確,如何去實現(xiàn)優(yōu)先級隊列了。
我們都知道,無論是哪一種隊列,必然是存在生產(chǎn)者和消費者兩個部分,對于優(yōu)先級隊列來說,更是如此。因此,咱們的實現(xiàn)思路,也將從這兩個部分來談。
1、生產(chǎn)者
對于生產(chǎn)者來說,他只需要推送一個任務(wù)及其優(yōu)先級過來,咱們就得根據(jù)優(yōu)先級處理他的任務(wù)。
由于,我們不大好判斷,到底會有多少種不同的優(yōu)先級傳過來,也無法確定,每種優(yōu)先級下有多少個任務(wù)要處理,所以,我們可以考慮使用map來存儲優(yōu)先級隊列。其中key為優(yōu)先級,value為屬于該優(yōu)先級下的任務(wù)隊列(即管道) 。
2、消費者
對于消費者來說,他需要獲取優(yōu)先級最高的任務(wù)進行消費。
但是,如果只按照上面所說的map來存儲優(yōu)先級隊列的話,我們是沒法找到優(yōu)先級最高的任務(wù)隊列的,因為map的元素是無序的。那么,我們怎么處理這個問題呢?
我們都知道,在golang的數(shù)據(jù)結(jié)構(gòu)里,切片的元素是具有有序性的。那么,我們只需要將所有的優(yōu)先級按從小到大的方式,存儲在一個切片里,就可以了。等到消費的時候,我們可以先從切片中,取出最大的優(yōu)先級,然后再根據(jù)這個key去優(yōu)先級隊列的map中查詢,是不是就可以了?
目標(biāo)規(guī)劃
想好了實現(xiàn)思路之后,我們就得對接下來的代碼實現(xiàn)做一個規(guī)劃了。
數(shù)據(jù)結(jié)構(gòu)
- 存儲優(yōu)先級隊列的map
- 存儲優(yōu)先級的切片
- 互斥鎖
- 其他......
生產(chǎn)者
- 添加任務(wù)到優(yōu)先級隊列
消費者
- 從優(yōu)先級隊列獲取任務(wù)
步步為營
1、數(shù)據(jù)流
(1)調(diào)用NewPriorityQueue() ,初始化優(yōu)先級隊列對象。
(2)初始化優(yōu)先級隊列map。
(3)開啟協(xié)程,監(jiān)聽一個接收推送任務(wù)的全局管道pushChan。
(4)用戶調(diào)用Push() ,推送的任務(wù)進入pushChan。
(5)推送的任務(wù)被加到優(yōu)先級隊列中。
(6)消費者從優(yōu)先級隊列中獲取優(yōu)先級最高的一個任務(wù)。
(7)消費者執(zhí)行任務(wù)。
2、數(shù)據(jù)結(jié)構(gòu)
(1)優(yōu)先級隊列對象
type PriorityQueue struct { mLock sync.Mutex // 互斥鎖,queues和priorities并發(fā)操作時使用 queues map[int]chan *task // 優(yōu)先級隊列map pushChan chan *task // 推送任務(wù)管道 priorities []int // 記錄優(yōu)先級的切片(優(yōu)先級從小到大排列) }
(2)任務(wù)對象
type task struct { priority int // 任務(wù)的優(yōu)先級 f func() // 任務(wù)的執(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 }
當(dāng)然,在這個過程中,我們需要對pushChan進行監(jiān)聽。如果有任務(wù)推送過來,咱們得處理。
func (pq *PriorityQueue) listenPushChan() { for { select { case taskEle := <-pq.pushChan: // TODO 這里接收到推送的任務(wù),并且準(zhǔn)備處理 } } }
將這個監(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)者推送任務(wù)
生產(chǎn)者推送任務(wù)的時候,我們只需要將任務(wù)放到pushChan中:
func (pq *PriorityQueue) Push(f func(), priority int) { pq.pushChan <- &task{ f: f, priority: priority, } }
5、將推送任務(wù)加到優(yōu)先級隊列中
這一步就比較關(guān)鍵了。我們前面談到,優(yōu)先級隊列最核心的數(shù)據(jù)結(jié)構(gòu)有兩個:優(yōu)先級隊列map和優(yōu)先級切片。因此,推送任務(wù)添加到優(yōu)先級隊列的操作,咱們得分兩種情況來看:
(1)之前已經(jīng)推過相同優(yōu)先級的任務(wù)
這種情況非常簡單,咱們其實只要操作優(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)先級的任務(wù) // 將推送的任務(wù)塞到對應(yīng)優(yōu)先級的隊列中 v <- taskEle continue } // todo 之前未推過相同優(yōu)先級任務(wù)的處理... } } }
(2)之前未推過相同優(yōu)先級的任務(wù)
這種情況會稍微復(fù)雜一些。我們不僅要將新的優(yōu)先級插入到優(yōu)先級切片正確的位置,而且要將任務(wù)添加到對應(yīng)優(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 { // 如果當(dāng)前優(yōu)先級切片沒有元素,則插入的index就是0 return 0 } length := rightIndex - leftIndex if pq.priorities[leftIndex] >= priority { // 如果當(dāng)前切片中最小的元素都超過了插入的優(yōu)先級,則插入位置應(yīng)該是最左邊 return leftIndex } if pq.priorities[rightIndex] <= priority { // 如果當(dāng)前切片中最大的元素都沒超過插入的優(yōu)先級,則插入位置應(yīng)該是最右邊 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)將推送任務(wù)放入優(yōu)先級隊列map也就順理成章。
// 創(chuàng)建一個新優(yōu)先級管道 pq.queues[priority] = make(chan *task, 10000) // 將任務(wù)塞到新的優(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() // 將推送的任務(wù)塞到對應(yīng)優(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) // 將任務(wù)塞到新的優(yōu)先級隊列中 pq.queues[priority] <- taskEle pq.mLock.Unlock() } } }
完成了生產(chǎn)者部分之后,接下來我們看看消費者。
6、消費者消費隊列
這里分成兩個步驟,首先咱們得拿到最高優(yōu)先級隊列的任務(wù),然后再去執(zhí)行任務(wù)。代碼如下:
// 消費者輪詢獲取最高優(yōu)先級的任務(wù) func (pq *PriorityQueue) Consume() { for { task := pq.Pop() if task == nil { // 未獲取到任務(wù),則繼續(xù)輪詢 continue } // 獲取到了任務(wù),就執(zhí)行任務(wù) task.f() } } // 取出最高優(yōu)先級隊列中的一個任務(wù) 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 { // 如果當(dāng)前優(yōu)先級的隊列沒有任務(wù),則看低一級優(yōu)先級的隊列中有沒有任務(wù) continue } // 如果當(dāng)前優(yōu)先級的隊列里有任務(wù),則取出一個任務(wù)。 return <-pq.queues[pq.priorities[i]] } // 如果所有隊列都沒有任務(wù),則返回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 // 推送任務(wù)管道 priorities []int // 記錄優(yōu)先級的切片(優(yōu)先級從小到大排列) } type task struct { priority int // 任務(wù)的優(yōu)先級 f func() // 任務(wù)的執(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() // 將推送的任務(wù)塞到對應(yīng)優(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) // 將任務(wù)塞到新的優(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 { // 如果當(dāng)前優(yōu)先級切片沒有元素,則插入的index就是0 return 0 } length := rightIndex - leftIndex if pq.priorities[leftIndex] >= priority { // 如果當(dāng)前切片中最小的元素都超過了插入的優(yōu)先級,則插入位置應(yīng)該是最左邊 return leftIndex } if pq.priorities[rightIndex] <= priority { // 如果當(dāng)前切片中最大的元素都沒超過插入的優(yōu)先級,則插入位置應(yīng)該是最右邊 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)先級隊列中的一個任務(wù) 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 { // 如果當(dāng)前優(yōu)先級的隊列沒有任務(wù),則看低一級優(yōu)先級的隊列中有沒有任務(wù) continue } // 如果當(dāng)前優(yōu)先級的隊列里有任務(wù),則取出一個任務(wù)。 return <-pq.queues[pq.priorities[i]] } // 如果所有隊列都沒有任務(wù),則返回null return nil } // 消費者輪詢獲取最高優(yōu)先級的任務(wù) func (pq *PriorityQueue) Consume() { for { task := pq.Pop() if task == nil { // 未獲取到任務(wù),則繼續(xù)輪詢 continue } // 獲取到了任務(wù),就執(zhí)行任務(wù) 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)先級任務(wù) for i := 0; i < 100; i++ { a := rand.Intn(10) go func(i int) { pq.Push(func() { fmt.Println("推送任務(wù)的編號為:", i) fmt.Println("推送的任務(wù)優(yōu)先級為:", a) fmt.Println("============") }, a) }(i) } // 這里會阻塞,消費者會輪詢查詢?nèi)蝿?wù)隊列 pq.Consume() }
發(fā)散思維
上面的方案的確是實現(xiàn)了優(yōu)先級隊列,但是,有一種極端情況:如果消費者的消費速度遠遠小于生產(chǎn)者的生產(chǎn)速度,并且高優(yōu)先級的任務(wù)被不斷插入,這樣,低優(yōu)先級的任務(wù)就會有“餓死”的風(fēng)險。
對于這種情況,我們在消費的時候,可以考慮給每一個優(yōu)先級隊列分配一個權(quán)重,高優(yōu)先級的隊列有更大的概率被消費,低優(yōu)先級的概率相對較小。感興趣的朋友們,可以自己去實現(xiàn)一下。
小結(jié)
本文和大家討論了優(yōu)先級隊列在golang中的一種實現(xiàn)方案,里面應(yīng)用到了切片、map、互斥鎖、管道等諸多golang特性,可以說是一個非常典型的案例。其實,優(yōu)先級隊列在實際的業(yè)務(wù)場景中使用廣泛,其實現(xiàn)方式也不止一種,我們需要根據(jù)實際的需求,選擇最優(yōu)解。
到此這篇關(guān)于golang優(yōu)先級隊列實現(xiàn)的文章就介紹到這了,更多相關(guān)golang優(yōu)先級隊列實現(xiàn)內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
GoFrame通用類型變量gvar與interface基本使用對比
這篇文章主要為大家介紹了GoFrame通用類型變量gvar與interface基本使用對比,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進步,早日升職加薪2022-06-06Golang String字符串類型轉(zhuǎn)Json格式
本文主要介紹了Golang String字符串類型轉(zhuǎn)Json格式的實現(xiàn),文中通過示例代碼介紹的非常詳細,對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2024-05-05使用Go?goroutine實現(xiàn)并發(fā)的Clock服務(wù)
這篇文章主要為大家詳細介紹了如何使用Go?goroutine實現(xiàn)并發(fā)的Clock服務(wù),文中的示例代碼講解詳細,感興趣的小伙伴可以跟隨小編一起學(xué)習(xí)一下2023-06-06