Содержание страницы

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

Исходный код: 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 меньше или равно нулю, размер очереди бесконечен.

class queue.PriorityQueue(maxsize=0)¶

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

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

exception queue.Empty¶: Исключение, возбуждаемое при вызове неблокирующего get() (или get_nowait()) на пустом объекте Queue.

exception queue.Full¶: Исключение, возбуждаемое при вызове неблокирующего put() (или put_nowait()) на полном объекте Queue.

17.7.1. Объекты очереди¶Queue Objects

Объекты очередей (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=True, timeout=None)¶: Помещает item в очередь. Если необязательный аргумент block равен true, а timeout равен None (по умолчанию), блокирует при необходимости до появления свободного места. Если timeout – положительное число, блокирует не более timeout секунд и вызывает исключение Full, если за это время свободное место не появилось. В противном случае (block равен false), помещает элемент в очередь, если свободное место доступно немедленно, иначе вызывает исключение Full (timeout в этом случае игнорируется).

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

Queue.get(block=True, timeout=None)¶: Удаляет и возвращает элемент из очереди. Если необязательный аргумент block равен true, а timeout равен None (по умолчанию), блокирует при необходимости до появления элемента. Если timeout – положительное число, блокирует не более timeout секунд и вызывает исключение Empty, если за это время элемент не появился. В противном случае (block равен false), возвращает элемент, если он доступен немедленно, иначе вызывает исключение Empty (timeout в этом случае игнорируется).

Queue.get_nowait()¶: Эквивалентно get(False).

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

Queue.task_done()¶

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

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

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

Queue.join()¶

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

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

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

def worker():
    while True:
        item = q.get()
        if item is None:
            break
        do_work(item)
        q.task_done()

q = queue.Queue()
threads = []
for i in range(num_worker_threads):
    t = threading.Thread(target=worker)
    t.start()
    threads.append(t)

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

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

# остановить workers
for i in range(num_worker_threads):
    q.put(None)
for t in threads:
    t.join()

См. также

Класс multiprocessing.Queue: Класс очереди для использования в контексте многопроцессорной обработки (а не многопоточности).

collections.deque – это альтернативная реализация неограниченных очередей с быстрыми атомарными операциями append() и popleft(), которые не требуют блокировки.

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

17.7.1. Объекты очереди¶Queue Objects

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