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 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 в этом случае игнорируется).

Возбуждает 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 в этом случае игнорируется).

До версии 3.0 в системах POSIX и во всех версиях в 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() разблокируется.

Waiting for task completionОжидание завершения задач

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

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()

# Включить рабочий поток.
threading.Thread(target=worker, daemon=True).start()

# Отправить тридцать запросов задач рабочему.
for item in range(30):
    q.put(item)

# Блокировать до завершения всех задач.
q.join()
print('All work completed')

Terminating queuesЗавершение очередей

Когда больше не нужны, объекты 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 ObjectsОбъекты 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).

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

Класс multiprocessing.Queue

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

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