Очереди¶Queues

Очередь¶Queue

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

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

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

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

Устарело с версии 3.8, будет удалено в версии 3.10: Параметр loop.

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

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

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

Приоритетная очередь¶Priority Queue

class asyncio.PriorityQueue¶

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

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

Очередь LIFO¶LIFO Queue

class asyncio.LifoQueue¶

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

Исключения¶Exceptions

exception asyncio.QueueEmpty¶

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

exception asyncio.QueueFull¶

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

Примеры¶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())