Очереди

Очередь

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.

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