快捷導(dǎo)航

python asyncio常規(guī)操作記錄

更新時間：2026年03月21日 14:01:58 作者：mingshili

本文介紹了Python asyncio的核心并發(fā)機制和API,本文通過實例代碼給大家介紹的非常詳細,對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值,需要的朋友參考下吧

Event Loop 執(zhí)行機制

Event loop 本質(zhì)是單線程內(nèi)的無限循環(huán)調(diào)度器，不斷從任務(wù)隊列取出就緒的任務(wù)執(zhí)行：

while True:
    events = poll_for_ready_events()   # 檢查就緒事件（IO完成、定時器到期、回調(diào)投遞）
    for event in events:
        execute(event.callback)        # 逐個執(zhí)行

協(xié)程遇到 await 時掛起自身、讓出控制權(quán)給 loop，loop 去執(zhí)行其他就緒任務(wù)；
等待的操作完成后，loop 恢復(fù)該協(xié)程繼續(xù)執(zhí)行。

一個線程最多運行一個 event loop。

三個核心 API 對比

1.asyncio.create_task(coro)

在當(dāng)前 loop 內(nèi)并發(fā)調(diào)度一個協(xié)程，立即返回 asyncio.Task，不阻塞。

async def main():
    task1 = asyncio.create_task(fetch_a())   # 立即調(diào)度，不等完成
    task2 = asyncio.create_task(fetch_b())   # 兩個協(xié)程并發(fā)運行
    result_a = await task1                   # 需要結(jié)果時再 await
    result_b = await task2

適用場景： 同一個 loop 內(nèi)同時跑多個異步 IO 任務(wù)（并發(fā)請求、并發(fā)監(jiān)聽等）。

2.asyncio.to_thread(func, *args)

把同步阻塞函數(shù)丟到線程池執(zhí)行，當(dāng)前協(xié)程掛起等待結(jié)果，但 event loop 不阻塞。

async def main():
    # time.sleep 是同步阻塞的，直接調(diào)用會卡死整個 loop
    await asyncio.to_thread(time.sleep, 5)   # 丟到線程池，loop 繼續(xù)處理別的
    # 典型場景：CPU 密集計算、沒有 async 版本的阻塞 IO 庫
    data = await asyncio.to_thread(read_big_file, path)

適用場景： 調(diào)用沒有 async 版本的阻塞函數(shù)（文件 IO、CPU 計算、舊的同步庫），
避免卡死 event loop。

await會卡住當(dāng)前協(xié)程嗎？

當(dāng)前協(xié)程會等待，但 event loop 和其他協(xié)程不受影響：

寫法	當(dāng)前協(xié)程	event loop	其他協(xié)程
`time.sleep(5)`	卡	卡	全卡
`await asyncio.to_thread(time.sleep, 5)`	等	不卡	正常跑
`await asyncio.sleep(5)`	等	不卡	正常跑

3.asyncio.run_coroutine_threadsafe(coro, loop)

從另一個線程向指定 loop 投遞一個協(xié)程，返回 concurrent.futures.Future。

# 在非 asyncio 的普通線程中：
future = asyncio.run_coroutine_threadsafe(play_audio(data), main_loop)
result = future.result(timeout=10)   # 阻塞等待結(jié)果（在當(dāng)前線程阻塞，不影響 loop）

適用場景： 普通線程需要讓某個 event loop 執(zhí)行一個協(xié)程（跨線程投遞異步任務(wù)）。

相關(guān) API：call_soon_threadsafe

loop.call_soon_threadsafe(callback, *args)

與 run_coroutine_threadsafe 類似，但投遞的是普通回調(diào)函數(shù)而非協(xié)程。

# 從 WebSocket 線程把音頻流注冊到主線程的 Mixer
self._main_loop.call_soon_threadsafe(mixer.add_stream, stream)

總結(jié)對比表

API	調(diào)用位置	投遞目標(biāo)	方向	返回值
`create_task(coro)`	協(xié)程內(nèi)	協(xié)程 → 同一個 loop	async → async	`asyncio.Task`
`to_thread(func)`	協(xié)程內(nèi)	同步函數(shù) → 線程池	async → sync → async	awaitable
`run_coroutine_threadsafe(coro, loop)`	任意線程	協(xié)程 → 另一個線程的 loop	sync → async	`concurrent.futures.Future`
`call_soon_threadsafe(cb)`	任意線程	回調(diào) → 另一個線程的 loop	sync → sync(callback)	`None`

速記口訣

create_task              → 同 loop 內(nèi)并發(fā)跑協(xié)程
to_thread                → 把同步阻塞的東西丟出去，別卡我的 loop
run_coroutine_threadsafe → 從外面往別人的 loop 里塞協(xié)程
call_soon_threadsafe     → 從外面往別人的 loop 里塞回調(diào)

項目中的實際應(yīng)用

本項目 AudioReceiverServer 中存在兩個線程各自運行一個 event loop：

主線程 event loop                     WebSocket 線程 event loop
┌───────────────────┐                ┌───────────────────┐
│  Mixer 協(xié)程        │                │  ws.recv() 協(xié)程 A  │
│  其他音頻處理 ...   │  ?── call_soon │  ws.recv() 協(xié)程 B  │
│                   │   threadsafe   │  ws.recv() 協(xié)程 C  │
└───────────────────┘                └───────────────────┘