Очередь — класс синхронизированной очереди

Исходный код: Lib/Queue.py

Модуль Queue реализует очереди с множеством производителей и потребителей. Он особенно полезен в многопоточной программировании, когда необходимо безопасно обмениваться информацией между несколькими потоками. Класс Queue в этом модуле реализует все необходимые семантики блокировки. Он зависит от наличия поддержки потоков в Python; см. модуль threading.

Модуль реализует три типа очереди, которые различаются только порядком извлечения элементов. В очереди FIFO (первым вошел — первым вышел) первые добавленные задачи извлекаются первыми. В очереди LIFO (последним вошел — первым вышел) последняя добавленная запись извлекается первой (работает как стек). В очереди с приоритетами записи хранятся в отсортированном порядке (используя модуль heapq), и первая извлекается запись с наименьшим значением.

Модуль Queue определяет следующие классы и исключения:

class Queue.Queue(maxsize=0)

Конструктор для очереди FIFO. maxsize — целое число, устанавливающее верхнюю границу количества элементов, которые могут быть помещены в очередь. Вставка будет блокироваться, когда этот размер будет достигнут, до тех пор, пока элементы очереди не будут извлечены. Если maxsize меньше или равно нулю, размер очереди является бесконечным.

class Queue.LifoQueue(maxsize=0)

Конструктор для очереди LIFO. maxsize — целое число, устанавливающее верхнюю границу количества элементов, которые могут быть помещены в очередь. Вставка будет блокироваться, когда этот размер будет достигнут, до тех пор, пока элементы очереди не будут извлечены. Если maxsize меньше или равно нулю, размер очереди является бесконечным.

Новое в версии 2.6.

class Queue.PriorityQueue(maxsize=0)

Конструктор для очереди с приоритетами. maxsize — целое число, устанавливающее верхнюю границу количества элементов, которые могут быть помещены в очередь. Вставка будет блокироваться, когда этот размер будет достигнут, до тех пор, пока элементы очереди не будут извлечены. Если maxsize меньше или равно нулю, размер очереди является бесконечным.

Элементы с наименьшими значениями извлекаются первыми (элемент с наименьшим значением — это тот, который возвращается методом sorted(list(entries))[0]). Типичный шаблон для записей — кортеж в форме: (priority_number, data).

Новое в версии 2.6.

exception Queue.Empty

Исключение, которое возникает, когда неблокирующий метод get() (или get_nowait()) вызывается для объекта Queue, который пуст.

exception Queue.Full

Исключение, которое возникает, когда неблокирующий метод put() (или put_nowait()) вызывается для объекта Queue, который заполнен.

1. Объекты очереди

Объекты очереди (Queue, LifoQueue или PriorityQueue) предоставляют общедоступные методы, описанные ниже.

Queue.qsize()

Возвращает приблизительный размер очереди. Обратите внимание, что qsize() > 0 не гарантирует, что последующий вызов get() не будет заблокирован, а qsize() < maxsize не гарантирует, что put() не будет заблокирован.

Queue.empty()

Возвращает True если очередь пуста, False в противном случае. Если empty() возвращает True, это не гарантирует, что последующий вызов put() не будет заблокирован. Аналогично, если empty() возвращает False, это не гарантирует, что последующий вызов get() не будет заблокирован.

Queue.full()

Возвращает True если очередь заполнена, False в противном случае. Если full() возвращает True, это не гарантирует, что последующий вызов get() не будет заблокирован. Аналогично, если full() возвращает False, это не гарантирует, что последующий вызов put() не будет заблокирован.

Queue.put(item[, block[, timeout]])

Поместить item в очередь. Если необязательные аргументы block истинны и timeout равен None (по умолчанию), блокировать, если необходимо, пока не освободится место. Если timeout — положительное число, блокировать не более timeout секунд и вызывать исключение Full, если свободного места не было в течение этого времени. В противном случае (block ложно), поместить элемент в очередь, если свободное место есть сразу же, иначе вызвать исключение Full (timeout игнорируется в этом случае).

Новое в версии 2.3: Параметр timeout.

Queue.put_nowait(item)

Эквивалентно put(item, False).

Queue.get([block[, timeout]])

Удалить и вернуть элемент из очереди. Если необязательные аргументы block истинны и timeout равен None (по умолчанию), блокировать, если необходимо, пока элемент не появится. Если timeout — положительное число, блокировать не более timeout секунд и вызывать исключение Empty, если элемент не появился в течение этого времени. В противном случае (block ложно), вернуть элемент, если он доступен немедленно, иначе вызвать исключение Empty (timeout игнорируется в этом случае).

Новое в версии 2.3: Параметр timeout.

Queue.get_nowait()

Эквивалентно get(False).

Два метода предлагаются для отслеживания того, были ли задачи, помещенные в очередь, полностью обработаны демоновыми потоками-потребителями.

Queue.task_done()

Указывает, что задача, ранее помещенная в очередь, завершена. Используется потоками-потребителями очереди. Для каждого вызова get() для получения задачи последующий вызов task_done() сообщает очереди, что обработка задачи завершена.

Если вызов join() в настоящее время заблокирован, он возобновится, когда все элементы будут обработаны (что означает, что вызов task_done() был получен для каждого элемента, который был помещен в очередь с помощью put()).

Вызывает исключение ValueError, если вызов был сделан больше раз, чем было помещено элементов в очередь.

Новое в версии 2.5.

Queue.join()

Блокируется, пока все элементы в очереди не будут извлечены и обработаны.

Счетчик незавершенных задач увеличивается всякий раз, когда элемент добавляется в очередь. Счетчик уменьшается, когда поток-потребитель вызывает task_done(), чтобы указать, что элемент был получен и вся работа с ним завершена. Когда счетчик незавершенных задач уменьшается до нуля, вызов join() разблокируется.

Новое в версии 2.5.

Пример того, как дождаться завершения задач в очереди:

def worker():
    while True:
        item = q.get()
        do_work(item)
        q.task_done()

q = Queue()
for i in range(num_worker_threads):
     t = Thread(target=worker)
     t.daemon = True
     t.start()

for item in source():
    q.put(item)

q.join()       # block until all tasks are done