Очереди

Очередь

class asyncio.Queue(maxsize=0)

Очередь типа «первым пришёл — первым вышел» (FIFO).

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

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

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

task_done()

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

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

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

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

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

class asyncio.PriorityQueue

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

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

Очередь LIFO

class asyncio.LifoQueue

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

Исключения

exception asyncio.QueueEmpty

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

exception asyncio.QueueFull

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

exception asyncio.QueueShutDown

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

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

Примеры

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

import asyncio
import random
import time


async def worker(name, queue):
    while True:
        # Get a "work item" out of the queue.
        sleep_for = await queue.get()

        # Sleep for the "sleep_for" seconds.
        await asyncio.sleep(sleep_for)

        # Notify the queue that the "work item" has been processed.
        queue.task_done()

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


async def main():
    # Create a queue that we will use to store our "workload".
    queue = asyncio.Queue()

    # Generate random timings and put them into the 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)

    # Create three worker tasks to process the queue concurrently.
    tasks = []
    for i in range(3):
        task = asyncio.create_task(worker(f'worker-{i}', queue))
        tasks.append(task)

    # Wait until the queue is fully processed.
    started_at = time.monotonic()
    await queue.join()
    total_slept_for = time.monotonic() - started_at

    # Cancel our worker tasks.
    for task in tasks:
        task.cancel()
    # Wait until all worker tasks are cancelled.
    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())