Как реализовать многопроцессорную приоритетную очередь в Python?

Увы, нигде не так просто, как изменить дисциплину организации очередей старого доброго Queue.Queue : последний на самом деле предназначен для разделения на подклассы в соответствии с шаблоном шаблонного метода , и переопределение только методов перехвата _put и / или _get может легко позволить изменить дисциплину организации очередей (в 2.6 предлагаются явные реализации LIFO и приоритета, но их было легко сделать даже в более ранние версии Python).

В общем случае для многопроцессорной обработки (несколько читателей, несколько писателей) я не вижу решения, как реализовать приоритетные очереди, кроме как отказаться от распределенной природы очереди; назначить один специальный вспомогательный процесс, который ничего не делает, кроме обработки очередей, отправлять ему (по существу) RPC для создания очереди с заданной дисциплиной, ставит и добирается до него, получает информацию об этом и т. д. Таким образом, возникают обычные проблемы с обеспечением того, чтобы каждый процесс знал о местонахождении вспомогательного процесса (например, хост и порт) и т. Д. (Проще, если процесс всегда запускается при запуске основным процессом). Довольно большая проблема, особенно если кто-то хочет сделать это с хорошей производительностью, защитой от сбоев вспомогательного процесса (требующих репликации данных на подчиненные процессы, распределения «главных выборов» среди подчиненных в случае сбоя мастера и т. Д.) И так далее. Выполнение этого с нуля похоже на работу доктора философии. Можно начать с работы Джонсона или использовать какой-то очень общий подход, такой как ActiveMQ .

Некоторые особые случаи (например, одиночный читатель, одиночный писатель) могут быть проще и в итоге могут оказаться быть быстрее для своей ограниченной области применения;

While this isn't an answer, maybe it can help you develop an multiprocessing queue.

Here's a simple priority queue class I wrote using Python's 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

Существует ошибка, препятствующая истинному FIFO.
Прочтите здесь .

Альтернативный способ создания многопроцессорной настройки очереди с приоритетом был бы отличным вариантом!

В зависимости от ваших требований вы можете использовать операционную систему и файловую систему в нескольких способов. Насколько большая очередь будет расти и как быстро она должна быть? Если очередь может быть большой, но вы готовы открывать пару файлов для каждого доступа к очереди, вы можете использовать реализацию BTree для хранения очереди и блокировки файлов для обеспечения монопольного доступа. Медленный, но надежный.

Если очередь останется относительно небольшой, и вам нужно, чтобы она работала быстро, вы можете использовать разделяемую память в некоторых операционных системах ...

Если очередь будет небольшой (тысячи записей), а вы этого не сделаете. мне не нужно чтобы быть действительно быстрым, вы могли бы использовать что-нибудь так же просто, как каталог с файлами, содержащими данные с блокировкой файлов. Я бы предпочел, если маленькая и медленная - это нормально.

Если очередь может быть большой и вы хотите, чтобы она была быстрой в среднем, то вам, вероятно, следует использовать процесс выделенного сервера, как предлагает Алекс. Однако это боль в шее.

Каковы ваши требования к характеристикам и размеру?

У меня был такой же вариант использования. Но с конечным числом приоритетов.

В конечном итоге я создаю одну очередь для каждого приоритета, и мои рабочие процессы будут пытаться получить элементы из этих очередей, начиная с самой важной очереди к менее важной ( переход из одной очереди в другую выполняется, когда очередь пуста)