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

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

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

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

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

Кроме того, модуль реализует «простую» очередь FIFO типа SimpleQueue, конкретная реализация которой предоставляет дополнительные гарантии в обмен на меньший набор возможностей.

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

class queue.Queue(maxsize=0): Конструктор очереди FIFO. maxsize — целое число, задающее верхний предел количества элементов, которые можно поместить в очередь. После достижения этого размера добавление блокируется до тех пор, пока элементы очереди не будут извлечены. Если maxsize меньше или равен нулю, размер очереди не ограничен.

class queue.LifoQueue(maxsize=0): Конструктор очереди LIFO. maxsize — целое число, задающее верхний предел количества элементов, которые можно поместить в очередь. После достижения этого размера добавление блокируется до тех пор, пока элементы очереди не будут извлечены. Если maxsize меньше или равен нулю, размер очереди не ограничен.

class queue.PriorityQueue(maxsize=0)

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

Сначала извлекаются элементы с наименьшими значениями (элемент с наименьшим значением — это элемент, который вернул бы min(entries)). Типичный формат элементов — кортеж вида: (priority_number, data).

Если элементы data нельзя сравнивать, данные можно обернуть в класс, который игнорирует элемент данных и сравнивает только номер приоритета:

from dataclasses import dataclass, field
from typing import Any

@dataclass(order=True)
class PrioritizedItem:
    priority: int
    item: Any=field(compare=False)

class queue.SimpleQueue

Конструктор неограниченной очереди FIFO. В простых очередях отсутствуют расширенные возможности, например отслеживание задач.

Простые очереди являются обобщёнными по типу своих элементов.

Добавлено в версии 3.7.

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

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

exception queue.ShutDown: Исключение, возникающее при вызове put() или get() для объекта Queue, который был остановлен.

Добавлено в версии 3.13.

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

Объекты очередей (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 в этом случае игнорируется).

Возбуждает ShutDown, если очередь остановлена.

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

Queue.get(block=True, timeout=None)

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

В POSIX-системах до версии 3.0, а также во всех версиях Windows, если block имеет значение true, а timeout равен None, эта операция выполняет непрерываемое ожидание на базовой блокировке. Это означает, что исключения не могут возникнуть; в частности, SIGINT не вызовет KeyboardInterrupt.

Возбуждает ShutDown, если очередь остановлена и пуста либо если очередь остановлена немедленно.

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

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

Queue.task_done()

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

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

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

Queue.join()

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

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

Ожидание завершения задач

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

import threading
import queue

q = queue.Queue()

def worker():
    while True:
        item = q.get()
        print(f'Working on {item}')
        print(f'Finished {item}')
        q.task_done()

# Turn-on the worker thread.
threading.Thread(target=worker, daemon=True).start()

# Send thirty task requests to the worker.
for item in range(30):
    q.put(item)

# Block until all tasks are done.
q.join()
print('All work completed')

Остановка очередей

Когда объекты Queue больше не нужны, их можно штатно завершить после опустошения или немедленно остановить.

Queue.shutdown(immediate=False)

Переводит экземпляр Queue в режим остановки.

Добавлять элементы в очередь больше нельзя. Последующие вызовы put() возбуждают ShutDown. Текущие заблокированные вызовы put() будут разблокированы и в ранее заблокированном потоке возбудят ShutDown.

Если immediate имеет значение false (по умолчанию), очередь можно штатно завершить, вызвав get() для извлечения уже добавленных задач.

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

После опустошения очереди последующие вызовы get() будут возбуждать ShutDown.

Если immediate имеет значение true, очередь останавливается немедленно. Очередь полностью опустошается, а счётчик незавершённых задач уменьшается на количество удалённых задач. Если число незавершённых задач равно нулю, ожидающие вызовы join() разблокируются. Кроме того, заблокированные вызовы get() разблокируются и возбудят ShutDown, поскольку очередь пуста.

Будьте осторожны при использовании join() со значением immediate true. В этом случае вызов join разблокируется, даже если задачи не были обработаны, что нарушает обычный инвариант ожидания завершения обработки очереди.

Добавлено в версии 3.13.

Объекты SimpleQueue

Объекты SimpleQueue предоставляют описанные ниже общедоступные методы.

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

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

SimpleQueue.put(item, block=True, timeout=None): Помещает item в очередь. Метод никогда не блокируется и всегда выполняется успешно (за исключением возможных низкоуровневых ошибок, например ошибки выделения памяти). Необязательные аргументы block и timeout игнорируются и предусмотрены только для совместимости с Queue.put().

Особенность реализации CPython: Этот метод реализован на C и допускает повторный вход. То есть вызов put() или get() может быть прерван другим вызовом put() в том же потоке без взаимной блокировки или повреждения внутреннего состояния очереди. Благодаря этому метод подходит для использования в деструкторах, например в методах __del__, или в обратных вызовах weakref.

SimpleQueue.put_nowait(item): Эквивалентен put(item, block=False) и предусмотрен для совместимости с Queue.put_nowait().

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

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

См. также

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

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