Python的通用优先队列

Question

我需要在我的Python代码中使用一个优先队列，并且：

• 正在寻找任何快速实现优先队列
• 最理想的情况是，队列应该是通用的（即适用于具有指定比较运算符的任何对象）。

在寻找高效的东西时，我找到了heapq，但是：

• 我正在寻找比实现在Native Python中的heapq更快的东西，因此它不够快。
• 它看起来很不错，但似乎只适用于整数。我想它可以与具有比较运算符的任何对象一起使用，但它没有指定需要哪些比较运算符。
• 更新：关于heapq中的比较，我可以像Charlie Martin建议的那样使用(priority, object)，或者为我的对象实现__cmp__。

- eliben

---

heapq 实现在 Python 中并不意味着它不快。为什么不直接使用它？只有在无法满足性能需求时才尝试其他替代方案。 - Jingguo Yao

Answer 1

您可以使用Queue.PriorityQueue。

请注意，Python并非强类型语言，因此您可以保存任何您喜欢的内容：只需创建一个元组(priority, thing)即可轻松实现。

- Charlie Martin

---

没有提供查看函数 :-( - Casebash

你是否对这个和heapq做过性能比较？我认为heapq会更快，因为它不需要进行任何锁定。 - Fred Foo

我在Python2.6中进行了一些简单的测试，结果表明heapq大约比PriorityQueue快两倍。 - simonb

Queue.PriorityQueue是同步的。对于不需要同步的情况，它会产生不必要的开销。 - Jingguo Yao

Python 是一种强类型语言，它并非静态、显式地类型化。 - user1277476

Answer 2

使用优先队列时，对于许多算法（Dijkstra算法，A*，OPTICS），减小关键字是必不可少的操作，我想知道为什么Python内置的优先队列不支持此操作。其他答案都没有提供支持此功能的解决方案。

一个支持减小关键字操作的优先队列是由Daniel Stutzbach实现的这个库，它在Python 3.5下完美地工作。

from heapdict import heapdict

hd = heapdict()
hd["two"] = 2
hd["one"] = 1
obj = hd.popitem()
print("object:",obj[0])
print("priority:",obj[1])

# object: one
# priority: 1

- Guy s

---

看起来是个合理的回答：投反对票的人应该在这里解释一下。 - WestCoastProjects

(heapdict)[]的文档中说hd.pop()，但调用heapdict({None: 1}).pop()会出现TypeError: pop() missing 1 required positional argument: 'key'，与常规字典类似。请改用popitem()[0]。 - user66081

文档已经修复 - 错误的 pop 示例已被更改为使用 popitem。 - user2357112

Answer 3

我最终实现了一个heapq的包装器，添加了一个字典来保持队列元素的唯一性。结果对于所有运算符都应该是相当有效的：

class PriorityQueueSet(object):

    """
    Combined priority queue and set data structure.

    Acts like a priority queue, except that its items are guaranteed to be
    unique. Provides O(1) membership test, O(log N) insertion and O(log N)
    removal of the smallest item.

    Important: the items of this data structure must be both comparable and
    hashable (i.e. must implement __cmp__ and __hash__). This is true of
    Python's built-in objects, but you should implement those methods if you
    want to use the data structure for custom objects.
    """

    def __init__(self, items=[]):
        """
        Create a new PriorityQueueSet.

        Arguments:
            items (list): An initial item list - it can be unsorted and
                non-unique. The data structure will be created in O(N).
        """
        self.set = dict((item, True) for item in items)
        self.heap = self.set.keys()
        heapq.heapify(self.heap)

    def has_item(self, item):
        """Check if ``item`` exists in the queue."""
        return item in self.set

    def pop_smallest(self):
        """Remove and return the smallest item from the queue."""
        smallest = heapq.heappop(self.heap)
        del self.set[smallest]
        return smallest

    def add(self, item):
        """Add ``item`` to the queue if doesn't already exist."""
        if item not in self.set:
            self.set[item] = True
            heapq.heappush(self.heap, item)

- eliben

---

看起来不错，但你应该使用"item in set"而不是"set.has_key(item)"。 前者更快（方法调用开销较小），而后者已在Python 3.0中被移除。 - Kiv

items=[] 这种写法不太好，因为列表是可变的。另外，在 __init__() 中可以使用 self.set=set(items)。 - Elazar

看起来比文档中提供的实现更干净。 - alecxe

@alecxe 我认为这是因为它不支持 decrease-key，而 decrease-key 是那些文档中整个 "REMOVED" 概念的原因（而正是这种 "REMOVED" 逻辑使函数看起来不太干净）。 - tscizzle

Answer 4

你可以使用heapq对非整数元素（元组）进行排序：

import heapq

heap = []
data = [(10,"ten"), (3,"three"), (5,"five"), (7,"seven"), (9, "nine"), (2,"two")]
for item in data:
    heapq.heappush(heap, item)
sorted_data = []
while heap:
    sorted_data.append(heapq.heappop(heap))
print(sorted_data)
data.sort()
print(data == sorted_data)

与置顶答案中推荐的queue.PriorityQueue选项相比，这将更快，而且与 queue.PriorityQueue 不同，heapq 在尝试从空堆中弹出时不会永远挂起。

- Adrian

---

这是一个好的方法，使用它时，有用的是添加一个虚拟计数器（每次调用heappush时始终增加为元组的第二个元素），以便当两个条目具有相同的优先级（意味着元组的第一个元素相等）时，按照它们被添加的顺序对元组进行排序。这在边缘情况下为优先队列提供了预期的结果。 - David Parks

Answer 5

你看过heapq页面上的“显示源代码”链接了吗？在该页面下方稍微往下找一点，有一个例子演示了如何使用元组（包含整数和字符）列表作为优先队列。

- Harper Shelby

---

我认错了（由Benjamin Peterson提出）。heapq使用C实现，速度很快。 - Eli Bendersky

Answer 6

我没有使用过它，但你可以尝试使用PyHeap。它是用C语言编写的，所以希望它足够快。

你确定heapq/PriorityQueue不够快吗？最好先使用其中之一并进行性能分析，看看它是否真的是你的性能瓶颈。

- Kiv

Answer 7

我正在使用 queue.PriorityQueue 在 Python 3 中实现一个 优先级队列，代码如下：

from queue import PriorityQueue

class PqElement(object):
    def __init__(self, value: int):
        self.val = value

    #Custom Compare Function (less than or equsal)
    def __lt__(self, other):
        """self < obj."""
        return self.val > other.val

    #Print each element function
    def __repr__(self):
        return f'PQE:{self.val}'

#Usage-
pq = PriorityQueue()
pq.put(PqElement(v))       #Add Item      - O(Log(n))
topValue = pq.get()        #Pop top item  - O(1)
topValue = pq.queue[0].val #Get top value - O(1)

- Abrar Jahin

---

似乎是一个最大队列而不是最小队列？如果不是，你能解释一下这里的极性吗？谢谢。 - WestCoastProjects

你说得对@WestCoastProjects，我已经提供了我的实现，以便任何人都可以根据需要进行修改。如果你把__lt__函数从 self.val > other.val 改成 self.val < other.val , 那么它就会变成最小队列。 - Abrar Jahin

Answer 8

这是高效的方法，适用于字符串或任何类型的输入 - :)

import itertools
from heapq import heappush, heappop

pq = []                         # list of entries arranged in a heap
entry_finder = {}               # mapping of tasks to entries
REMOVED = '<removed-task>'      # placeholder for a removed task
counter = itertools.count()     # unique sequence count

def add_task(task, priority=0):
    'Add a new task or update the priority of an existing task'
    if task in entry_finder:
        remove_task(task)
    count = next(counter)
    entry = [priority, count, task]
    entry_finder[task] = entry
    heappush(pq, entry)

def remove_task(task):
    'Mark an existing task as REMOVED.  Raise KeyError if not found.'
    entry = entry_finder.pop(task)
    entry[-1] = REMOVED

def pop_task():
    'Remove and return the lowest priority task. Raise KeyError if empty.'
    while pq:
        priority, count, task = heappop(pq)
        if task is not REMOVED:
            del entry_finder[task]
            return task
    raise KeyError('pop from an empty priority queue')

Reference: http://docs.python.org/library/heapq.html

- codersofthedark

---

保持与4月1日发布的答案一致，使用object()作为哨兵/删除。 - Dima Tisnek

Answer 9

我在https://pypi.python.org/pypi/fibonacci-heap-mod使用了一个优先队列/斐波那契堆。

它不够快（删除最小值时有较大的常数c，复杂度为O(c*logn)）。但查找最小值、插入、降低键值和合并都是O(1) - 换句话说，它是懒惰的。

如果在CPython上速度太慢，可以尝试Pypy、Nuitka甚至CPython+Numba :)

- user1277476

Answer 10

一个简单的实现：

由于PriorityQueue是低优先级优先，因此...

from queue import PriorityQueue


class PriorityQueueWithKey(PriorityQueue):
    def __init__(self, key=None, maxsize=0):
        super().__init__(maxsize)
        self.key = key

    def put(self, item):
        if self.key is None:
            super().put((item, item))
        else:
            super().put((self.key(item), item))

    def get(self):
        return super().get(self.queue)[1]


a = PriorityQueueWithKey(abs)
a.put(-4)
a.put(-3)
print(*a.queue)

- AsukaMinato