相信對于隊(duì)列的概念大家都不會陌生，這種先入先出的數(shù)據(jù)結(jié)構(gòu)應(yīng)用很廣泛，像一般的生產(chǎn)消費(fèi)都會用到隊(duì)列，關(guān)于Queue的用法介紹可以參考我之前的文章 python中的Queue與多進(jìn)程（multiprocessing）還有棧，棧是一種先入后出的數(shù)據(jù)結(jié)構(gòu),面優(yōu)先隊(duì)列有別于普通的隊(duì)列與棧，在實(shí)現(xiàn)上，它一般通過堆這一數(shù)據(jù)結(jié)構(gòu)，而堆其實(shí)是一種完全二叉樹，它會對進(jìn)入容器的元素進(jìn)行排序（根據(jù)事先指定的規(guī)則），出隊(duì)的順序則會是二叉樹的根結(jié)點(diǎn)代表的元素。接下來介紹幾種優(yōu)先隊(duì)列的實(shí)現(xiàn)。

通過heapq模塊

heapq是一個(gè)二叉堆的實(shí)現(xiàn)，它內(nèi)部使用內(nèi)置的list對象，它無論插入還是獲取最小元素復(fù)雜度都在O(log n)。這里主要用到它的heappush與heappop方法，heappush 方法需要傳入兩個(gè)參數(shù)，一個(gè)是列表(list)，另外是一個(gè)對象，這里的對象須是可比較對象，就是它可以通過cmp方法來比較大小，以下是在 python2 中的代碼實(shí)現(xiàn)

#coding:gbk import heapqtasks = [] heapq.heappush(tasks,(10,'aaa')) heapq.heappush(tasks,(40,'bbb')) heapq.heappush(tasks,(30,'ccc')) heapq.heappush(tasks,(20,'ddd'))while tasks:task = heapq.heappop(tasks)print(task)

運(yùn)行結(jié)果如下

(10, 'aaa') (20, 'ddd') (30, 'ccc') (40, 'bbb')

可以看到，我放入 tasks 列表里的元素是個(gè) set 對象，對象第一個(gè)元素是個(gè) int 類型的數(shù)字，如果使用cmp方法進(jìn)行比較的話

>>> cmp(10,20) -1 >>> cmp(10,10) 0 >>> cmp(10,5) 1

對于小于，等于，大于分別返回的是-1，0，1，其實(shí)這也是在定義sorted的實(shí)現(xiàn)方法,

>>> sorted([(10,'aaaa'),(30,'bbbb')]) [(10, 'aaaa'), (30, 'bbbb')] >>> sorted([(40,'aaaa'),(30,'bbbb')]) [(30, 'bbbb'), (40, 'aaaa')] >>> sorted([(30,'aaaa'),(30,'bbbb')]) [(30, 'aaaa'), (30, 'bbbb')] >>> sorted([(30,'bbbb'),(30,'abbb')]) [(30, 'abbb'), (30, 'bbbb')]

可以看到在sorted方法里，它的排序算法是通過比較第一個(gè)元素的大小，小的排在前面，第一個(gè)元素相同再比較第二個(gè)元素，看返回之前的代碼，heapq.heappush 將 set 元素添加到列表元素以后，將對其進(jìn)行重新排序，將最小的放在前面，于是就得到了上面的打印結(jié)果。

上面是使用python自帶的 set 數(shù)據(jù)結(jié)構(gòu)，可否自定義一種類型呢，比較在實(shí)現(xiàn)生活中，在上班的第一件事是給自已寫一下今天要完成哪些事情，其實(shí)哪些事情的優(yōu)先級比較高就是先做哪些事情，其實(shí)在上面也說到 sorted 方法，這個(gè)方法其實(shí)就是在調(diào)用對象的 __cmp__ 方法，好么我可以單獨(dú)定義一個(gè)帶有 __cmp__ 方法的對象則可以實(shí)現(xiàn)優(yōu)先隊(duì)列中的對象排序。

#coding:gbk import heapq# 使用heapq實(shí)現(xiàn)優(yōu)先隊(duì)列 #定義一個(gè)可比較對象 class CompareAble:def __init__(self,priority,jobname):self.priority = priorityself.jobname = jobnamedef __cmp__(self, other):if self.priority < other.priority:return -1elif self.priority == other.priority:return 0else:return 1joblist = []heapq.heappush(joblist,CompareAble(80,'eat')) heapq.heappush(joblist,CompareAble(70,'a write plan2')) heapq.heappush(joblist,CompareAble(70,'write plan')) heapq.heappush(joblist,CompareAble(90,'sleep')) heapq.heappush(joblist,CompareAble(100,'write code'))while joblist:task = heapq.heappop(joblist)print(task.priority,task.jobname)

運(yùn)行結(jié)果：

(70, 'write plan') (70, 'a write plan2') (80, 'eat') (90, 'sleep') (100, 'write code')

上面的compareAble 類初始化有兩個(gè)參數(shù)，一個(gè)是優(yōu)先級，一個(gè)是事情的名字，我這里定義的是優(yōu)先級數(shù)值越小排序越靠前，也可以定義成數(shù)值越大越靠前。如果優(yōu)先級相同，則按照插入順序來排序。

通過Queue,PriorityQueue類型實(shí)現(xiàn)

這個(gè)優(yōu)先級隊(duì)列內(nèi)部使用了heapq,不同的是PriorityQueue的操作是同步的，提供鎖操作，支持并發(fā)的生產(chǎn)者和消費(fèi)者，而且它的接口更加友好，它繼承自Queue，所以好多Queue的方法可以直接使用

#coding:gbk import heapq from queue import Queue,PriorityQueue# 使用heapq實(shí)現(xiàn)優(yōu)先隊(duì)列 #定義一個(gè)可比較對象 class CompareAble:def __init__(self,priority,jobname):self.priority = priorityself.jobname = jobnamedef __cmp__(self, other):if self.priority < other.priority:return -1elif self.priority == other.priority:return 0else:return 1pq = PriorityQueue() pq.put(CompareAble(80,'eat')) pq.put(CompareAble(70,'a write plan2')) pq.put(CompareAble(70,'write plan')) pq.put(CompareAble(90,'sleep')) pq.put(CompareAble(100,'write code'))while pq.qsize()!= 0:task = pq.get_nowait()print(task.jobname,task.priority)

接下來通過一個(gè)生產(chǎn)消費(fèi)的實(shí)例來說明優(yōu)先隊(duì)列的使用

有三個(gè)生產(chǎn)者和二個(gè)消費(fèi)者，生產(chǎn)者向隊(duì)列中生產(chǎn)有優(yōu)先級的任務(wù)，消費(fèi)者也是優(yōu)先消費(fèi)高級別的任務(wù)

#coding:gbk from queue import PriorityQueue import time import random import threading# 使用heapq實(shí)現(xiàn)優(yōu)先隊(duì)列 #定義一個(gè)可比較對象 class CompareAble:def __init__(self,priority,jobname):self.priority = priorityself.jobname = jobnamedef __cmp__(self, other):if self.priority < other.priority:return -1elif self.priority == other.priority:return 0else:return 1tasks = [(i, "do task %s"%i) for i in range(10,100,5)] def produce(pq,lock):while True:lock.acquire()task = tasks[random.randint(0,len(tasks)-1)]print('put %s %s in pq'%(task[0],task[1]))pq.put(CompareAble(task[0],task[1]))time.sleep(1)lock.release()def consumer(pq,lock):while True:lock.acquire()task = pq.get_nowait()if task:print(task.priority, task.jobname)else:time.sleep(1)lock.release()if __name__ == '__main__':task_queue = PriorityQueue()task_lock = threading.Lock()for i in range(3):t = threading.Thread(target=produce,args=(task_queue,task_lock))t.setDaemon(False)t.start()for i in range(2):t = threading.Thread(target=consumer,args=(task_queue,task_lock))t.setDaemon(False)t.start()

運(yùn)行結(jié)果：

put 30 do task 30 in pq put 20 do task 20 in pq put 75 do task 75 in pq (20, 'do task 20') (30, 'do task 30') put 20 do task 20 in pq put 15 do task 15 in pq put 70 do task 70 in pq (15, 'do task 15') (20, 'do task 20') put 85 do task 85 in pq put 10 do task 10 in pq put 30 do task 30 in pq (10, 'do task 10') (30, 'do task 30') put 70 do task 70 in pq put 10 do task 10 in pq put 55 do task 55 in pq (10, 'do task 10') (55, 'do task 55') put 20 do task 20 in pq put 45 do task 45 in pq put 75 do task 75 in pq (20, 'do task 20') (45, 'do task 45') put 40 do task 40 in pq put 40 do task 40 in pq ...

可以看出，每次取出來的都是當(dāng)前隊(duì)列中 priority 最小的數(shù)

python3 中的使用方法

上面的代碼無法在python3中運(yùn)行，主要是因?yàn)閜ython3沒有cmp方法，運(yùn)行得到的異常信息是

TypeError: unorderable types: CompareAble() < CompareAble()

就是沒有一個(gè)cmp 的操作

需要在上面定義一個(gè) __lt__ 方法

#coding:gbk from queue import PriorityQueue import time import random import threading# 使用heapq實(shí)現(xiàn)優(yōu)先隊(duì)列 #定義一個(gè)可比較對象 class CompareAble:def __init__(self,priority,jobname):self.priority = priorityself.jobname = jobname# def __cmp__(self, other):# if self.priority < other.priority:# return -1# elif self.priority == other.priority:# return 0# else:# return 1def __lt__(self, other):if self.priority <= other.priority:return Falseelse:return Truetasks = [(i, "do task %s"%i) for i in range(10,100,5)] def produce(pq,lock):while True:lock.acquire()task = tasks[random.randint(0,len(tasks)-1)]print('put %s %s in pq'%(task[0],task[1]))pq.put(CompareAble(task[0],task[1]))lock.release()time.sleep(1)def consumer(pq,lock):while True:lock.acquire()try:if pq.empty():continuetask = pq.get_nowait()if task:print(task.priority, task.jobname)finally:lock.release()time.sleep(1)if __name__ == '__main__':task_queue = PriorityQueue()task_lock = threading.Lock()for i in range(3):t = threading.Thread(target=produce,args=(task_queue,task_lock))t.setDaemon(False)t.start()for i in range(2):t = threading.Thread(target=consumer,args=(task_queue,task_lock))t.setDaemon(False)t.start()

上面的代碼我修改了一點(diǎn)對于大小的判斷，與之前的是反的，這里 priority 越大則越先返回，上面的代碼在 python2 中也可以運(yùn)行，所有如果為了兼容性可以選擇定義使用 __lt__ 方法。

參考文章
用Python實(shí)現(xiàn)優(yōu)先級隊(duì)列的3種方法
python的優(yōu)先隊(duì)列示例

更多文章請點(diǎn)擊查看我的個(gè)人博客

楊彥星 | 序語程言?www.yangyanxing.com

也歡迎關(guān)注我的個(gè)人公眾賬號 序語程言

總結(jié)

以上是生活随笔為你收集整理的python 优先队列_python中使用优先队列的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。