очередь — Класс синхронизированной очереди

Исходный код: 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, который полон.

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