如何在 Python 中实现优先级队列？

Posted 2023-02-21

【中文标题】如何在 Python 中实现优先级队列？

【英文标题】：How to implement Priority Queues in Python?

【发布时间】：2012-04-15 16:31:51

【问题描述】：

很抱歉提出这么愚蠢的问题，但 Python 文档令人困惑......

链接 1：队列实现 http://docs.python.org/library/queue.html

它说 Queue 有一个优先级队列的类。但我找不到如何实现它。

class Queue.PriorityQueue(maxsize=0)

链接 2：堆实现 http://docs.python.org/library/heapq.html

这里他们说我们可以使用 heapq 间接实现优先级队列

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'

Python 中最有效的优先级队列实现是什么？以及如何实现？

【问题讨论】：

A generic priority queue for Python的可能重复

【参考方案1】：

在任何语言中都没有“最有效的优先级队列实现”之类的东西。

优先级队列是关于权衡的。见http://en.wikipedia.org/wiki/Priority_queue

您应该根据您打算如何使用它来选择这两者之一：

O(log(N))插入时间和O(1)findMin+deleteMin时间，或者 O(1)插入时间和O(log(N))findMin+deleteMin时间

在后一种情况下，您可以选择使用斐波那契堆实现优先级队列：http://en.wikipedia.org/wiki/Heap_(data_structure)#Comparison_of_theoretic_bounds_for_variants（如您所见，heapq 基本上是一棵二叉树，必须有 O(log(N)) 用于插入和findMin+deleteMin)

如果是处理特殊属性的数据（比如有界数据），那么可以实现O(1)插入和O(1)findMin+deleteMin时间。您只能对某些类型的数据执行此操作，否则您可能会滥用优先级队列来违反排序时的O(N log(N)) 约束。

要以任何语言实现任何队列，您只需定义insert(value) 和extractMin() -> value 操作即可。这通常只涉及底层堆的最小包装；请参阅http://en.wikipedia.org/wiki/Fibonacci_heap 来实现您自己的，或使用类似堆的现成库，如配对堆（Google 搜索显示http://svn.python.org/projects/sandbox/trunk/collections/pairing_heap.py）

---

如果您只关心您引用的两个中哪一个更有效（上面包含的 http://docs.python.org/library/heapq.html#priority-queue-implementation-notes 中基于 heapq 的代码，与 Queue.PriorityQueue 相比），那么：

对于Queue.PriorityQueue 实际在做什么，网络上似乎没有任何容易找到的讨论；您必须深入研究代码，该代码链接到帮助文档：http://hg.python.org/cpython/file/2.7/Lib/Queue.py

   224     def _put(self, item, heappush=heapq.heappush):
   225         heappush(self.queue, item)
   226 
   227     def _get(self, heappop=heapq.heappop):
   228         return heappop(self.queue)

正如我们所见，Queue.PriorityQueue 也使用heapq 作为底层机制。因此它们同样糟糕（渐近地说）。 Queue.PriorityQueue 可能允许并行查询，所以我敢打赌，它可能会产生更多的开销。但是因为您知道底层实现（和渐近行为）必须相同，所以最简单的方法就是在同一个大型数据集上运行它们。

(请注意Queue.PriorityQueue 似乎没有删除条目的方法，而heapq 有。但是这是一把双刃剑：良好的优先级队列实现可能允许您删除 O( 1) 或 O(log(N)) 时间，但如果您使用您提到的 remove_task 函数，并让那些僵尸任务累积在您的队列中，因为您没有从最小值中提取它们，那么您将看到渐近减速否则你不会看到。当然，你不能用Queue.PriorityQueue首先做到这一点，所以这里无法进行比较。）

【讨论】：

我从理论上很好地理解了优先级队列，因此可能的 DS。但问题在于它在 Python 中的实现，它具有非常不同的 DS 集。

谢谢@ninjagecko .. 你提出了一个很好的理论，人们必须知道它才能决定正确的 DS .. -:)

【参考方案2】：

Queue 模块中的版本是implemented 使用heapq 模块，因此它们对于底层堆操作具有相同的效率。

也就是说，Queue 版本速度较慢，因为它添加了锁、封装和一个不错的面向对象的 API。

priority queue suggestions shown in the heapq docs 旨在展示如何向优先级队列添加其他功能（例如排序稳定性和更改先前入队任务的优先级的能力）。如果您不需要这些功能，那么基本的 heappush 和 heappop 功能将为您提供最快的性能。

【参考方案3】：

虽然这个问题已经被回答并被标记为接受，这里仍然是一个简单的优先队列的自定义实现，没有使用任何模块来理解它是如何工作的。

# class for Node with data and priority
class Node:

  def __init__(self, info, priority):
    self.info = info
    self.priority = priority

# class for Priority queue 
class PriorityQueue:

  def __init__(self):
    self.queue = list()
    # if you want you can set a maximum size for the queue

  def insert(self, node):
    # if queue is empty
    if self.size() == 0:
      # add the new node
      self.queue.append(node)
    else:
      # traverse the queue to find the right place for new node
      for x in range(0, self.size()):
        # if the priority of new node is greater
        if node.priority >= self.queue[x].priority:
          # if we have traversed the complete queue
          if x == (self.size()-1):
            # add new node at the end
            self.queue.insert(x+1, node)
          else:
            continue
        else:
          self.queue.insert(x, node)
          return True

  def delete(self):
    # remove the first node from the queue
    return self.queue.pop(0)

  def show(self):
    for x in self.queue:
      print str(x.info)+" - "+str(x.priority)

  def size(self):
    return len(self.queue)

在这里找到完整的代码和解释：https://www.studytonight.com/post/implementing-priority-queue-in-python（更新的 URL）

【讨论】：

以下操作输出错误：pq.insert([8, 'item2']) pq.insert([10, 'item1']) pq.show() pq.insert([12, 'item3']) pq.insert([9, 'item4']) pq.show()