QueuesОчереди

Исходный код: Lib/asyncio/queues.py

---

Очереди asyncio спроектированы так, чтобы быть похожими на классы модуля queue. Хотя очереди asyncio не являются потокобезопасными, они предназначены для использования именно в async/await коде.

Обратите внимание, что методы очередей asyncio не имеют параметра timeout; используйте функцию asyncio.wait_for() для выполнения операций с очередью с тайм-аутом.

Смотрите также раздел Примеры ниже.

QueueОчередь

class asyncio.Queue(maxsize=0)¶

Очередь «первым вошёл – первым вышел» (FIFO).

Если maxsize меньше или равно нулю, размер очереди бесконечен. Если это целое число больше 0, то await put() блокируется, когда очередь достигает maxsize, пока элемент не будет удалён вызовом get().

В отличие от queue из стандартной библиотеки threading, размер очереди всегда известен и может быть получен вызовом метода qsize().

Изменено в версии 3.10: Удалён параметр loop.

Этот класс является не потокобезопасным.

maxsize¶

Количество элементов, допустимое в очереди.

empty()¶

Возвращает True, если очередь пуста, False в противном случае.

full()¶

Возвращает True, если в очереди есть maxsize элементов.

Если очередь была инициализирована с maxsize=0 (по умолчанию), то full() никогда не возвращает True.

async get()¶

Удаляет и возвращает элемент из очереди. Если очередь пуста, ждёт, пока элемент не станет доступен.

Возбуждает QueueShutDown, если очередь была закрыта и пуста, или если очередь была закрыта немедленно.

get_nowait()¶

Возвращает элемент, если он немедленно доступен, иначе возбуждает QueueEmpty.

async join()¶

Блокируется до тех пор, пока все элементы в очереди не будут получены и обработаны.

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

async put(item)¶

Помещает элемент в очередь. Если очередь полна, ждёт, пока не освободится место, прежде чем добавить элемент.

Возбуждает QueueShutDown, если очередь была закрыта.

put_nowait(item)¶

Помещает элемент в очередь без блокировки.

Если свободный слот недоступен, вызывает QueueFull.

qsize()¶

Возвращает количество элементов в очереди.

shutdown(immediate=False)¶

Переводит экземпляр Queue в режим завершения работы.

Очередь больше не может расти. Последующие вызовы put() вызывают QueueShutDown. Текущие заблокированные вызывающие put() будут разблокированы и вызовут QueueShutDown в ранее ожидающей задаче.

Если immediate равен false (по умолчанию), очередь может быть нормально завершена с помощью вызовов get() для извлечения задач, которые уже были загружены.

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

Как только очередь станет пуста, последующие вызовы get() будут вызывать QueueShutDown.

Если immediate равен true, очередь завершается немедленно. Очередь опустошается полностью, и количество незавершённых задач уменьшается на количество опустошённых задач. Если незавершённых задач нет, вызывающие join() разблокируются. Кроме того, заблокированные вызывающие get() разблокируются и вызовут QueueShutDown, поскольку очередь пуста.

Будьте осторожны при использовании join() с immediate, установленным в true. Это разблокирует ожидание завершения (join) даже если над задачами не было выполнено никакой работы, нарушая обычный инвариант присоединения к очереди.

Добавлено в версии 3.13.

task_done()¶

Указывает, что ранее помещённый в очередь элемент работы завершён.

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

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

Вызывает ValueError, если вызван большее количество раз, чем было элементов, помещённых в очередь.

Priority QueueОчередь с приоритетом

class asyncio.PriorityQueue¶

Вариант Queue; извлекает записи в порядке приоритета (сначала наименьший).

Записи обычно являются кортежами вида (priority_number, data).

LIFO QueueОчередь LIFO

class asyncio.LifoQueue¶

Вариант Queue, который извлекает последние добавленные записи первыми (последним пришёл – первым ушёл).

ExceptionsИсключения

exception asyncio.QueueEmpty¶

Это исключение вызывается, когда метод get_nowait() вызывается на пустой очереди.

exception asyncio.QueueFull¶

Исключение вызывается, когда метод put_nowait() вызывается на очереди, которая достигла своего maxsize.

exception asyncio.QueueShutDown¶

Исключение вызывается, когда put() или get() вызывается на очереди, которая была завершена (shut down).

Добавлено в версии 3.13.

ExamplesПримеры

Очереди могут использоваться для распределения рабочей нагрузки между несколькими одновременными задачами:

import asyncio
import random
import time


async def worker(name, queue):
    while True:
        # Взять «элемент работы» из очереди.
        sleep_for = await queue.get()

        # Уснуть на «sleep_for» секунд.
        await asyncio.sleep(sleep_for)

        # Уведомить очередь, что «элемент работы» обработан.
        queue.task_done()

        print(f'{name} has slept for {sleep_for:.2f} seconds')


async def main():
    # Создать очередь для хранения нашей «нагрузки».
    queue = asyncio.Queue()

    # Сгенерировать случайные временные интервалы и поместить их в очередь.
    total_sleep_time = 0
    for _ in range(20):
        sleep_for = random.uniform(0.05, 1.0)
        total_sleep_time += sleep_for
        queue.put_nowait(sleep_for)

    # Создать три задачи-обработчика для параллельной обработки очереди.
    tasks = []
    for i in range(3):
        task = asyncio.create_task(worker(f'worker-{i}', queue))
        tasks.append(task)

    # Дождаться полной обработки очереди.
    started_at = time.monotonic()
    await queue.join()
    total_slept_for = time.monotonic() - started_at

    # Отменить наши рабочие задачи.
    for task in tasks:
        task.cancel()
    # Дождаться отмены всех рабочих задач.
    await asyncio.gather(*tasks, return_exceptions=True)

    print('====')
    print(f'3 workers slept in parallel for {total_slept_for:.2f} seconds')
    print(f'total expected sleep time: {total_sleep_time:.2f} seconds')


asyncio.run(main())