Документация Python неофициальный перевод
Содержание страницы

queue – синхронизированный класс очередиqueue – A synchronized queue class

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

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

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

class queue.Queue(maxsize)
Конструктор очереди FIFO. maxsize – целое число, которое задает верхнюю границу количества элементов, которые можно поместить в очередь. Вставка будет блокироваться, как только этот размер достигнут, пока элементы очереди не будут потреблены. Если maxsize меньше или равно нулю, размер очереди бесконечен.
class queue.LifoQueue(maxsize)
Конструктор очереди LIFO. maxsize – целое число, которое задает верхнюю границу количества элементов, которые можно поместить в очередь. Вставка будет блокироваться, как только этот размер достигнут, пока элементы очереди не будут потреблены. Если maxsize меньше или равно нулю, размер очереди бесконечен.
class queue.PriorityQueue(maxsize)

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

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

exception queue.Empty
Исключение возникает при вызове неблокирующего get() (или get_nowait()) для пустого объекта очереди.
exception queue.Full
Исключение возникает при вызове неблокирующего put() (или put_nowait()) для полного объекта очереди.

Объекты очередейQueue Objects

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

Queue.qsize()
Возвращает приблизительный размер очереди. Обратите внимание: qsize() > 0 не гарантирует, что последующий вызов get() не заблокируется, равно как qsize() < maxsize не гарантирует, что put() не заблокируется.
Queue.put(item[, block[, timeout]])
Помещает item в очередь. Если необязательный аргумент block равен true, а timeout равен None (значение по умолчанию), при необходимости блокируется до появления свободного места. Если timeout – положительное число, блокируется не более чем на timeout секунд и возбуждает исключение Full, если свободное место не появилось за это время. В противном случае (block равен false) помещает элемент в очередь, если свободное место есть немедленно; иначе возбуждает исключение Full (в этом случае timeout игнорируется).
Queue.put_nowait(item)
Эквивалентно put(item, False).
Queue.get([block[, timeout]])
Удаляет и возвращает элемент из очереди. Если необязательный аргумент block равен true и timeout равен None (по умолчанию), при необходимости блокируется до появления элемента. Если timeout – положительное число, он блокируется не более чем на timeout секунд и возбуждает исключение Empty, если за это время элемент не появился. В противном случае (block равен false) возвращает элемент, если он доступен немедленно; иначе возбуждает исключение Empty (в этом случае timeout игнорируется).
Queue.get_nowait()
Эквивалентно get(False).

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

Queue.task_done()

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

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

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

Queue.join()

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

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

Пример ожидания завершения поставленных в очередь задач:

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.set_daemon(True)
     t.start()

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

q.join()       # ожидать завершения всех задач