Как исправить RuntimeError: Event loop is closed в asyncio

Ошибка RuntimeError: Event loop is closed означает, что код пытается запланировать задачу, выполнить callback, закрыть сетевой транспорт или обратиться к объекту asyncio после закрытия цикла событий. Исправление сводится к двум правилам: цикл должен закрываться только после завершения всех асинхронных операций, а объекты, привязанные к циклу, нельзя повторно использовать после его закрытия.

Примеры ниже рассчитаны на CPython с модулем asyncio. Основной рекомендуемый вариант для обычных программ — один вызов asyncio.run() в точке входа. Для библиотек, Jupyter, тестовых фреймворков и приложений с собственным циклом событий способ запуска необходимо согласовывать с управляющей средой.

1. Найдите место, где цикл закрывается раньше времени

Сначала изучите полный traceback, а не только последнюю строку. Важен первый вызов из вашего кода перед переходом во внутренние функции asyncio. Обычно проблема относится к одной из следующих ситуаций:

после asyncio.run() повторно используется асинхронный клиент, сессия, очередь, блокировка или задача;
цикл закрывается вручную через loop.close(), пока ещё существуют незавершённые задачи;
фоновая задача создана через asyncio.create_task(), но программа завершается, не дождавшись её;
асинхронный ресурс закрывается уже после закрытия цикла;
синхронная функция многократно вызывает asyncio.run(), сохраняя между вызовами объекты из предыдущего цикла;
код запускается внутри среды, где цикл событий уже работает.

Для диагностики временно включите режим отладки. Он помогает обнаруживать забытые coroutine-объекты, медленные callbacks и некорректное завершение задач:

import asyncio

async def main():
    ...

if __name__ == "__main__":
    asyncio.run(main(), debug=True)

Также можно запустить программу с переменной окружения PYTHONASYNCIODEBUG=1. Это штатный механизм отладки asyncio, но сообщения зависят от конкретного сценария и версии Python.

2. Используйте один asyncio.run() в точке входа

asyncio.run() создаёт новый цикл событий, выполняет переданную coroutine, завершает асинхронные генераторы, закрывает executor и затем закрывает цикл. После возврата из asyncio.run() этот цикл использовать нельзя.

Корректная структура программы:

import asyncio

async def load_data():
    await asyncio.sleep(0.1)
    return {"status": "ok"}

async def main():
    result = await load_data()
    print(result)

if __name__ == "__main__":
    asyncio.run(main())

Проблемная структура — несколько отдельных запусков, между которыми сохраняется объект, связанный с первым циклом:

import asyncio

queue = None

async def create_queue():
    global queue
    queue = asyncio.Queue()

async def use_queue():
    await queue.put("value")

asyncio.run(create_queue())
asyncio.run(use_queue())

Первый вызов завершает и закрывает свой цикл. Во втором вызове создаётся уже другой цикл. Даже если конкретный объект в определённой версии Python не выдаёт ошибку сразу, такая архитектура ненадёжна: асинхронные задачи и ресурсы должны создаваться и использоваться в пределах одного запуска.

Исправленный вариант:

import asyncio

async def main():
    queue = asyncio.Queue()
    await queue.put("value")
    print(await queue.get())

if __name__ == "__main__":
    asyncio.run(main())

Не храните в глобальных переменных задачи, Future, сетевые клиенты, соединения, очереди, блокировки и другие асинхронные объекты, если программа может запускать несколько независимых циклов событий. Создавайте их внутри общей coroutine и закрывайте до выхода из неё.

3. Не закрывайте цикл вручную без необходимости

В прикладном коде ручное управление через asyncio.new_event_loop(), run_until_complete() и loop.close() обычно не требуется. Ошибка часто появляется из-за безусловного закрытия цикла в блоке finally, когда фоновые операции ещё не завершены.

Рискованный вариант:

import asyncio

loop = asyncio.new_event_loop()

try:
    loop.run_until_complete(main())
finally:
    loop.close()

Сам по себе этот шаблон допустим только при полном контроле над жизненным циклом. Если main() создаёт фоновые задачи и не ожидает их завершения, они останутся незавершёнными к моменту loop.close().

Для обычной программы замените ручное управление на:

if __name__ == "__main__":
    asyncio.run(main())

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

import asyncio

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

try:
    loop.run_until_complete(main())
finally:
    pending = asyncio.all_tasks(loop)

    for task in pending:
        task.cancel()

    if pending:
        loop.run_until_complete(
            asyncio.gather(*pending, return_exceptions=True)
        )

    loop.run_until_complete(loop.shutdown_asyncgens())
    loop.run_until_complete(loop.shutdown_default_executor())
    loop.close()

Этот код применим, когда приложение само создаёт цикл и полностью им управляет. Не используйте его внутри фреймворка или интерактивной среды, которые владеют циклом самостоятельно. Метод shutdown_default_executor() относится к управлению executor цикла; при использовании asyncio.run() отдельный вызов не требуется.

4. Дожидайтесь фоновых задач

asyncio.create_task() запускает coroutine конкурентно, но не делает её автоматически частью ожидаемого результата. Если сохранить задачу и не выполнить await, основной код может завершиться раньше.

Проблемный пример:

import asyncio

async def worker():
    await asyncio.sleep(1)
    print("done")

async def main():
    asyncio.create_task(worker())

asyncio.run(main())

main() завершается сразу после создания задачи. При завершении asyncio.run() оставшиеся задачи отменяются. Если задача или используемая библиотека некорректно обрабатывает завершение, последующие действия могут обратиться к уже закрытому циклу.

Исправление — явно дождаться задачи:

async def main():
    task = asyncio.create_task(worker())
    await task

Для нескольких независимых операций используйте asyncio.gather():

async def main():
    await asyncio.gather(
        worker(),
        worker(),
        worker(),
    )

В Python, где доступен asyncio.TaskGroup, структурированную конкурентность можно оформить так:

async def main():
    async with asyncio.TaskGroup() as group:
        group.create_task(worker())
        group.create_task(worker())

TaskGroup ожидает завершения созданных в нём задач при выходе из контекстного менеджера и организует обработку ошибок группы. Наличие этого API следует проверять по документации именно установленной версии Python.

5. Закрывайте асинхронные ресурсы до завершения main()

Сетевые клиенты, соединения с базой данных и другие ресурсы могут выполнять асинхронные действия при закрытии. Если их деструктор срабатывает после завершения asyncio.run(), он уже не сможет обратиться к закрытому циклу.

Проблемная схема:

client = None

async def main():
    global client
    client = AsyncClient()
    await client.request()

asyncio.run(main())

# Закрытие выполняется слишком поздно
asyncio.run(client.close())

Второй asyncio.run() создаёт другой цикл, тогда как клиент мог быть связан с первым. Правильный подход — создать, использовать и закрыть ресурс в одной coroutine:

async def main():
    client = AsyncClient()

    try:
        await client.request()
    finally:
        await client.close()

asyncio.run(main())

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

async def main():
    async with AsyncClient() as client:
        await client.request()

Названия методов close(), aclose() и поддержка async with зависят от конкретной библиотеки. Проверяйте её официальную документацию и не подменяйте синхронное закрытие асинхронным или наоборот.

6. Не вызывайте asyncio.run() внутри уже работающего цикла

В Jupyter, GUI-фреймворке, веб-сервере или асинхронном тестовом окружении цикл событий может уже выполняться. Вызов asyncio.run() из работающего цикла не является корректным способом запуска вложенной coroutine.

Внутри асинхронной функции используйте обычный await:

async def handler():
    result = await load_data()
    return result

В интерактивной среде с поддержкой top-level await выполняйте:

result = await load_data()

Не закрывайте цикл, полученный от фреймворка, сервера или интерактивной оболочки. Его жизненным циклом управляет сама среда. Попытка вызвать loop.close() может нарушить последующие запросы, тесты или ячейки.

7. Проверьте повторное использование объекта после первого запуска

Характерный сценарий — класс создаёт асинхронный клиент один раз, а его метод вызывается через отдельный asyncio.run() несколько раз:

service = Service()

asyncio.run(service.fetch())
asyncio.run(service.fetch())

Если Service хранит внутри соединение, задачу, Future или пул, второй вызов может обратиться к объекту первого, уже закрытого цикла.

Исправьте одним из двух способов. Первый — выполнить все операции в одном запуске:

async def main():
    service = Service()

    try:
        await service.fetch()
        await service.fetch()
    finally:
        await service.close()

asyncio.run(main())

Второй — создавать полностью новый экземпляр ресурса для каждого независимого запуска:

async def run_once():
    service = Service()

    try:
        await service.fetch()
    finally:
        await service.close()

asyncio.run(run_once())
asyncio.run(run_once())

Второй способ допустим только тогда, когда экземпляры действительно не разделяют асинхронное состояние.

8. Добавьте контролируемое завершение приложения

Серверы, боты и длительно работающие процессы должны завершаться в определённом порядке:

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

Минимальный шаблон с событием остановки:

import asyncio

async def worker(stop_event):
    while not stop_event.is_set():
        await asyncio.sleep(0.5)

async def main():
    stop_event = asyncio.Event()
    task = asyncio.create_task(worker(stop_event))

    try:
        await asyncio.sleep(2)
    finally:
        stop_event.set()
        await task

if __name__ == "__main__":
    asyncio.run(main())

Если задача может зависнуть, стратегию тайм-аута и отмены нужно проектировать отдельно. Не закрывайте цикл как способ принудительно остановить работающую coroutine.

Как проверить исправление

Запустите программу с asyncio.run(main(), debug=True).
Выполните сценарий, который ранее приводил к ошибке.
Убедитесь, что полный traceback больше не содержит обращения к закрытому циклу.
Проверьте, что не появились предупреждения Task was destroyed but it is pending и coroutine was never awaited.
Повторите запуск или операцию несколько раз, если ошибка возникала при повторном вызове.
Проверьте ветку аварийного завершения: исключение, отмену задачи и остановку приложения.

Итоговый чек-лист

В программе есть одна основная coroutine и один вызов asyncio.run().
Все фоновые задачи сохраняются и ожидаются через await, gather() или группу задач.
Асинхронные ресурсы закрываются до завершения основной coroutine.
Объекты из одного цикла не используются в другом.
loop.close() не вызывается для цикла, которым управляет внешняя среда.
При ручном управлении циклом незавершённые задачи отменяются и обрабатываются до закрытия.
Исправление проверено в режиме отладки asyncio.