Очереди

Queue

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

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

get_nowait()

Возвращает элемент, если он доступен немедленно; в противном случае вызывает исключение QueueEmpty.

async join()

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

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

async put(item)

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

put_nowait(item)

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

Если свободного места нет, вызывает исключение QueueFull.

qsize()

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

task_done()

Сообщает, что ранее добавленная в очередь рабочая задача завершена.

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

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

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

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

class asyncio.PriorityQueue

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

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

Очередь LIFO

class asyncio.LifoQueue

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

Исключения

exception asyncio.QueueEmpty

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

exception asyncio.QueueFull

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

Примеры

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

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