目录
相同优先级，不保证先进先出。
改进版，相同优先级，保证先入先出
相同优先级，不保证先进先出。
import queue
# 创建优先级队列
pq = queue.PriorityQueue()
# 添加元素 (优先级, 数据) – 数字越小优先级越高
pq.put((1, "高优先级任务"))
pq.put((3, "低优先级任务"))
pq.put((2, "中优先级任务"))
# 获取元素（按优先级顺序）
while not pq.empty():
priority, task = pq.get()
print(f"优先级: {priority}, 任务: {task}")
改进版，相同优先级，保证先入先出
import queue
import itertools
class FIFOPriorityQueue:
def __init__(self):
self.pq = queue.PriorityQueue()
self.counter = itertools.count() # 自动递增的计数器
def put(self, priority, item):
# sequence 确保相同优先级时按入队顺序排序
seq = next(self.counter)
self.pq.put((priority, seq, item))
def get_priority(self, timeout=None):
try:
priority, seq, item = self.pq.get(timeout=timeout)
return priority, item
except queue.Empty:
print(f"获取超时，队列为空")
raise
def get_nowait(self):
"""非阻塞获取"""
return self.get(timeout=0)
def get(self, timeout=None):
try:
priority, seq, item = self.pq.get(timeout=timeout)
# print(f"取出: 优先级={priority}, 序号={seq}, 项目={item}")
return item
except queue.Empty:
print(f"获取超时，队列为空")
raise # 重新抛出异常，让调用者处理
def empty(self):
return self.pq.empty()
def qsize(self):
return self.pq.qsize()
def clear(self):
"""安全地清空队列"""
try:
while not self.pq.empty():
try:
self.pq.get_nowait()
except queue.Empty:
break
except Exception as e:
print(f"清空队列时出错: {e}")
if __name__ == '__main__':
pq = FIFOPriorityQueue()
pq.put(3, "1优先级任务")
pq.put(0, "第一个0优先级任务")
pq.put(0, "第二个0优先级任务")
pq.put(1, "1优先级任务")
pq.put(0, "第三个0优先级任务")
print(pq.qsize() )
# pq.clear()
while not pq.empty():
priority,task = pq.get_priority(timeout=1)
print(f"优先级: {priority}, 任务: {task}")
# 输出:
# 优先级: 0, 任务: 第一个0优先级任务
# 优先级: 0, 任务: 第二个0优先级任务
# 优先级: 0, 任务: 第三个0优先级任务
# 优先级: 1, 任务: 1优先级任务
