Асинхронноcть и пакет asyncio

Асинхронное программирование представляет подход к написанию кода, который позволяет программе выполнять другие задачи, пока она ожидает завершения ресурсоемких операций ввода-вывода (I/O). В отличие от мультипроцессорности, которая задействует разные ядра CPU, асинхронность в Python работает в рамках одного потока на одном ядре. Она использует так называемый Event Loop (цикл событий). Когда одна задача переходит в режим ожидания (например, ждет ответа от сервера), Event Loop временно переключается на выполнение другой задачи. Как только первая задача получает данные, цикл возвращается к ней. Благодаря это достигается высокая производительность на I/O задачах при минимальном потреблении RAM (в отличие от процессов).

Отличие обычного синхронного кода от асинхронного можно продемонстрировать на следующем примере:

• Синхронный повар ставит чайник и неподвижно стоит у плиты 5 минут, ожидая закипания, и только потом начинает резать салат.

• Асинхронный повар ставит чайник, и пока тот греется, сразу начинает резать салат. Он эффективно использует время ожидания.

При этом не следует путать асинхронность и мультипроцессорность:

• Асинхронность: идеальна для I/O-bound задач с большим количеством сетевых соединений или ожиданий (парсинг тысяч веб-страниц, создание чат-ботов, веб-серверов, работа с базами данных). Она позволяет держать тысячи открытых соединений одновременно в одном потоке.

• Мультипроцессорность (multiprocessing): нужна для CPU-bound задач, где процессор должен непрерывно считать (рендеринг, математика, криптография). Асинхронность тут не поможет, так как процессор не "ждет", а работает.

Библиотека asyncio

Для работы с асинхронным кодом в Python используется встроенная библиотека asyncio и два ключевых слова:

• async: объявляет асинхронную функцию. Такая функция при вызове возвращает специальный объект - корутину (coroutine).

• await: ставится перед вызовом другой корутины. Это указание для Event Loop: "Я жду этот результат, можешь пока заняться другими задачами".

Определим простейшую корутину:

import asyncio

async def get_message():
    print("Hello...")
    return "Hello"

Здесь определена асинхронная функция, которая выводит на консоль строку и возвращает ее в качестве результата. Но в реальности эта функция возвращает корутину. Так, выведет результат ее вызова:

import asyncio

async def get_message():
    print("Hello...")
    return "Hello World"


result = get_message()
print(result)

И здесь мы получим консольный вывод типа следующего:

<coroutine object get_message at 0x7dc4d91bdd80>
<sys>:0: RuntimeWarning: coroutine 'get_message' was never awaited

Во-первых, обратите внимание, что при вызове функции get_message() на консоль НЕ выводится строка "Hello...", хотя в самой функции подобное действие имеется. Во-вторых, при выводе реультата функции мы получаем не строку, а данные корутины (в моем случае, это <coroutine object get_message at 0x7dc4d91bdd80>

И также стоит отметить предупреждение о том, что корутина не "ожидается" ("awaited") или, технически говоря, не вызывается с помощью ключевого слова await

Запуск корутины

Существует множество способов запустить корутину. Самый простой - использовать встроенный метод asyncio.run(coroutine), в который передается объект корутины. Например:

import asyncio

async def get_message():
    print("Hello...")
    return "Hello World"

asyncio.run(get_message())

В итоге на консоль будет выведена строка "Hello...".

Более того, поскольку наша асинхронная функция возвращает некоторый объект (в нашем случае строку "Hello World"), то мы ее можем получить как результат метода asyncio.run:

import asyncio

async def get_message():
    print("Hello...")
    return "Hello World"

result = asyncio.run(get_message())     # Hello...
print(result)       # Hello World

При вызове asyncio.run() эта функция автоматически выполняет следующие действия:

1. Запускает новый цикл событий Event Loop. Задача цикла событий - планировать параллельные задачи и связывать с обратными вызовами такие события, как завершение, отмена и исключения.

2. Оборачивает объект корутины в задачу.

3. Устанавливала обратный вызов по завершении задачи.

4. Проверяет задачу до ее завершения.

5. Возвращает результат.

await

Вернемся к предупреждению, которое нам выводила консоль при попытке запуска асинхронной функции в самом начале:

<sys>:0: RuntimeWarning: coroutine 'get_message' was never awaited

Повторюсь, что предупреждение о том, что корутина не "ожидается" ("awaited") или, технически говоря, не вызывается с помощью ключевого слова await. Но в программе выше мы не использовали await, а вместо этого использовали функцию asyncio.run(), которая обычно служит главным точкой входа для запуска всей асинхронной программы. Она создает новый цикл событий (event loop), запускает переданную корутину как главную задачу, а после ее завершения корректно закрывает цикл и очищает ресурсы. Но она применяется только в синхронном коде (обычно на самом верхнем уровне, например, в блоке if __name__ == "__main__""). И ее нельзя вызвать изнутри уже работающего цикла событий, иначе мы столкнемся с ошибкой. Например:

import asyncio

async def get_message():
    print("Hello...")
    return "Hello World"

async def print_message():
    result = asyncio.run(get_message())
    print(result)

asyncio.run(print_message())

Здесь на верхнем уровне вне каких-любо функций с помощь asyncio.run() вызываем асинхронную функцию print_message(). Однако внутри функции print_message() вызываем другую асинхронную функцию - get_message(). Но этот код работать не будет, а при его запуске мы столкнемся с ошибкой:

RuntimeError: asyncio.run() cannot be called from a running event loop

В случае выше с помощью asyncio.run(print_message()) запускается event loop, но затем опять же мы запускаем другую функцию в рамках этого же цикла - asyncio.run(get_message()). Для решения этой проблемы нам поможет ключевое слово await

Ключевое слово await используется внутри уже запущенной корутины для ожидания завершения другой задачи или операции. Он приостанавливает выполнение текущей корутины и передает управление циклу событий до тех пор, пока ожидаемый объект (например, другая корутина) не вернет результат.

await применяется только внутри функции, объявленной через async def (то есть внутри других корутин), и не может использоваться в обычном синхронном коде (вне функции async def). Например, изменим предыдущий код, использовав await:

import asyncio

async def get_message():
    print("Hello...")
    return "Hello World"

async def print_message():
    # Исопльзуем await
    result = await get_message())    # Hello...
    print(result)       # Hello World

asyncio.run(print_message())

В итоге мы получим консольный вывод, который и должен быть:

Hello...
Hello World

asyncio.sleep

Конечно, в примере выше не было смысла делать такие простые функции как get_message/print_message асинхронными. Тем не менее в Python есть ряд асинхронных функций. Одной из самых популярных из них является функция asyncio.sleep(), которая устанавляется задержку выполнения и качестве параметра принимает время задержки в секундах:

import asyncio

async def main():
    print("Привет...")
    # Имитируем асинхронное ожидание (например, запрос к базе данных)
    # Обычный time.sleep() здесь использовать нельзя, так как он заблокирует весь поток
    await asyncio.sleep(2)  # задержка в 2 секунды
    print("...Мир!")

if __name__ == "__main__":
    # Запуск главного цикла событий и выполнение функции
    asyncio.run(main())

Параллельное выполнение задач

Сама по себе одна корутина не дает преимуществ. Преимущество корутин заключается в возможности их параллельного выполнения. Поскольку объект корутины представляет собой переменную, которую можно передать функции, эти объекты можно связывать и объединять в цепочки или создавать последовательно.

Для одновременного выполнения нескольких корутин применяется асинхронная функция asyncio.gather(), которая принимает набор корутин:

import asyncio
import time

async def fetch_data(source_id, delay):
    print(f"[Старт] Загрузка данных из Источника {source_id}...")
    await asyncio.sleep(delay) # Имитация сетевой задержки
    print(f"[Готово] Данные из Источника {source_id} получены.")
    return f"данные {source_id}"

async def main():
    start_time = time.time()
    
    # Запускаем три задачи параллельно
    results = await asyncio.gather(
        fetch_data(1, 2),
        fetch_data(2, 3),
        fetch_data(3, 1)
    )
    
    end_time = time.time()
    print(f"Все результаты: {results}")
    print(f"Общее время выполнения: {end_time - start_time:.2f} сек.")
    
if __name__ == "__main__":
    asyncio.run(main())

Этот код запускает три задачи загрузки данных одновременно и измеряет общее время их выполнения с помощью библиотеки asyncio.

В этом коде асинхронная функция fetch_data(source_id, delay) имитирует сетевой запрос с заданными номером источника и задержкой. Внутри этой функции с помощью asyncio.sleep(delay) приостанавливаем работу функции на время задержки, имитируя тем самым долгую работу и освобождая процессор для выполнения других задач.

В асинхронной функции main() запускаем три задачи параллельно (три вызова fetch_data с задержками 2, 3 и 1 секунд). Для этого вызовы fetch_data (которые по сути являются корутинами) передаем в асинхронную функцию asyncio.gather(...):

results = await asyncio.gather(
        fetch_data(1, 2),
        fetch_data(2, 3),
        fetch_data(3, 1)
    )

В конце asyncio.run(main()) создает цикл событий и запускает главную функцию программы. Результат работы программы:

[Старт] Загрузка данных из Источника 1...
[Старт] Загрузка данных из Источника 2...
[Старт] Загрузка данных из Источника 3...
[Готово] Данные из Источника 3 получены.
[Готово] Данные из Источника 1 получены.
[Готово] Данные из Источника 2 получены.
Все результаты: ["данные 1", "данные 2", "данные 3"]
Общее время выполнения: 3.00 сек.

В итоге программа не ждет завершения каждой задачи по очереди. Задачи выполняются одновременно. А общее время работы программы равно времени самой долгой задачи, то есть около 3 секунд (а не 6 секунд, если бы задачи шли друг за другом).

Стоит отметить, что мы могли бы упорядочить передачу корутин, например, объединить их в список, что полезно, если такой набор передается из вне:

tasks = [
    fetch_data(1, 2),
    fetch_data(2, 3),
    fetch_data(3, 1)
]
results = await asyncio.gather(*tasks)

Создание фоновых задач

Иногда нужно запустить задачу в фоне и сразу продолжить выполнение основного кода, не дожидаясь ее окончания. Для этого применяется встроенная асинхронная функция asyncio.create_task(). Она возвращает объект Task, который дает контроль над фоновым процессом:

• task.cancel(): принудительно останавливает задачу.

• await task: позволяет дождаться ее окончания и получить результат (если функция что-то возвращает через return).

Например:

import asyncio

async def background_logger():
    while True:
        print("[Фон] Проверка статуса системы...")
        await asyncio.sleep(1)

async def main():
    # Создаем фоновую задачу (она начинает выполняться сразу)
    task = asyncio.create_task(background_logger())
    
    print("[Главный] Начало важной работы.")
    await asyncio.sleep(3.5) # Имитируем основную работу
    print("[Главный] Важная работа завершена.")
    
    # Отменяем фоновую задачу, иначе она будет работать вечно
    task.cancel()
    
if __name__ == "__main__":
    asyncio.run(main())

Этот код запускает фоновую задачу, которая каждую секунду выводит в консоль сообщение о проверке системы, пока в это же время в главной функции имитируется другая работа. Для этого определена асинхронная функция background_logger, которая работает в бесконечном цикле while True. Каждую секунду она печатает "[Фон] Проверка статуса системы..."" и делает паузу на одну секунду через await asyncio.sleep(1).

Главная асинхронная функция - main() создает фоновую задачу task с помощью asyncio.create_task(), которая сразу начинает выполняться параллельно.

Далее главная функция выводит сообщение "[Главный] Начало важной работы.." и засыпает» на 3,5 секунды через await asyncio.sleep(3.5), имитируя полезную нагрузку. За это время фоновая задача успевает выполниться примерно 3 раза.

В конце выводит сообщение "[Главный] Важная работа завершена.." и отменяет фоновую задачу методом task.cancel(), чтобы остановить бесконечный цикл.

Ожидаемый вывод в консоли:

[Главный] Начало важной работы.
[Фон] Проверка статуса системы...
[Фон] Проверка статуса системы...
[Фон] Проверка статуса системы...
[Главный] Важная работа завершена.

Что касается механики работы функции, то asyncio.create_task() оборачивает корутину в объект Task и сразу планирует ее выполнение в цикле событий (event loop) в фоновом режиме. И здесь на отметить ряд моментов:

• Немедленный запуск (без ожидания через await)

При вызове корутины обычным способом (например, background_logger()), она не начинает выполняться тотчас же. Она начнет работать только тогда, когда мы напишем await background_logger(), при этом основной код заблокируется.

asyncio.create_task(background_logger()) работает иначе:

1. Она добавляет задачу в очередь цикла событий.

2. Код тут же идет дальше к следующей строке, не дожидаясь завершения задачи.

3. Задача начнет выполняться при первой же паузе (когда главный поток встретит любой await).

• Конкурентность (Concurrency)

  create_task() позволяет выполнять несколько задач конкурентно на одном потоке. Пока одна задача "спит" (например, из-за вызова await asyncio.sleep()) или ждет ответа от сети, цикл событий переключается на выполнение фоновой задачи.

Например, без create_task (последовательно):

await task1()  # Ждем 3 секунды
await task2()  # Только потом ждем еще 3 секунды
# Всего уйдет 6 секунд.

С create_task (параллельно/конкурентно):

t1 = asyncio.create_task(task1())  # Запущена в фоне
t2 = asyncio.create_task(task2())  # Запущена в фоне

await t1
await t2

# Обе задачи работают одновременно. Всего уйдет 3 секунды.