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

Question

有谁熟悉我如何在 python 中实现多处理优先级队列？

Answer 1

唉，这绝不是像改变旧的 Queue.Queue 的排队规则那么简单：后者实际上是根据模板方法模式设计的子类，并且只覆盖挂钩方法 _put 和/或_get 可以轻松地允许更改排队规则（在 2.6 中提供了显式 LIFO 和优先级实现，但即使在早期版本的 Python 中也很容易实现）。

对于多处理，在一般情况下（多个读取器，多个写入器），除了放弃队列的分布式特性外，我没有看到如何实现优先级队列的解决方案；指定一个特殊的辅助进程，它除了处理队列之外什么都不做，向它发送（本质上）RPC 以创建具有指定规则的队列，执行 put 和 get 操作，获取有关它的信息等等。因此，在确保每个进程都知道 aux proc 的位置（例如主机和端口）等方面会遇到常见的问题（如果进程总是在主 proc 启动时产生，则更容易）。一个相当大的问题，特别是如果想要以良好的性能做到这一点，可以防止 aux proc 的崩溃（需要将数据复制到从属进程，如果主崩溃，则在从属之间分布式“主选举”等）等等。从头开始听起来像是博士的工作。可以从Johnson's 工作开始，或者采用一些非常通用的方法，例如ActiveMQ。

一些特殊情况（例如，单个阅读器、单个编写器）可能更容易，并且由于其有限的应用领域而变得更快；但随后应该为该有限区域制定一个非常具体的受限规范——结果不会构成（通用）“多处理队列”，而是仅适用于给定的一组受限需求。

Answer 2

有一个错误会阻止真正的 FIFO。
阅读here。

另一种构建优先级队列多处理设置的方法肯定很棒！

Answer 3

虽然这不是一个答案，但也许它可以帮助您开发一个多处理队列。

这是我使用 Python 的 Array 编写的一个简单的优先级队列类：

class PriorityQueue():
    """A basic priority queue that dequeues items with the smallest priority number."""
    def __init__(self):
        """Initializes the queue with no items in it."""
        self.array = []
        self.count = 0

    def enqueue(self, item, priority):
        """Adds an item to the queue."""
        self.array.append([item, priority])
        self.count += 1

    def dequeue(self):
        """Removes the highest priority item (smallest priority number) from the queue."""
        max = -1
        dq = 0
        if(self.count > 0):
            self.count -= 1

            for i in range(len(self.array)):
                if self.array[i][1] != None and self.array[i][1] > max:
                    max = self.array[i][1]

            if max == -1:
                return self.array.pop(0)
            else:
                for i in range(len(self.array)):
                    if self.array[i][1] != None and self.array[i][1] <= max:
                        max = self.array[i][1]
                        dq = i
                return self.array.pop(dq)

    def requeue(self, item, newPrio):
        """Changes specified item's priority."""
        for i in range(len(self.array)):
            if self.array[i][0] == item:
                self.array[i][1] = newPrio
                break

    def returnArray(self):
        """Returns array representation of the queue."""
        return self.array

    def __len__(self):
        """Returnes the length of the queue."""
        return self.count

Answer 4

我有同样的用例。但优先级有限。

我最终要做的是为每个优先级创建一个队列，我的流程工作人员将尝试从这些队列中获取项目，从最重要的队列开始到不太重要的队列（从一个队列移动到另一个队列是当队列为空时完成）

Answer 5

根据您的要求，您可以通过多种方式使用操作系统和文件系统。队列将增长多大，它必须有多快？如果队列可能很大，但您愿意为每个队列访问打开几个文件，您可以使用 BTree 实现来存储队列和文件锁定以强制执行独占访问。缓慢但稳健。

如果队列将保持相对较小并且您需要它更快，您也许可以在某些操作系统上使用共享内存...

如果队列很小（1000 个条目）并且您不需要 它真的很快，你可以使用一些东西 就像一个包含文件的目录一样简单，该目录包含文件锁定的数据。如果小而慢是可以的，这将是我的偏好。

如果队列可能很大并且您希望它平均速度快，那么您可能应该使用 Alex 建议的专用服务器进程。然而，这是一种颈部疼痛。

您的性能和尺寸要求是什么？

Answer 6

受@user211505 建议的启发，我整理了一些又快又脏的东西。

请注意，这并不是多处理生产环境中优先级队列难题的完整解决方案。但是，如果您只是在闲逛或需要一些东西来完成一个短期项目，那么这可能适合您：

from time import sleep
from datetime import datetime
from Queue import Empty
from multiprocessing import Queue as ProcessQueue

class SimplePriorityQueue(object):
    '''
    Simple priority queue that works with multiprocessing. Only a finite number 
    of priorities are allowed. Adding many priorities slow things down. 

    Also: no guarantee that this will pull the highest priority item 
    out of the queue if many items are being added and removed. Race conditions
    exist where you may not get the highest priority queue item.  However, if 
    you tend to keep your queues not empty, this will be relatively rare.
    '''
    def __init__(self, num_priorities=1, default_sleep=.2):
        self.queues = []
        self.default_sleep = default_sleep
        for i in range(0, num_priorities):
            self.queues.append(ProcessQueue())

    def __repr__(self):
        return "<Queue with %d priorities, sizes: %s>"%(len(self.queues), 
                    ", ".join(map(lambda (i, q): "%d:%d"%(i, q.qsize()), 
                                enumerate(self.queues))))

    qsize = lambda(self): sum(map(lambda q: q.qsize(), self.queues))

    def get(self, block=True, timeout=None):
        start = datetime.utcnow()
        while True:
            for q in self.queues:
                try:
                    return q.get(block=False)
                except Empty:
                    pass
            if not block:
                raise Empty
            if timeout and (datetime.utcnow()-start).total_seconds > timeout:
                raise Empty

            if timeout:
                time_left = (datetime.utcnow()-start).total_seconds - timeout
                sleep(time_left/4)
            else:
                sleep(self.default_sleep)

    get_nowait = lambda(self): self.get(block=False)

    def put(self, priority, obj, block=False, timeout=None):
        if priority < 0 or priority >= len(self.queues):
            raise Exception("Priority %d out of range."%priority)
        # Block and timeout don't mean much here because we never set maxsize
        return self.queues[priority].put(obj, block=block, timeout=timeout)