BlockBuster 簡介：我的非同步事件循環被封鎖了嗎？

Python 3.5 引入了非同步 I/O 作為執行緒的替代方案來處理並發。非同步 I/O 和 Python 中的 asyncio 實現的優勢在於，透過不產生記憶體消耗大的作業系統線程，系統使用更少的資源並且更具可擴展性。此外，在 asyncio 中，調度點透過 await 語法明確定義，而在基於線程的並發中，GIL 可能會在難以預測的程式碼點釋放。因此，基於 asyncio 的並發系統更容易理解和調試。最終，可以取消 asyncio 任務，而這在使用執行緒時不容易做到。

但是，為了真正受益於這些優勢，在非同步協程中避免阻塞呼叫非常重要。阻塞呼叫可以是網路呼叫、檔案系統呼叫、sleep 呼叫等等。這些阻塞呼叫是有害的，因為在底層，asyncio 使用單執行緒事件循環來並發運行協程。因此，如果在協程中進行阻塞調用，它會阻塞整個事件循環和所有協程，從而影響應用程式的整體效能。

以下是一個阻塞呼叫阻止程式碼並發執行的範例：

import asyncio
import datetime
import time

async def example(name):
    print(f"{datetime.datetime.now()}: {name} start")
    time.sleep(1)  # time.sleep 是一个阻塞函数
    print(f"{datetime.datetime.now()}: {name} stop")

async def main():
    await asyncio.gather(example("1"), example("2"))

asyncio.run(main())

運行結果類似：

<code>2025-01-07 18:50:15.327677: 1 start
2025-01-07 18:50:16.328330: 1 stop
2025-01-07 18:50:16.328404: 2 start
2025-01-07 18:50:17.333159: 2 stop</code>

可以看到，兩個協程沒有並發運行。

為了克服這個問題，你需要使用非阻塞等效項或將執行延遲到執行緒池：

import asyncio
import datetime
import time

async def example(name):
    print(f"{datetime.datetime.now()}: {name} start")
    await asyncio.sleep(1)  # 将阻塞的 time.sleep 调用替换为非阻塞的 asyncio.sleep 协程
    print(f"{datetime.datetime.now()}: {name} stop")

async def main():
    await asyncio.gather(example("1"), example("2"))

asyncio.run(main())

運行結果類似：

<code>2025-01-07 18:53:53.579738: 1 start
2025-01-07 18:53:53.579797: 2 start
2025-01-07 18:53:54.580463: 1 stop
2025-01-07 18:53:54.580572: 2 stop</code>

這裡兩個協程並發運行。

現在的問題是，並不總是很容易識別一個方法是否阻塞。特別是如果程式碼庫很大或使用第三方函式庫。有時，阻塞呼叫是在程式碼的深層部分進行的。

例如，這段程式碼是否阻塞？

import blockbuster
from importlib.metadata import version

async def get_version():
    return version("blockbuster")

Python 是否在啟動時將包元資料載入記憶體？是在載入 blockbuster 模組時完成的嗎？或當我們呼叫 version() 時？結果是否被緩存，後續呼叫將是非阻塞的嗎？正確答案是在呼叫 version() 時完成的，它涉及讀取已安裝套件的 METADATA 檔案。且結果沒有被緩存。因此，version() 是一個阻塞調用，應該始終推遲到線程中。如果不深入研究 importlib 的程式碼，很難​​知道這個事實。

偵測阻塞呼叫的一種方法是啟動 asyncio 的偵錯模式來記錄耗時過長的阻塞呼叫。但這並不是最有效的方法，因為許多短於觸發超時時間的阻塞仍然會損害效能，並且測試/開發中的阻塞時間可能與生產環境中的阻塞時間不同。例如，如果資料庫必須獲取大量數據，則資料庫呼叫在生產環境中可能需要更長時間。

這就是 BlockBuster 發揮作用的地方！啟動後，BlockBuster 將修補幾個阻塞的 Python 框架方法，如果它們從 asyncio 事件循環調用，則會引發錯誤。預設修補的方法包括 os、io、time、socket、sqlite 模組的方法。有關 BlockBuster 偵測到的方法的完整列表，請參閱項目自述文件。然後，你可以在單元測試或開發模式中啟動 BlockBuster 來捕捉任何阻塞呼叫並修復它們。如果你知道 JVM 中很棒的 BlockHound 函式庫，它的原理相同，但適用於 Python。 BlockHound 是 BlockBuster 的一個很好的靈感來源，感謝創作者。

讓我們看看如何在上面阻塞程式碼片段上使用 BlockBuster。

首先，我們要安裝 blockbuster 套件

import asyncio
import datetime
import time

async def example(name):
    print(f"{datetime.datetime.now()}: {name} start")
    time.sleep(1)  # time.sleep 是一个阻塞函数
    print(f"{datetime.datetime.now()}: {name} stop")

async def main():
    await asyncio.gather(example("1"), example("2"))

asyncio.run(main())

然後，我們可以使用 pytest fixture 和 blockbuster_ctx() 方法來在每個測試開始時啟動 BlockBuster，並在拆卸期間停用它。

<code>2025-01-07 18:50:15.327677: 1 start
2025-01-07 18:50:16.328330: 1 stop
2025-01-07 18:50:16.328404: 2 start
2025-01-07 18:50:17.333159: 2 stop</code>

如果你用 pytest 運行這個，你會得到

import asyncio
import datetime
import time

async def example(name):
    print(f"{datetime.datetime.now()}: {name} start")
    await asyncio.sleep(1)  # 将阻塞的 time.sleep 调用替换为非阻塞的 asyncio.sleep 协程
    print(f"{datetime.datetime.now()}: {name} stop")

async def main():
    await asyncio.gather(example("1"), example("2"))

asyncio.run(main())

注意: 通常，在一個真實的項目中，blockbuster() fixture 將在一個 conftest.py 檔案中設定。

結論

我相信 BlockBuster 在 asyncio 專案中非常有用。它已經幫助我在我參與的專案中檢測到許多阻塞呼叫問題。但這並不是靈丹妙藥。特別是，有些第三方函式庫不使用 Python 框架方法來與網路或檔案系統交互，而是包裝 C 函式庫。對於這些庫，可以在測試設定中新增規則來觸發這些庫的阻塞呼叫。 BlockBuster 也是開源的：非常歡迎貢獻，以便在核心專案中為您的最喜歡的庫添加規則。如果你看到問題和可以改進的地方，我很樂意在專案問題追蹤器中收到你的回饋。

一些連結：

• GitHub 專案
• 問題
• Pypi 上的包包

以上是BlockBuster 簡介：我的非同步事件循環被封鎖了嗎？的詳細內容。更多資訊請關注PHP中文網其他相關文章！