Структуры данных
Очереди типов FIFO и LIFO
Очередь deque модуля collections
Класс Queue модуля queue
Классы LifoQueue и PriorityQueue модуля queue
Понятие кучи (heap). Модуль heapq
Класс array. Методы класса array
Обучающие курсы
Python Стандартная библиотека Python Python ООП Django C/C++ ООП C++ Машинное обучение Нейронные сети Структуры данных
Телеграм-каналы
Канал selfedu_sb Чат по Python Чат по Django Чат по машинному обучению Чат по Java и С++
Наши каналы

Класс Queue модуля queue

Курс по стандартной библиотеке: https://stepik.org/a/259466?utm_source=proproprogs

Смотреть материал на YouTube | RuTube

На прошлом занятии мы с вами познакомились с работой класса deque, реализующий двустороннюю очередь, и который эффективен при добавлении/удалении элементов с ее концов. Вообще говоря, с помощью deque можно строить разную логику обработки граничных элементов, создавая, в том числе, очереди типа FIFO и LIFO. Однако класс deque лучше не применять при многопоточном программировании, т.к. он не поддерживает механизмы блокировок. Для этого лучше подходят другие специализированные классы из модуля queue:

queue.Queue – потокобезопасная реализация очереди типа FIFO;
queue.LifoQueue – потокобезопасная реализация очереди типа LIFO;
queue.PriorityQueue – потокобезопасная реализация очереди с приоритетами.

Они имеют несколько более сложную реализацию, чем очередь deque, поэтому в одно-двух поточных приложениях предпочтение следует отдавать deque. Но при большом числе потоков лучше применять приведенные выше классы очередей.

Очередь queue.Queue типа FIFO

Первый потокобезопасный класс queue.Queue позволяет добавлять элемент на одном конце очереди, а удалять на другом, используя методы:

put(item[, block=True[, timeout=None]]). Добавляет элемент в очередь. Операция блокирует выполнение текущего потока, если очередь полна и указан параметр block=True. Если же указана длительность ожидания timeout, операция прервётся по истечении указанного времени.
get([block=True[, timeout=None]]). Возвращает элемент из головы очереди и удаляет его. Блокирует поток, если очередь пуста и задан параметр block=True. При наличии таймаута операция прерывается по истечению отведённого времени.

Давайте на конкретном примере посмотрим, как она работает. Создать объект очереди можно командами:

import queue
 
q_fifo = queue.Queue() # неограниченная очередь FIFO
q_fifo7 = queue.Queue(maxsize=7) # очередь FIFO; максимум 7 элементов

Непосредственно после создания очереди пустые, не содержат ни одного элемента. Это можно определить с помощью метода empty():

q_fifo.empty() # True

Добавим в очередь q_fifo данные из следующего списка:

data = ['car', 7, True, (1, 2, 3), ['a', 'b'], {5, 4}, {'key1': 7, 'key2': -7}, 10]

Выполним это с помощью метода put:

for d in data:
    q_fifo.put(d, block=False)

Я прописал дополнительно параметр block=False, чтобы в случае каких-либо ошибок программа сразу прерывалась, не блокируя главный поток.

Теперь извлечем данные из очереди q_fifo путем вызова метода get():

while not q_fifo.empty():
    d = q_fifo.get(block=False)
    print(d)

Мы видим, что данные извлекаются в том же порядке, что были помещены:

car
7
True
(1, 2, 3)
['a', 'b']
{4, 5}
{'key1': 7, 'key2': -7}
10

Однако такой подход к чтению очередного значения не потокобезопасный, т.к. метод empty() использует текущее состояние очереди и она может измениться, пока дело дойдет до выполнения метода get(). Надежнее использовать механизм обработки исключений, например, так:

while not q_fifo.empty():
    try:
        d = q_fifo.get(block=False)
        print(d)
    except queue.Empty as e:
        print("Empty: " + str(e))
        break

В этом случае при наличии пустой очереди непосредственно в момент ее чтения возникает исключение типа queue.Empty и цикл завершается.

Дополнительные методы класса queue.Queue

Помимо методов put(), get() и empty() класс queue.Queue содержит еще такие полезные методы:

task_done(). Сообщает о завершении обработки конкретного элемента, ранее полученного из очереди. Обычно применяется совместно с механизмом .join(), позволяющим ожидать завершения всех заданий в очереди.
join(). Блокирует основной поток до тех пор, пока все задания в очереди не будут обработаны. Метод полезен для координации завершения работы большого числа задач, запущенных параллельно.
full(). Аналогична предыдущему методу, но проверяет полноту очереди относительно максимального установленного размера.
qsize(). Возвращает приблизительное количество элементов в очереди. Поскольку очередь постоянно меняется, возвращаемое значение носит скорее оценочный характер.

Например, воспользуемся методом qsize() и выведем количество элементов в очереди:

print("Число элементов в очереди:", q_fifo.qsize())

Работа с ограниченной очередью queue.Queue

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

q_fifo7 = queue.Queue(maxsize=7) # очередь FIFO; максимум 7 элементов

Давайте посмотрим, что будет, если в нее попытаться добавить 8 элементов из списка data:

for d in data:
    q_fifo7.put(d, block=False)

При добавлении 8-го элемента возникает исключение типа queue.Full. То есть, мы не можем добавлять очередной элемент в заполненную очередь. Для обработки этого исключения поместим вызов метода put() в блок try/except:

for d in data:
    try:
        q_fifo7.put(d, block=False)
    except queue.Full as e:
        print("Full: " + str(e))
        break

Конечно, в многопоточной программе параметр block всюду следует оставлять по умолчанию True, иначе могут возникать проблемы с совместным использованием объекта очереди из разных потоков.

Чтение данных выполняется аналогичным образом:

while not q_fifo7.empty():
    try:
        d = q_fifo7.get() # block=True
        print(d)
    except queue.Empty as e:
        print("Empty: " + str(e))
        break

Заключение

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

В качестве типовых сценариев использования класса queue.Queue можно выделить следующие:

Передача сообщений между потоками. Один поток ("производитель") генерирует события или данные и кладёт их в очередь, второй поток ("потребитель") берёт элементы из очереди и обрабатывает их.
Буферизация данных. Например, сетевые соединения часто используют очередь для временного сохранения полученных пакетов данных до момента их полной обработки.
Защита критичных ресурсов. Даже если сама логика вашего приложения однопоточная, использование очереди может упростить работу с внешними API или базами данных, создавая барьер против чрезмерных обращений.