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

exception queue.Empty

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

exception queue.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() в настоящее время блокируется, он возобновится, когда все элементы будут обработаны (что означает, что вызов task_done() был получен для каждого элемента, который был помещен в очередь с помощью put()).

Поднимает исключение 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)

# block until all tasks are done
q.join()

# stop workers
for i in range(num_worker_threads):
    q.put(None)
for t in threads:
    t.join()