Python中的loop.run_in_executor基本用法
更新時間：2025年12月15日 11:42:57 作者：無風(fēng)聽海
文章介紹了在Python的異步編程中使用`loop.run_in_executor`來執(zhí)行阻塞操作的方法,包括基本用法、線程池和進(jìn)程池的選擇、與協(xié)程的差異對比、進(jìn)階用法、注意事項以及真實場景舉例,感興趣的朋友跟隨小編一起看看吧
? 一、背景：為什么需要loop.run_in_executor
在 Python 的異步編程（asyncio）中，協(xié)程可以并發(fā)運行，提高效率，但它們依賴于“非阻塞的 I/O”。如果你在協(xié)程中調(diào)用了一個阻塞的操作（比如 time.sleep()、requests.get() 等），它會阻塞整個事件循環(huán)，導(dǎo)致其他協(xié)程也無法繼續(xù)執(zhí)行。
為了解決這個問題，Python 提供了 loop.run_in_executor()，允許你把阻塞的同步代碼“放在后臺線程或進(jìn)程中執(zhí)行”，從而不影響主事件循環(huán)。
?? 二、基本用法和執(zhí)行流程
? 基本結(jié)構(gòu)：
loop.run_in_executor(executor, func, *args)
executor: 指定使用的執(zhí)行器（線程池或進(jìn)程池），為None時使用默認(rèn)線程池。func: 要執(zhí)行的同步阻塞函數(shù)。*args: 傳遞給函數(shù)的參數(shù)。
? 最小示例（阻塞函數(shù)放到線程中）：
import asyncio
import time
def blocking_func(name):
print(f"開始阻塞任務(wù) {name}")
time.sleep(3)
print(f"結(jié)束阻塞任務(wù) {name}")
return f"{name} done"
async def main():
loop = asyncio.get_running_loop()
# 把阻塞函數(shù)丟進(jìn)默認(rèn)線程池
result = await loop.run_in_executor(None, blocking_func, "任務(wù)A")
print(result)
asyncio.run(main())? 輸出：
開始阻塞任務(wù) 任務(wù)A
結(jié)束阻塞任務(wù) 任務(wù)A
任務(wù)A done
注意：雖然
blocking_func()是阻塞的，但不會阻塞asyncio主事件循環(huán)，因此你可以同時運行其他協(xié)程。
?? 三、線程池 vs 進(jìn)程池
1. ThreadPoolExecutor（默認(rèn)）
- 用于 I/O 密集型任務(wù)（網(wǎng)絡(luò)請求、文件 I/O 等）
- 啟動快，線程共享內(nèi)存，效率高
run_in_executor(None, ...)就是使用默認(rèn)線程池
2. ProcessPoolExecutor
- 用于 CPU 密集型任務(wù)（圖像處理、數(shù)據(jù)加密、科學(xué)計算等）
- 每個進(jìn)程獨立內(nèi)存，更耗資源，但避免 GIL 限制
- 用于充分利用多核 CPU
示例（使用 ProcessPoolExecutor）：
from concurrent.futures import ProcessPoolExecutor
def compute(n):
return sum(i * i for i in range(n))
async def main():
loop = asyncio.get_running_loop()
with ProcessPoolExecutor() as executor:
result = await loop.run_in_executor(executor, compute, 10_000_000)
print(result)
asyncio.run(main())?? 四、run_in_executor 與協(xié)程的差異對比
| 特性 | 協(xié)程 (async def) | run_in_executor |
|---|---|---|
| 是否阻塞事件循環(huán) | 否 | 否 |
| 適用于 | 異步 I/O 操作 | 同步阻塞操作 |
| 是否需要線程或進(jìn)程 | 否 | 是（線程或進(jìn)程池） |
| 是否自動并發(fā) | 是 | 是 |
| 是否可中斷 | 可以使用 asyncio.CancelledError | 線程執(zhí)行不能中斷 |
?? 五、與其他異步寫法對比
? 錯誤做法（會阻塞整個事件循環(huán)）：
import asyncio
import time
async def wrong():
time.sleep(2) # 阻塞整個事件循環(huán)！
print("完成")
asyncio.run(wrong())? 正確做法（使用run_in_executor）：
async def correct():
loop = asyncio.get_running_loop()
await loop.run_in_executor(None, time.sleep, 2)
print("完成")
?? 六、進(jìn)階用法：并行多個任務(wù)
import asyncio
import time
def task(name, duration):
print(f"開始 {name}")
time.sleep(duration)
print(f"結(jié)束 {name}")
return name
async def main():
loop = asyncio.get_running_loop()
tasks = [
loop.run_in_executor(None, task, 'A', 2),
loop.run_in_executor(None, task, 'B', 3),
loop.run_in_executor(None, task, 'C', 1),
]
results = await asyncio.gather(*tasks)
print("全部完成：", results)
asyncio.run(main())? 輸出（并行）：
開始 A
開始 B
開始 C
結(jié)束 C
結(jié)束 A
結(jié)束 B
全部完成： ['A', 'B', 'C']
?? 七、注意事項與潛在陷阱
| 問題 | 描述與解決 |
|---|---|
| 共享資源問題 | 多線程操作同一個變量可能會出錯，考慮加鎖或使用 asyncio.Queue |
| 線程池大小限制 | 默認(rèn)線程池大小有限（通常為 CPU 核心數(shù)的 5 倍），可手動調(diào)整 |
| 異常處理 | 在線程中拋出的異常必須在主線程中 await 時捕獲 |
| 進(jìn)程池不能用 lambda | 進(jìn)程池中的函數(shù)必須是可序列化的，不能是匿名函數(shù)或本地函數(shù) |
| 不能中斷線程任務(wù) | 一旦 run_in_executor 開始執(zhí)行函數(shù)，無法強(qiáng)制中斷線程任務(wù) |
?? 八、真實場景舉例
示例 1：讀取大文件（I/O 密集型）
def read_file(path):
with open(path, 'r') as f:
return f.read()
data = await loop.run_in_executor(None, read_file, "bigfile.txt")示例 2：調(diào)用同步網(wǎng)絡(luò)庫（如requests）
import requests
def fetch(url):
response = requests.get(url)
return response.text
html = await loop.run_in_executor(None, fetch, "https://example.com")實際建議：使用
httpx.AsyncClient替代requests
?? 九、封裝通用工具函數(shù)（推薦寫法）
import asyncio
from typing import Callable, Any
from functools import partial
async def to_thread(func: Callable, *args, **kwargs) -> Any:
loop = asyncio.get_running_loop()
return await loop.run_in_executor(None, partial(func, *args, **kwargs))
# 使用方式
result = await to_thread(my_blocking_function, arg1, arg2)? 十、Python 3.9+ 新特性：asyncio.to_thread
從 Python 3.9 起，你可以直接用內(nèi)置的 asyncio.to_thread() 來代替 run_in_executor(None, ...)，更加簡潔：
import asyncio
def blocking_func():
...
await asyncio.to_thread(blocking_func)等價于：
await loop.run_in_executor(None, blocking_func)
?? 總結(jié)
| 特點 | 內(nèi)容 |
|---|---|
| 功能 | 異步執(zhí)行阻塞的同步函數(shù) |
| 用法 | await loop.run_in_executor(executor, func, *args) |
| 默認(rèn)線程池 | executor=None |
| 用于場景 | 文件 I/O、網(wǎng)絡(luò)請求、同步數(shù)據(jù)庫操作、CPU 密集計算 |
| 替代方案 | Python 3.9+: asyncio.to_thread() |
到此這篇關(guān)于Python中的loop.run_in_executor基本用法的文章就介紹到這了,更多相關(guān)Python loop.run_in_executor內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
您可能感興趣的文章:
- Python解決多線程運行異步代碼報錯"There?is?no?current?event?loop"
- python uvloop事件循環(huán)庫使用功能示例探究
- 簡單理解Python中的事件循環(huán)EventLoop
- python定時任務(wù)timeloop庫用法實例詳解
- 詳解python ThreadPoolExecutor異常捕獲
- python中ThreadPoolExecutor線程池和ProcessPoolExecutor進(jìn)程池
- python 多進(jìn)程并行編程 ProcessPoolExecutor的實現(xiàn)
- Python線程池模塊ThreadPoolExecutor用法分析
相關(guān)文章
Python經(jīng)驗總結(jié):兩種Type?Error問題
這篇文章主要介紹了Python經(jīng)驗總結(jié):兩種Type?Error問題,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2023-09-09
Python中使用Counter進(jìn)行字典創(chuàng)建以及key數(shù)量統(tǒng)計的方法
今天小編就為大家分享一篇Python中使用Counter進(jìn)行字典創(chuàng)建以及key數(shù)量統(tǒng)計的方法，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2018-07-07
python數(shù)據(jù)可視化Seaborn畫熱力圖
這篇文章主要介紹了數(shù)據(jù)可視化Seaborn畫熱力圖，熱力圖的想法其實很簡單，用顏色替換數(shù)字，下面我們來看看文章對操作過程的具體介紹吧，需要的小伙伴可以參考一下具體內(nèi)容，希望對你有所幫助2022-01-01
這篇文章主要介紹了Python的列表和列表算法，小編感覺這篇文章具有一定參考價值，需要的朋友可以了解下，希望能給你帶來幫助2021-08-08
這篇文章主要介紹了python實現(xiàn)括號匹配及匹配格式的相關(guān)知識，非常不錯，具有一定的參考借鑒價值,需要的朋友可以參考下2018-08-08
今天小編就為大家分享一篇對Tensorflow中的變量初始化函數(shù)詳解，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2018-07-07
這篇文章主要跟大家介紹了關(guān)于Python中內(nèi)置常量的相關(guān)資料，文中介紹的非常詳細(xì)，對大家具有一定的參考學(xué)習(xí)價值，需要的朋友們下面來一起看吧。2017-05-05
經(jīng)常獲得了一個用戶提交的當(dāng)前日期，我們需要以這個日期為依據(jù)返回它的前一天、后一天的日期或者轉(zhuǎn)換操作等。用Python可以非常簡單的解決這些關(guān)于日期計算的問題2014-03-03
這篇文章主要為大家詳細(xì)介紹了python實現(xiàn)尋找最長回文子序列的方法，具有一定的參考價值，感興趣的小伙伴們可以參考一下2018-06-06
python concurrent.futures模塊的使用測試
大家都知道concurrent.futures 是 3.2 中引入的新模塊，它為異步執(zhí)行可調(diào)用對象提供了高層接口，今天通過本文給大家介紹python concurrent.futures模塊的使用測試 ，感興趣的朋友一起看看吧2021-07-07