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.

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

Объекты очередей (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, 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.

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

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

Queue.task_done()

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

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

Вызывает исключение 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')

Объекты 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(), которым не требуется блокировка и которые также поддерживают индексирование.