17.7. queue – Класс синхронизированной очереди

Исходный код: 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).

exceptionqueue.Empty¶

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

exceptionqueue.Full¶

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

17.7.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=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()

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

Смотрите также

Класс multiprocessing.Queue

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

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