Python asyncio: async и await

Модуль asyncio в Python позволяет писать конкурентный код в одном потоке с помощью ключевых слов async и await. Вместо того чтобы блокировать выполнение во время ожидания сетевых ответов или чтения файлов, программа на asyncio приостанавливает ожидающую задачу и немедленно переключается на другую работу — возобновляя её, когда результат будет готов. Это делает asyncio идеальным инструментом для I/O-нагруженных программ: веб-скраперов, API-клиентов и чат-серверов.

В этой главе рассматривается:

• Что такое async-функции (корутины) и чем они отличаются от обычных функций
• Цикл событий и то, как asyncio планирует работу
• Ожидание результатов, конкурентный запуск задач с помощью asyncio.gather и asyncio.create_task
• Обработка исключений и таймаутов в асинхронном коде
• asyncio.Queue для паттернов производитель-потребитель
• Когда использовать asyncio, а когда обратиться к threading

Зачем нужен asyncio

Рассмотрим программу, которая последовательно обращается к двум API:

import time

def fetch(name, delay):
    time.sleep(delay)          # blocks the whole program
    return f'data from {name}'

start = time.perf_counter()
r1 = fetch('API A', 1)
r2 = fetch('API B', 1)
print(f'Done in {time.perf_counter() - start:.1f}s')
# Done in 2.0s

Оба вызова выполняются последовательно — 2 секунды в сумме, хотя каждый запрос требует лишь 1 секунды ожидания. С asyncio программа приостанавливает fetch('API A', ...) во время ожидания, немедленно запускает fetch('API B', ...), и оба завершаются примерно за 1 секунду:

import asyncio
import time

async def fetch(name, delay):
    await asyncio.sleep(delay)   # suspends only this coroutine
    return f'data from {name}'

async def main():
    start = time.perf_counter()
    r1, r2 = await asyncio.gather(fetch('API A', 1), fetch('API B', 1))
    print(f'Done in {time.perf_counter() - start:.1f}s')
    # Done in 1.0s

asyncio.run(main())

Корутины: async def и await

Функция, определённая с помощью async def, называется корутинной функцией. При её вызове тело не выполняется сразу — возвращается объект корутины, которым должен управлять цикл событий.

async def greet(name):
    print(f'Hello, {name}!')

# Calling it returns a coroutine object, nothing is printed yet
coro = greet('World')
print(type(coro))   # <class 'coroutine'>

# Run it properly
import asyncio
asyncio.run(greet('World'))
# Hello, World!

Внутри корутины await приостанавливает выполнение до тех пор, пока ожидаемый объект (другая корутина, Task или Future) не вернёт результат. Пока одна корутина приостановлена, цикл событий свободен для выполнения других корутин.

import asyncio

async def step_one():
    print('Step 1: start')
    await asyncio.sleep(1)     # suspend for 1 second
    print('Step 1: end')
    return 'result-1'

async def main():
    value = await step_one()   # wait for step_one to finish
    print(value)

asyncio.run(main())
# Step 1: start
# Step 1: end
# result-1

Что можно ожидать с await

• Другую корутину async def
• asyncio.Task (созданный с помощью asyncio.create_task)
• asyncio.Future
• Любой объект с методом __await__

Использовать await нельзя вне функции async def.

Цикл событий

Цикл событий — это планировщик asyncio. Он ведёт очередь корутин и задач, выполняет каждую из них до момента встречи с await, а затем переключается на следующий готовый элемент. Как правило, в каждом потоке существует один цикл событий.

asyncio.run(coro) — стандартная точка входа для asyncio-программ. Он создаёт новый цикл событий, выполняет заданную корутину до завершения, закрывает цикл и возвращает результат:

import asyncio

async def compute():
    await asyncio.sleep(0)   # yield control once
    return 6 * 7

result = asyncio.run(compute())
print(result)   # 42

В большинстве приложений управлять циклом напрямую не нужно — asyncio.run берёт на себя создание и завершение.

Конкурентный запуск задач

asyncio.gather

asyncio.gather(*coroutines) планирует конкурентное выполнение всех переданных корутин и возвращает их результаты в том же порядке:

import asyncio

async def fetch_data(name, delay):
    print(f'Start fetching {name}')
    await asyncio.sleep(delay)
    print(f'Done fetching {name}')
    return f'data from {name}'

async def main():
    results = await asyncio.gather(
        fetch_data('API A', 1),
        fetch_data('API B', 2),
        fetch_data('API C', 1),
    )
    print(results)

asyncio.run(main())
# Start fetching API A
# Start fetching API B
# Start fetching API C
# Done fetching API A
# Done fetching API C
# Done fetching API B
# ['data from API A', 'data from API B', 'data from API C']

Все три корутины стартуют немедленно. Общее время выполнения соответствует самой медленной корутине (2 с), а не их сумме (4 с).

asyncio.create_task

asyncio.create_task(coro) оборачивает корутину в Task и планирует её скорое выполнение. В отличие от gather, создание задачи запускает её в фоне, пока текущая корутина продолжает работать:

import asyncio

async def background_job(name, delay):
    print(f'{name}: start')
    await asyncio.sleep(delay)
    print(f'{name}: end')
    return f'{name} done'

async def main():
    t1 = asyncio.create_task(background_job('Task A', 1))
    t2 = asyncio.create_task(background_job('Task B', 2))

    # Both tasks are already scheduled; await collects their results
    result1 = await t1
    result2 = await t2
    print(result1, result2)

asyncio.run(main())
# Task A: start
# Task B: start
# Task A: end
# Task B: end
# Task A done Task B done

Используйте create_task, когда хотите запустить задачу немедленно и планируете позднее собрать её результат (или отменить её). Используйте gather, когда нужно запустить фиксированную группу корутин и дождаться их всех вместе.

Чередующийся вывод

Хороший способ увидеть цикл событий в действии — наблюдать, как задачи чередуются:

import asyncio

async def count_down(name, seconds):
    for i in range(seconds, 0, -1):
        print(f'{name}: {i}')
        await asyncio.sleep(1)
    print(f'{name}: done!')

async def main():
    await asyncio.gather(
        count_down('Task A', 3),
        count_down('Task B', 2),
    )

asyncio.run(main())
# Task A: 3
# Task B: 2
# Task A: 2
# Task B: 1
# Task A: 1
# Task B: done!
# Task A: done!

Обе задачи используют один поток; цикл событий переключается между ними при каждом await asyncio.sleep(1).

Обработка исключений

Исключения, возникшие внутри корутины, распространяются через await точно так же, как в синхронном коде. Используйте обычный блок try/except:

import asyncio

async def risky_task():
    await asyncio.sleep(0.1)
    raise ValueError('something went wrong')

async def main():
    try:
        await risky_task()
    except ValueError as e:
        print(f'Caught: {e}')

asyncio.run(main())
# Caught: something went wrong

При использовании asyncio.gather, если одна корутина вызывает исключение, остальные по умолчанию не отменяются, однако исключение повторно возникает при await вызова gather. Передайте return_exceptions=True, чтобы собирать исключения как возвращаемые значения:

import asyncio

async def good():
    return 'ok'

async def bad():
    raise RuntimeError('oops')

async def main():
    results = await asyncio.gather(good(), bad(), return_exceptions=True)
    for r in results:
        if isinstance(r, Exception):
            print(f'Error: {r}')
        else:
            print(f'Result: {r}')

asyncio.run(main())
# Result: ok
# Error: oops

Таймауты с asyncio.wait_for

asyncio.wait_for(coro, timeout) выполняет корутину и отменяет её, если она не завершилась в течение заданного числа секунд, вызывая asyncio.TimeoutError:

import asyncio

async def slow_operation():
    await asyncio.sleep(5)
    return 42

async def main():
    try:
        result = await asyncio.wait_for(slow_operation(), timeout=1.0)
        print(result)
    except asyncio.TimeoutError:
        print('Timed out — operation cancelled')

asyncio.run(main())
# Timed out — operation cancelled

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

asyncio.Queue для паттернов производитель-потребитель

asyncio.Queue — это потокобезопасная, асинхронная очередь. Она идеально подходит для разделения производителей (кода, генерирующего работу) и потребителей (кода, обрабатывающего её):

import asyncio

async def producer(queue):
    for i in range(1, 4):
        print(f'Produced item {i}')
        await queue.put(i)
        await asyncio.sleep(0.1)
    await queue.put(None)   # sentinel to signal consumers to stop

async def consumer(queue):
    while True:
        item = await queue.get()
        if item is None:
            break
        print(f'Consumed item {item}')

async def main():
    q = asyncio.Queue()
    await asyncio.gather(producer(q), consumer(q))

asyncio.run(main())
# Produced item 1
# Consumed item 1
# Produced item 2
# Consumed item 2
# Produced item 3
# Consumed item 3

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

asyncio против threading

И asyncio, и модуль threading в Python позволяют выполнять работу конкурентно, но делают это по-разному:

Используйте asyncio, когда вы управляете I/O-кодом и можете применять async-совместимые библиотеки (например, aiohttp, asyncpg). Используйте threading, когда полагаетесь на сторонние блокирующие библиотеки, которые нельзя сделать асинхронными.

Для истинного CPU-параллелизма обратитесь к multiprocessing или concurrent.futures.ProcessPoolExecutor.

Распространённые ошибки

Забытый await: Вызов async-функции без await возвращает объект корутины и ничего не делает. Python выдаёт RuntimeWarning: coroutine '...' was never awaited, чтобы помочь обнаружить такую ошибку.

async def main():
    asyncio.sleep(1)   # BUG: returns a coroutine, does not sleep
    await asyncio.sleep(1)   # correct

Блокировка цикла событий: Запуск медленного синхронного кода (плотного цикла, блокирующего сетевого вызова, time.sleep) внутри корутины замораживает весь цикл событий. Оберните блокирующие вызовы в asyncio.to_thread (Python 3.9+), чтобы выполнить их в пуле потоков без блокировки:

import asyncio
import time

def blocking_task():
    time.sleep(2)   # simulates a slow blocking operation
    return 'done'

async def main():
    result = await asyncio.to_thread(blocking_task)
    print(result)

asyncio.run(main())
# done

Использование asyncio.run внутри работающего цикла: Jupyter Notebook уже запускает цикл событий. Используйте await coro напрямую в ячейках notebook или установите nest_asyncio, чтобы разрешить вложенные циклы.

Краткий справочник

Практика