相信對于隊列的概念大家都不會陌生，這種先入先出的數據結構應用很廣泛，像一般的生產消費都會用到隊列，關于Queue的用法介紹可以參考我之前的文章 python中的Queue與多進程（multiprocessing）還有棧，棧是一種先入后出的數據結構,面優先隊列有別于普通的隊列與棧，在實現上，它一般通過堆這一數據結構，而堆其實是一種完全二叉樹，它會對進入容器的元素進行排序（根據事先指定的規則），出隊的順序則會是二叉樹的根結點代表的元素。接下來介紹幾種優先隊列的實現。

通過heapq模塊

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

#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)

運行結果如下

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

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

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

對于小于，等于，大于分別返回的是-1，0，1，其實這也是在定義sorted的實現方法,

>>> 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方法里，它的排序算法是通過比較第一個元素的大小，小的排在前面，第一個元素相同再比較第二個元素，看返回之前的代碼，heapq.heappush 將 set 元素添加到列表元素以后，將對其進行重新排序，將最小的放在前面，于是就得到了上面的打印結果。

上面是使用python自帶的 set 數據結構，可否自定義一種類型呢，比較在實現生活中，在上班的第一件事是給自已寫一下今天要完成哪些事情，其實哪些事情的優先級比較高就是先做哪些事情，其實在上面也說到 sorted 方法，這個方法其實就是在調用對象的 __cmp__ 方法，好么我可以單獨定義一個帶有 __cmp__ 方法的對象則可以實現優先隊列中的對象排序。

#coding:gbk import heapq# 使用heapq實現優先隊列 #定義一個可比較對象 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)

運行結果：

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

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

通過Queue,PriorityQueue類型實現

這個優先級隊列內部使用了heapq,不同的是PriorityQueue的操作是同步的，提供鎖操作，支持并發的生產者和消費者，而且它的接口更加友好，它繼承自Queue，所以好多Queue的方法可以直接使用

#coding:gbk import heapq from queue import Queue,PriorityQueue# 使用heapq實現優先隊列 #定義一個可比較對象 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)

接下來通過一個生產消費的實例來說明優先隊列的使用

有三個生產者和二個消費者，生產者向隊列中生產有優先級的任務，消費者也是優先消費高級別的任務

#coding:gbk from queue import PriorityQueue import time import random import threading# 使用heapq實現優先隊列 #定義一個可比較對象 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()

運行結果：

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 ...

可以看出，每次取出來的都是當前隊列中 priority 最小的數

python3 中的使用方法

上面的代碼無法在python3中運行，主要是因為python3沒有cmp方法，運行得到的異常信息是

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

就是沒有一個cmp 的操作

需要在上面定義一個 __lt__ 方法

#coding:gbk from queue import PriorityQueue import time import random import threading# 使用heapq實現優先隊列 #定義一個可比較對象 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()

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

參考文章
用Python實現優先級隊列的3種方法
python的優先隊列示例

更多文章請點擊查看我的個人博客

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

也歡迎關注我的個人公眾賬號 序語程言

總結

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

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。