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

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

Основное решение для Python 3.7 и новее — запускать верхнеуровневую корутину через asyncio.run() и не закрывать loop вручную:

import asyncio


async def main() -> None:
    await asyncio.sleep(0.1)
    print("Работа завершена")


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

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

Почему возникает ошибка

Закрытый event loop нельзя использовать повторно. После вызова loop.close() методы планирования задач и callback-функций могут завершаться исключением RuntimeError. Само закрытие необратимо.

Типичная ошибочная последовательность выглядит так:

import asyncio

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

loop.run_until_complete(asyncio.sleep(0.1))
loop.close()

# Ошибка: цикл уже закрыт.
loop.run_until_complete(asyncio.sleep(0.1))

Похожая проблема возникает, когда функция или класс закрывает цикл, хотя цикл был создан вызывающим кодом:

import asyncio


def run_job(loop: asyncio.AbstractEventLoop) -> None:
    loop.run_until_complete(asyncio.sleep(0.1))
    loop.close()  # Неправильно: функция не владеет циклом.


loop = asyncio.new_event_loop()
run_job(loop)

# Вызывающий код ожидает, что loop ещё доступен.
loop.run_until_complete(asyncio.sleep(0.1))

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

Не исправляйте ошибку постоянным вызовом asyncio.new_event_loop() перед каждой операцией. Это скрывает нарушение жизненного цикла, оставляет незавершённые задачи и может приводить к утечкам сетевых соединений, файловых дескрипторов и потоков executor.

Шаг 1. Найдите место закрытия цикла

Изучите traceback и найдите первый вызов вашего кода перед строкой из asyncio. Затем проверьте проект на следующие операции:

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

Если loop передаётся между объектами, временно добавьте проверку непосредственно перед проблемной операцией:

if loop.is_closed():
    raise RuntimeError("Попытка использовать уже закрытый event loop")

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

Шаг 2. Оставьте один верхнеуровневый запуск

Частая причина ошибки — несколько последовательных вызовов asyncio.run() при повторном использовании объектов, привязанных к предыдущему циклу:

import asyncio


class Client:
    def __init__(self) -> None:
        self.loop = None

    async def connect(self) -> None:
        self.loop = asyncio.get_running_loop()

    async def send(self) -> None:
        self.loop.call_soon(lambda: None)


client = Client()

asyncio.run(client.connect())
asyncio.run(client.send())  # client хранит ссылку на закрытый loop.

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

Перенесите все операции в одну корутину:

import asyncio


class Client:
    async def connect(self) -> None:
        self.loop = asyncio.get_running_loop()

    async def send(self) -> None:
        self.loop.call_soon(lambda: None)


async def main() -> None:
    client = Client()
    await client.connect()
    await client.send()


asyncio.run(main())

Ещё надёжнее не сохранять loop в поле объекта без необходимости. Внутри корутины получайте текущий цикл через asyncio.get_running_loop() непосредственно перед использованием.

Шаг 3. Не закрывайте loop внутри библиотечного кода

Функция, принимающая существующий цикл, должна оставить его открытым:

import asyncio


async def perform_job() -> str:
    await asyncio.sleep(0.1)
    return "done"


def run_with_existing_loop(
    loop: asyncio.AbstractEventLoop,
) -> str:
    return loop.run_until_complete(perform_job())

Закрытие выполняется только там, где цикл был создан:

loop = asyncio.new_event_loop()

try:
    asyncio.set_event_loop(loop)
    result = run_with_existing_loop(loop)
    print(result)
finally:
    loop.close()

Для обычного приложения ручное управление не требуется: предпочтительнее asyncio.run(). Ручной loop оправдан, когда приложение интегрируется с фреймворком, GUI, сервером или другим компонентом, который задаёт собственный жизненный цикл.

Шаг 4. Завершите фоновые задачи до закрытия

Ошибка может появляться не в основном коде, а в callback-функции, которая срабатывает уже после начала завершения. Перед закрытием цикла необходимо остановить производителей новых задач: таймеры, очереди, сетевые клиенты, наблюдатели файлов и рабочие потоки.

В приложении с asyncio.run() храните ссылки на созданные задачи и отменяйте их внутри main():

import asyncio


async def worker() -> None:
    try:
        while True:
            await asyncio.sleep(1)
    except asyncio.CancelledError:
        # Здесь закрываются ресурсы конкретного worker.
        raise


async def main() -> None:
    task = asyncio.create_task(worker())

    try:
        await asyncio.sleep(0.1)
    finally:
        task.cancel()
        await asyncio.gather(task, return_exceptions=True)


asyncio.run(main())

task.cancel() только запрашивает отмену. Нужно передать управление циклу и дождаться завершения задачи. В примере это делает asyncio.gather(). Без ожидания код очистки внутри finally у фоновой корутины может не выполниться до закрытия loop.

Если задача использует сетевой клиент, поток, subprocess или файл, закрывайте ресурс до отмены зависимой инфраструктуры. Для объектов с асинхронным контекстным менеджером используйте async with:

async def main() -> None:
    async with create_client() as client:
        await client.send_request()

Конкретный метод создания и закрытия клиента зависит от используемой библиотеки. Нельзя заменять документированный метод освобождения ресурса произвольным sleep().

Шаг 5. При ручном управлении соблюдайте порядок shutdown

Следующий шаблон предназначен для Python 3.9 и новее, поскольку использует loop.shutdown_default_executor():

import asyncio


async def main() -> None:
    await asyncio.sleep(0.1)


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()

Порядок действий здесь принципиален:

основная корутина прекращает создание новой работы;
оставшиеся задачи получают запрос на отмену;
цикл выполняется до завершения отменённых задач;
закрываются асинхронные генераторы;
завершается executor по умолчанию;
только после этого вызывается loop.close().

Для Python 3.7–3.8 метод shutdown_default_executor() в этом шаблоне неприменим. Следует использовать документацию именно установленной ветки Python и не копировать вызов отсутствующего API.

Шаг 6. Остановите внешние потоки и callback-источники

Рабочий поток может сохранять ссылку на loop и отправлять в него callback через call_soon_threadsafe(). Если главный поток уже закрыл loop, очередной callback завершится ошибкой.

Правильная последовательность завершения:

установить флаг остановки рабочего потока;
прекратить получение новых событий из внешнего источника;
дождаться выхода потока;
отменить оставшиеся asyncio-задачи;
закрыть event loop.

Проверка loop.is_closed() внутри потока может уменьшить число ошибочных вызовов, но не устраняет гонку. Надёжное решение — гарантировать, что поток завершился до loop.close().

Шаг 7. Включите отладочный режим asyncio

Для диагностики можно включить документированный debug-режим:

import asyncio


async def main() -> None:
    await asyncio.sleep(0.1)


asyncio.run(main(), debug=True)

Он помогает обнаруживать некоторые неправильно ожидаемые корутины, медленные callback-функции и небезопасные вызовы API из другого потока. Диагностические сообщения не заменяют анализ traceback: ошибка всё равно исправляется изменением порядка запуска и завершения.

Отдельный случай: уже запущенный event loop

В интерактивной среде, серверном фреймворке или GUI цикл событий может уже принадлежать платформе. В таком контексте не следует безусловно вызывать asyncio.run() или закрывать полученный loop.

Если код уже выполняется внутри async def, вызывайте корутину через await:

async def handler() -> None:
    result = await perform_job()
    print(result)

Сообщение asyncio.run() cannot be called from a running event loop отличается от Event loop is closed, хотя обе ошибки часто появляются после попыток вручную управлять циклом. Не применяйте исправление для одной ошибки к другой без проверки фактического traceback.

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

После изменения кода выполните сценарии нормального завершения, обработки исключения и принудительной отмены операции. Исправление можно считать корректным, когда выполняются все условия:

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

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

Источники

Ссылки ведут на общую ветку документации Python 3. Доступ к веб-проверке в момент подготовки материала отсутствовал, поэтому для версионно-зависимого поведения следует дополнительно открыть документацию конкретной установленной версии Python, выбрав её в переключателе версий на сайте.