策略

警告

策略已被弃用,并将在 Python 3.16 中移除。建议用户使用 asyncio.run() 函数或带有 loop_factoryasyncio.Runner 来使用所需的循环实现。

事件循环策略是一个全局对象,用于获取和设置当前的事件循环,以及创建新的事件循环。默认策略可以被替换内置替代方案以使用不同的事件循环实现,或者被自定义策略替换,该策略可以覆盖这些行为。

策略对象为每个 上下文 获取和设置一个单独的事件循环。默认情况下,这是按线程的,尽管自定义策略可以以不同的方式定义 上下文

自定义事件循环策略可以控制 get_event_loop()set_event_loop()new_event_loop() 的行为。

策略对象应实现 AbstractEventLoopPolicy 抽象基类中定义的 API。

获取和设置策略

以下函数可用于获取和设置当前进程的策略

asyncio.get_event_loop_policy()

返回当前进程范围的策略。

自 3.14 版本弃用:get_event_loop_policy() 函数已弃用,并将在 Python 3.16 中移除。

asyncio.set_event_loop_policy(policy)

将当前进程范围的策略设置为 policy

如果 policy 设置为 None,则恢复默认策略。

自 3.14 版本弃用:set_event_loop_policy() 函数已弃用,并将在 Python 3.16 中移除。

策略对象

抽象事件循环策略基类定义如下

class asyncio.AbstractEventLoopPolicy

asyncio 策略的抽象基类。

get_event_loop()

获取当前上下文的事件循环。

返回实现 AbstractEventLoop 接口的事件循环对象。

此方法不应返回 None

在 3.6 版本中更改。

set_event_loop(loop)

将当前上下文的事件循环设置为 loop

new_event_loop()

创建并返回一个新的事件循环对象。

此方法不应返回 None

自 3.14 版本弃用:AbstractEventLoopPolicy 类已弃用,并将在 Python 3.16 中移除。

asyncio 附带以下内置策略

class asyncio.DefaultEventLoopPolicy

默认的 asyncio 策略。在 Unix 上使用 SelectorEventLoop,在 Windows 上使用 ProactorEventLoop

无需手动安装默认策略。asyncio 已配置为自动使用默认策略。

在 3.8 版本中更改:在 Windows 上,现在默认使用 ProactorEventLoop

在 3.14 版本中更改:如果未设置事件循环,默认 asyncio 策略的 get_event_loop() 方法现在会引发 RuntimeError

自 3.14 版本弃用:DefaultEventLoopPolicy 类已弃用,并将在 Python 3.16 中移除。

class asyncio.WindowsSelectorEventLoopPolicy

使用 SelectorEventLoop 事件循环实现的替代事件循环策略。

可用性: Windows。

自 3.14 版本弃用:WindowsSelectorEventLoopPolicy 类已弃用,并将在 Python 3.16 中移除。

class asyncio.WindowsProactorEventLoopPolicy

使用 ProactorEventLoop 事件循环实现的替代事件循环策略。

可用性: Windows。

自 3.14 版本弃用:WindowsProactorEventLoopPolicy 类已弃用,并将在 Python 3.16 中移除。

自定义策略

要实现新的事件循环策略,建议子类化 DefaultEventLoopPolicy 并重写需要自定义行为的方法,例如。

class MyEventLoopPolicy(asyncio.DefaultEventLoopPolicy):

    def get_event_loop(self):
        """Get the event loop.

        This may be None or an instance of EventLoop.
        """
        loop = super().get_event_loop()
        # Do something with loop ...
        return loop

asyncio.set_event_loop_policy(MyEventLoopPolicy())