Нужно ли отслеживать завершение каждой корутины вручную?

В асинхронном программировании на Python корутины — это функции, определённые с async def, которые могут приостанавливать своё выполнение. Когда вы запускаете корутину, она не выполняется сразу, а возвращает объект корутины, который нужно запустить в цикле событий (event loop).

Как запускаются корутины

Есть два основных способа запуска:

• Ожидание (await): Используя await coroutine(), вы приостанавливаете текущую корутину до завершения вызванной. Это простой способ, но он выполняет задачи последовательно.
• Создание задачи (Task): Используя asyncio.create_task(), вы оборачиваете корутину в объект Task и "запускаете её в фоне". Цикл событий будет управлять её выполнением конкурентно с другими задачами.

Нужно ли отслеживать завершение вручную?

При использовании create_task явное отслеживание не требуется — цикл событий выполнит задачу. Однако возникает ключевая проблема: если задача выбросит исключение, и её результат никто не ожидает (не делает await), это исключение может быть потеряно, что затрудняет отладку.

Поэтому, хотя сам факт завершения можно не отслеживать, отслеживание исключений и результатов — важно. Для этого используют несколько подходов.

Практические примеры управления задачами

Пример 1: Запуск задачи и игнорирование результата (риск потерянных исключений).

import asyncio

async def my_coroutine():
    await asyncio.sleep(1)
    print("Задача завершена")

async def main():
    # Задача запущена, но не ожидается
    task = asyncio.create_task(my_coroutine())
    # main продолжает работу, не дожидаясь task
    await asyncio.sleep(2)

asyncio.run(main())

Пример 2: Безопасный сбор результатов нескольких задач с помощью asyncio.gather.

async def fetch_data(url):
    await asyncio.sleep(0.5)
    return f"Данные с {url}"

async def main():
    tasks = [
        fetch_data("https://api.example.com/1"),
        fetch_data("https://api.example.com/2")
    ]
    # gather запускает все корутины и ждёт их завершения
    results = await asyncio.gather(*tasks, return_exceptions=True)
    for result in results:
        if isinstance(result, Exception):
            print(f"Ошибка: {result}")
        else:
            print(result)

Пример 3: Мониторинг задач с помощью asyncio.wait для более гибкого контроля.

async def worker(name, delay):
    await asyncio.sleep(delay)
    return f"{name} done"

async def main():
    tasks = [asyncio.create_task(worker(f"Task-{i}", i)) for i in range(1, 4)]
    # Ждём завершения всех задач
    done, pending = await asyncio.wait(tasks, return_when=asyncio.ALL_COMPLETED)
    for task in done:
        print(task.result())

Вывод

Явно отслеживать завершение каждой корутины не обязательно, так как цикл событий сделает это. Однако для создания надёжных приложений необходимо предусмотреть обработку исключений и сбор результатов, особенно когда задачи запускаются "в фоне". Используйте asyncio.gather или asyncio.wait для управления группой задач и всегда обрабатывайте потенциальные ошибки.