Очереди

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

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

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

Очередь

class asyncio.Queue(maxsize=0, *, loop=None)

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

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

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

Этот класс не защищён от одновременного доступа.

maxsize

Максимальное количество элементов в очереди.

empty()

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

full()

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

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

coroutine get()

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

get_nowait()

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

coroutine join()

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

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

coroutine put(item)

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

put_nowait(item)

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

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

qsize()

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

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.

Примеры

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

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