在上一節,我們講述了最簡單最基礎的后線程的建立,現在我們將進行拓展
Flask應用中處理異步事件(后臺線程+事件循環)的方法(1)
在我們的實際應用當中,我們需要定義三個東西
-
一個多線程的信號旗,這里的信號旗指的是多線程編程中一個非常基礎且重要的同步工具
它只有兩種狀態:
Clear (清除/未設置): 信號燈是紅色的,旗子是放下的。
Set (設置): 信號燈是綠色的,旗子是舉起的。
當進程調用event.wait(),如果此時事件是 “Clear” 狀態(紅燈),那么這個線程就會停下來,進入等待狀態,被阻塞。 -
一個 循環的對象
-
一個新線程
接下來在這里我們看到了,有一個target參數,這個就是調用我們的一個說明書
# 1. 創建一個線程同步工具mcp_ready_event = threading.Event()# 2. 創建一個新的 asyncio 事件循環對象background_event_loop = asyncio.new_event_loop()# 3. 創建一個新線程,目標是運行 start_event_loop 函數loop_thread = threading.Thread(target=start_event_loop, args=(background_event_loop, mcp_ready_event),daemon=Truename="name")loop_thread.start() # 啟動這個新線程# 4. 主線程在這里暫停,等待后臺線程通知logging.info("主線程正在等待 MCP 初始化...")is_ready = mcp_ready_event.wait(timeout=30.0) # 暫停30秒,等待加載
那么我們的說明書怎么寫呢?這里舉例mcp的連接初始化
初始化這里,無論成功還是失敗,初始化流程走完后,調用 .set() 來釋放正在 wait() 的主線程。
# ... existing code ...
async def initialize_mcp_connection(ready_event: threading.Event):"""在應用啟動時啟動MCP服務器并建立一個持久的會話。完成后通過 event 通知主線程。"""# ...try:# 初始化代碼except Exception as e:# ... 錯誤處理 ...finally:logging.info("MCP 初始化過程結束,通知主線程。")ready_event.set()
初始化完畢之后,我們就進行接下來的開始線程
這里的參數 loop: asyncio.AbstractEventLoop 和 ready_event: threading.Event 是什么意思?
這是一種現代 Python 的語法,叫做 類型提示 (Type Hinting)。它本身不強制任何東西,但它極大地增強了代碼的可讀性和可維護性,并且能被很多開發工具利用來進行代碼檢查。
- loop: asyncio.AbstractEventLoop: 這行代碼的意思是,我們期望 loop 這個參數是一個 asyncio.AbstractEventLoop 類型的對象。它告訴任何閱讀或調用這個函數的人:“請給我一個 asyncio 的事件循環對象。”
- ready_event: threading.Event: 同理,這表示我們期望 ready_event 這個參數是一個我們剛才講的 threading.Event 類型的對象。
def start_event_loop(loop: asyncio.AbstractEventLoop, ready_event: threading.Event):"""在一個線程中啟動事件循環,并在啟動時安排MCP初始化。"""asyncio.set_event_loop(loop) loop.create_task(initialize_mcp_connection(ready_event)) logging.info("后臺事件循環已啟動,MCP 初始化任務已安排。")loop.run_forever()
需要注意這里start_event_loop 函數中的 loop,和主線程中的 background_event_loop,雖然名字不同,但它們指向的是內存中同一個事件循環對象。
最后只需要運行一個專門用于從一個普通線程向一個正在運行 asyncio 事件循環的線程提交任務。
asyncio.run_coroutine_threadsafe(coro, loop):