有没有办法检查 Python 中的优先级队列中是不是存在优先级？

Posted 2023-02-16

【中文标题】有没有办法检查 Python 中的优先级队列中是不是存在优先级？

【英文标题】：Is there a way to check if a priority exists in a priority queue in Python?有没有办法检查 Python 中的优先级队列中是否存在优先级？

【发布时间】：2021-02-09 18:36:55

【问题描述】：

我对 Python 比较陌生。我正在尝试使用 queue.PriorityQueue() class 并有一个问题。有没有办法检查特定优先级是否存在？

例如，我有以下(priority, element)：

(0,A), (1,B), (2,C), (2,D)

有没有办法检查优先级2 是否存在？ （是的，因为 C 和 D 都是 2。） 我可以获取特定优先级的元素吗？所以，如果我想要优先级为2 的元素，它会给我C 和D？

我在优先级队列上看到的唯一官方文档是：Priority Queue official documentation

还有关于这个类的其他文档吗？喜欢我可以使用的方法吗？可用的类/字段的结构？

【问题讨论】：

期望PriorityQueue 支持这不是合理的事情，因为它不知道任何给定元素的优先级是什么 - 它只关心它们如何相互比较。

您需要此功能是否有特定原因？据我所知，感兴趣的方法是.put()、.get()，以及Queue 对象具有的任何其他方法。如所见docs.python.org/3/library/queue.html#queue-objects

@Axe319，我正在创建一个基于特定值 (v) 的优先级队列。如果我有 2 个具有相同 v 的元素，我希望它们为该优先级 v 形成一个列表。我尝试了这个，我得到了相同 v 的多个条目。如果我可以检查优先级，我可以自己创建列表。跨度>

【参考方案1】：

我的理解是，您的最终目标是将具有相同优先级的任务放在列表中并一起返回。所以：

如果您查看queue.PriorityQueue 的代码（我看过），您会发现它基于heapq 模块的heappush 和heappop 方法，该模块实现了堆队列算法.见heapq。如果您在此页面上进一步查看，它们甚至显示了如何使用 heapq 来实现优先级队列。此实现比您需要的要复杂一些，因为它支持更改已添加任务的优先级的能力，并且它不能完全处理具有相同优先级的多个任务。但这些改变很容易做到：

from heapq import heappush, heappop

class PriorityQueue:
    def __init__(self):
        self._pq = [] # list of entries arranged in a heap
        self._priority_finder =  # mapping of priority to entries

    def add_task(self, task, priority=0):
        'Add a new task'
        # any tasks with this priority?
        entry = self._priority_finder.get(priority)
        if entry:
            entry[1].append(task)
        else:
            entry = [priority, [task]]
            self._priority_finder[priority] = entry
            heappush(self._pq, entry)

    def pop_task(self):
        'Remove and return the lowest priority tasks. Raise KeyError if empty.'
        if not self._pq:
            raise KeyError('pop from an empty priority queue')
        priority, tasks = heappop(self._pq)
        del self._priority_finder[priority]
        return priority, tasks

    def __bool__(self):
        'return True if any tasks on the queue'
        return True if self._pq else False

pq = PriorityQueue()
pq.add_task('a', 4) # task 'a' with priority 4
pq.add_task('b', 2)
pq.add_task('c', 4)
pq.add_task('d', 2)
pq.add_task('e', 1)
while pq:
    print(pq.pop_task())

打印：

(1, ['e'])
(2, ['b', 'd'])
(4, ['a', 'c'])

【讨论】：

谢谢@Booboo！如果你不介意我问，这段代码在官方文档中的哪里？

非常感谢！

我刚刚意识到我错过了函数pop_task 中的一个重要del self._priority_finder[priority] 语句。我已经相应地更新了答案。对此感到抱歉。