Asyncio у Python: від теорії до практичного застосування 💻

Одним із найпотужніших модулів стандартної бібліотеки Python є asyncio, але водночас це один із найбільш неправильно зрозумілих інструментів у світі програмування. Багато розробників, коли asyncio був би оптимальним рішенням, звертаються до потоків (threads) або процесів, або ж використовують ключові слова async/await без глибокого розуміння їхньої справжньої механіки. Ця стаття детально розглядає корисні патерни та концепції, які ви зможете застосовувати щодня після того, як побудуєте правильну ментальну модель роботи асинхронного коду.

Проблема, яку вирішує asyncio, полягає у фундаментальній невідповідності між швидкістю обчислень процесора та повільністю операцій введення/виведення (I/O). Хоча ваш центральний процесор може виконувати мільярди операцій за секунду, мережевий запит або читання з диска займає десятки чи сотні мілісекунд. Якщо код просто чекає на завершення цих повільних операцій послідовно, мозок комп'ютера — CPU — більшу частину часу просто зростає в очікуванні.

Проблема I/O: Коли процесор чекає

Коли програма виконує низку операцій, що вимагають взаємодії з зовнішнім світом (база даних, API, файлова система), вона стикається з так званим вузьким місцем I/O. Розглянемо приклад:

result1 = fetch_from_database(query1) # очікування 100мс
result2 = call_external_api(url)      # очікування 200мс
result3 = read_large_file(path)       # очікування 50мс
# Загальний час: ~350мс

У цьому сценарії CPU не робить нічого корисного протягом більшої частини часу.

Потоки (Threading) — це один зі способів використати цей час, дозволяючи робити кілька речей одночасно. Однак потоки мають свої недоліки: вони вимагають пам'яті, потребують перемикання між завданнями та можуть містити помилки. Крім того, існує так званий GIL (Global Interpreter Lock), який обмежує паралельність у Python.

Asyncio працює інакше. Замість виконання багатьох речей одночасно, він виконує одну річ за раз. Але коли ця річ чекає на завершення I/O операції, asyncio перемикається на іншу задачу. Таким чином, asyncio використовує лише один потік і не споживає надто багато ресурсів.

Механізм роботи: Цикл подій (Event Loop) та корутини

Серцем asyncio є цикл подій (event loop). Він функціонує як планувальник завдань, який керує роботою програми. Його завдання такі:
1. Виконувати ко рутину до моменту досягнення ключового слова await.
2. Налаштовувати I/O операцію (наприклад, запит до мережі).
3. Переходити до наступної ко рутини.
4. Повертатися до першої ко рутини, коли I/O операція завершиться.

Функція asyncio.run() є сучасним точкам входу (Python 3.7+). Вона створює цикл подій, запускає вашу ко рутину та забезпечує її коректне очищення після виконання.

Ко рутина — це спеціальний тип функції, який визначається за допомогою async def. Коли ви викликаєте ко рутину, вона не починає виконуватися одразу, натомість вона повертає об'єкт ко рутини. Щоб змусити її працювати, необхідно використовувати await або запланувати її виконання.

Послідовне виконання: Якщо ви викликаєте ко рутини послідовно (кожна чекає завершення попередньої), це не є конкурентністю.
Конкурентність: Для досягнення справжньої конкурентності потрібні механізми планування, такі як asyncio.gather() або завдання (tasks).

Конкурентне виконання за допомогою asyncio.gather()

Щоб запустити кілька ко рутин одночасно і чекати на їхній спільний результат, використовується функція asyncio.gather(). Це дозволяє значно скоротити загальний час виконання.

Наприклад, якщо ми маємо 3 мережеві запити, кожен з яких займає ~1 секунду:

# Запуск усіх трьох одночасно
results = await asyncio.gather(
    fetch_user(1), fetch_user(2), fetch_user(3)
)
# Загальний час виконання буде близько 1с, а не 3с

У цьому випадку всі три завдання стартують негайно і завершуються приблизно одночасно.

Обробка помилок у gather()

При використанні gather() можна налаштувати обробку винятків так, щоб програма не варіювала при невдалому запиті. Встановлення параметра return_exceptions=True змушує функцію повертати самі об'єкти помилок як значення, замість того, щоб негайно викликати виняток.

results = await asyncio.gather(
    fetch_user(1), fetch_user(2), fetch_user(999), 
    return_exceptions=True # Повертаємо винятки як значення
)
for result in results:
    if isinstance(result, Exception):
        print(f"Помилка: {result}")
    else:
        print(result)

Завдання (Tasks): Запуск у фоновому режимі

Якщо вам потрібно запустити ко рутину і не чекати її завершення негайно — тобто виконати її "і забути", — використовується asyncio.create_task(). Ця функція планує виконання ко рутини в циклі подій, дозволяючи основному потоку продовжувати роботу без зупинки на очікування фонового завдання. Це ідеально підходить для завдань у фоновому режимі.