Сообщить об ошибке.

Функции создания и получения текущего цикла событий.

Цикл событий - это ядро ​​любого приложения на основе модуля asyncio. Объект цикла событий, своими методами запускает асинхронные задачи и обратные вызовы, выполняют сетевые операции ввода-вывода, а так же запускают подпроцессы.

Разработчики приложений, для запуска цикла событий асинхронного кода, должны использовать высокоуровневые функции модуля asyncio, такие как asyncio.run(), так как редко нужно ссылаться на объект цикла или вызывать его методы. Низкоуровневые функции и объекты (низкоуровневый API) предназначен для авторов библиотек и фреймворков, которым требуется более тонкий контроль над поведением цикла событий.

Функции получения, установки и создания цикла событий.

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


asyncio.get_running_loop():

Функция asyncio.get_running_loop() возвращает запущенный цикл событий в текущем потоке ОС.

Если нет запущенного цикла событий, то возникает ошибка RuntimeError. Эта функция может быть вызвана только из сопрограммы или обратного вызова.

asyncio.get_event_loop():

Функция asyncio.get_event_loop() получает текущий цикл событий.

Если в текущем потоке ОС не установлен текущий цикл событий, то поток ОС считается основным и функция asyncio.set_event_loop() еще не была вызвана. При этом модуль asyncio автоматически создаст новый цикл событий и установит его как текущий.

Так как функция asyncio.get_event_loop() имеет довольно сложное поведение, особенно при использовании настраиваемых политик цикла событий, то использование функции asyncio.get_running_loop() предпочтительнее, чем asyncio.get_event_loop() в сопрограммах и обратных вызовах.

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

asyncio.set_event_loop(loop):

Функция asyncio.set_event_loop() устанавливает цикл loop как текущий цикл событий для текущего потока ОС.

asyncio.new_event_loop():

Метод asyncio.new_event_loop() создает новый объект цикла событий loop.

Возможности низкоуровнего API цикла событий:

Реализации цикла событий.

Asyncio поставляется с двумя разными реализациями цикла событий: SelectorEventLoop и ProactorEventLoop.

По умолчанию asyncio настроен на использование SelectorEventLoop в Unix и ProactorEventLoop в Windows.

asyncio.SelectorEventLoop:

Класс asyncio.SelectorEventLoop представляет собой цикл событий, основанный на модуле selectors.

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

import asyncio
import selectors

selector = selectors.SelectSelector()
loop = asyncio.SelectorEventLoop(selector)
asyncio.set_event_loop(loop)

Доступность: Unix, Windows.

asyncio.ProactorEventLoop:

Класс asyncio.ProactorEventLoop представляет собой цикл событий для Windows, использующий порты завершения ввода/вывода IOCP.

Доступность: Windows.

Более подробно смотрите документацию MSDN по портам завершения ввода/вывода.

asyncio.AbstractEventLoop:

Класс asyncio.AbstractEventLoop представляет собой абстрактный базовый класс для asyncio-совместимых циклов событий.

В подразделе "Возможности низкоуровнего API цикла событий" перечислены разделы, в которых представлена документация по низкоуровневым методам, которые должна была определить альтернативная реализация asyncio.AbstractEventLoop.

Примеры использования низкоуровнего API циклов событий:

Обратите внимание, что пример целенаправленно показывают, как использовать низкоуровневые API-интерфейсы цикла событий, таких как loop.run_forever() и loop.call_soon(). Современные приложения asyncio очень редко необходимо писать таким образом.

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

Привет, мир с loop.call_soon().

Пример использования метода loop.call_soon() для планирования обратного вызова. Обратный вызов отображает "Hello World", а затем останавливает цикл событий.

import asyncio

def hello_world(loop):
    """
    Обратный вызов для печати "Hello World" 
    и остановки цикла событий
    """
    print('Hello World')
    # прерываем цикл событий
    loop.stop()

# получение текущего цикла событий
loop = asyncio.get_event_loop()

# Планирует вызов функции `hello_world()`
loop.call_soon(hello_world, loop)

# Блокирующий вызов прерывается loop.stop()
try:
    loop.run_forever()
finally:
    # освобождение ресурсов
    # потребляемых циклом событий
    loop.close()