Лучший способ очистить многопроцессорную очередь

Question

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

Какой самый безопасный способ извлечь все элементы из очереди? Я перепробовал несколько методов, и все они, кажется, имеют проблемы.

Метод 1 - неблокирующее получает

packets = []
while True:
    try:
        packets.append(queue.get(block=False))
    except Queue.Empty:
        break

Метод 1 имеет проблему, где queue.get(block=False) Кажется, Queue.Error, даже если в очереди еще остались пакеты. Я думаю, что это происходит, если он не может сразу получить блокировку Конечным эффектом является то, что очередь очищается только частично, в то время как queue.qsize() показывает гораздо больше ожидающих в очереди. Количество пакетов в очереди продолжает расти по мере выполнения программы, а обработка отстает. Я могу обойти эту проблему, опустошая очередь более 4-5 раз в секунду, но я не контролирую скорость поступления пакетов, поэтому я не уверен, что это постоянное решение.

Метод 2 - Получает блокировку

packets = []
while True:
    try:
        packets.append(queue.get(block=True, timeout=0.01))
    except Queue.Empty:
        break

Метод 2 устраняет проблему растущей очереди, но имеет другую проблему, когда очистка очереди занимает относительно длительное время. Я также обеспокоен тем, что l oop может никогда не завершиться, если участвующие процессы всегда сбрасывают пакеты в очередь до истечения времени ожидания.

Каков наилучший способ сделать это?

Answer 1

не могли бы вы сделать что-то вроде

packets = []
while failures < 3:
    try:
        packets.append(queue.get(block=False))
        failures = 0
    except Queue.Empty:
        failures += 1
        time.sleep(0.01)

или еще лучше, вероятно

while not queue.empty():
    try:
        packets.append(queue.get(block=False))
    except Queue.Empty:
        time.sleep(0.01)