策略

警告

策略已被棄用,並將在 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())