如何在 Python 中创建唯一值优先级队列？

Posted 2023-02-16

【中文标题】如何在 Python 中创建唯一值优先级队列？

【英文标题】：How can I make a unique value priority queue in Python?

【发布时间】：2011-05-13 20:05:43

【问题描述】：

Python 有 Queue.PriorityQueue，但我看不到让其中的每个值唯一的方法，因为没有检查值是否已经存在的方法（如 find(name) 或类似的）。此外，PriorityQueue 需要优先级保持在值内，所以我什至无法搜索我的值，因为我还必须知道优先级。您将使用 (0.5, myvalue) 作为 PriorityQueue 中的值，然后它将按元组的第一个元素排序。

另一方面，collections.deque 类确实提供了一个函数来检查一个值是否已经存在并且使用起来更加自然（没有锁定，但仍然是原子的），但它不提供按优先级排序的方法.

*** 上还有其他一些使用 heapq 的实现，但 heapq 也在值内使用优先级（例如，在元组的第一个位置），因此对于比较现有值似乎不是很好。

Creating a python priority Queue

https://***.com/questions/3306179/priority-queue-problem-in-python

创建具有唯一值的原子优先级队列（=可以从多个线程中使用）的最佳方法是什么？

我想添加的示例：

优先级：0.2，值：value1 优先级：0.3，值：value2 优先级：0.1，值：value3（应首先自动检索） 优先级：0.4，值：value1（不应再次添加，即使它具有不同的优先级）

【参考方案1】：

您可以将优先级队列与集合结合起来：

import heapq

class PrioritySet(object):
    def __init__(self):
        self.heap = []
        self.set = set()

    def add(self, d, pri):
        if not d in self.set:
            heapq.heappush(self.heap, (pri, d))
            self.set.add(d)

    def get(self):
        pri, d = heapq.heappop(self.heap)
        self.set.remove(d)
        return d

这将使用您在其中一个链接问题中指定的优先级队列。我不知道这是否是您想要的，但是通过这种方式将集合添加到任何类型的队列中都相当容易。

【讨论】：

我建议不要使用像self.set这样的内置函数名称

也许 pop 是 get 的更好名称 :)

【参考方案2】：

这是一种方法。我基本上从他们如何在 Queue.py 中定义 PriorityQueue 开始，并在其中添加了一个集合来跟踪唯一键：

from Queue import PriorityQueue
import heapq

class UniquePriorityQueue(PriorityQueue):
    def _init(self, maxsize):
#        print 'init'
        PriorityQueue._init(self, maxsize)
        self.values = set()

    def _put(self, item, heappush=heapq.heappush):
#        print 'put',item
        if item[1] not in self.values:
            print 'uniq',item[1]
            self.values.add(item[1])
            PriorityQueue._put(self, item, heappush)
        else:
            print 'dupe',item[1]

    def _get(self, heappop=heapq.heappop):
#        print 'get'
        item = PriorityQueue._get(self, heappop)
#        print 'got',item
        self.values.remove(item[1])
        return item

if __name__=='__main__':
    u = UniquePriorityQueue()

    u.put((0.2, 'foo'))
    u.put((0.3, 'bar'))
    u.put((0.1, 'baz'))
    u.put((0.4, 'foo'))

    while not u.empty():
        item = u.get_nowait()
        print item

Boaz Yaniv 领先我几分钟，但我想我也会发布我的，因为它支持 PriorityQueue 的完整界面。我留下了一些未注释的打印语句，但注释掉了我在调试时输入的语句。 ;)

【讨论】：

感谢您的回答。还不知道 _init、_put 和 _get 方法，但是它们在扩展队列时非常实用。由于你们都使用过套装，我现在确信这些是正确的方法；）

【参考方案3】：

如果您想稍后确定任务的优先级。

u = UniquePriorityQueue()

u.put((0.2, 'foo'))
u.put((0.3, 'bar'))
u.put((0.1, 'baz'))
u.put((0.4, 'foo'))
# Now `foo`'s priority is increased.
u.put((0.05, 'foo'))

这是另一个遵循官方指南的实现：

import heapq
import Queue

class UniquePriorityQueue(Queue.Queue):
    """
    - https://github.com/python/cpython/blob/2.7/Lib/Queue.py
    - https://docs.python.org/3/library/heapq.html
    """

    def _init(self, maxsize):
        self.queue = []
        self.REMOVED = object()
        self.entry_finder = 

    def _put(self, item, heappush=heapq.heappush):
        item = list(item)
        priority, task = item
        if task in self.entry_finder:
            previous_item = self.entry_finder[task]
            previous_priority, _ = previous_item
            if priority < previous_priority:
                # Remove previous item.
                previous_item[-1] = self.REMOVED
                self.entry_finder[task] = item
                heappush(self.queue, item)
            else:
                # Do not add new item.
                pass
        else:
            self.entry_finder[task] = item
            heappush(self.queue, item)

    def _qsize(self, len=len):
        return len(self.entry_finder)

    def _get(self, heappop=heapq.heappop):
        """
        The base makes sure this shouldn't be called if `_qsize` is 0.
        """
        while self.queue:
            item = heappop(self.queue)
            _, task = item
            if task is not self.REMOVED:
                del self.entry_finder[task]
                return item
        raise KeyError('It should never happen: pop from an empty priority queue')

【参考方案4】：

我喜欢@Jonny Gaines Jr. 的回答，但我认为它可以简化。 PriorityQueue 使用底层列表，因此您可以定义：

class PrioritySetQueue(PriorityQueue):
    def _put(self, item):
        if item not in self.queue:
            super(PrioritySetQueue, self)._put(item)

【讨论】：

我认为 if 条件应该是：if item not in map(lambda x:x[1], self.queue):