Python利用heapq实现优先级队列的方法

陆遥

发布时间：2026-10-12 | 3人浏览过

原创

本文详解如何利用Python内置heapq模块构建优先级队列。通过取反优先级数值适配最小堆特性，并结合递增索引元组解决同优先级元素的排序冲突。代码涵盖基础实现、并发安全注意事项及时间复杂度分析，提供可直接复用的完整类定义。

基于最小堆实现最大优先级队列

Python标准库中的heapq模块默认维护的是最小堆（Min-Heap），即每次弹出队列的永远是当前堆中最小的元素。若要实现“数值越大优先级越高”的最大优先级队列，只需在存储时将优先级取负数即可。这样最小的负数对应最大的原始优先级，完美契合heappush与heappop的行为逻辑。

from heapq import heappush, heappop

class PriorityQueue:
    def __init__(self):
        self._queue = []

    def put(self, item, priority):
        heappush(self._queue, (-priority, item))

    def get(self):
        return heappop(self._queue)[1]

q = PriorityQueue()
q.put('world', 1)
q.put('hello', 2)
print(q.get())
print(q.get())

引入自增索引解决同优先级排序冲突

上述基础版本在元素可比较时运行正常，但当两个对象的优先级相同时，若对象本身未实现比较运算符（如未定义__lt__），Python在对比(priority, item)元组时会抛出TypeError。例如：

>>> a = (1, 'foo')
>>> b = (1, 'bar')
>>> a < b
Traceback (most recent call last):
...
TypeError: ' < ' not supported between instances of 'str' and 'str'

修复该问题的核心在于保证元组的第三个字段具有唯一性且不依赖对象本身的比较能力。引入一个不断递增的整数索引_index，将存储结构升级为三元组(-priority, _index, item)。Python进行元组比较时会从左至右逐项判断，一旦前两项确定顺序（或第一项相同而第二项不同），后续的比较便不会触发，从而彻底绕过item的比较逻辑。同时，严格递增的索引值天然保证了先进先出（FIFO）的稳定性。

import heapq

class PriorityQueue:
    def __init__(self):
        self._queue = []
        self._index = 0

    def push(self, item, priority):
        heapq.heappush(self._queue, (-priority, self._index, item))
        self._index += 1

    def pop(self):
        return heapq.heappop(self._queue)[-1]


class Item:
    def __init__(self, name):
        self.name = name
    def __repr__(self):
        return f'Item({self.name!r})'

q = PriorityQueue()
q.push(Item('foo'), 1)
q.push(Item('bar'), 5)
q.push(Item('spam'), 4)
q.push(Item('grok'), 1)
print(q.pop())  # Item('bar')
print(q.pop())  # Item('spam')
print(q.pop())  # Item('foo')
print(q.pop())  # Item('grok')

多线程环境的同步处理

当优先级队列需要在多个线程间共享时，直接调用push/pop方法会引发竞态条件，导致堆结构损坏或数据丢失。此时必须包裹线程安全机制，推荐使用threading.Lock对出入队操作进行临界区保护：

import threading

lock = threading.Lock()

def safe_push(queue, item, priority):
    with lock:
        queue.push(item, priority)

def safe_pop(queue):
    with lock:
        return queue.pop()