文章目录
- loop.run_in_executor(executor, func, *args) executor: 指定使用的执行器(线程池或进程池),为 None 时使用默认线程池。 func: 要执行的同步阻塞函数。 *args: 传递给函数的参数。
- import asyncio import time def blocking_func(name): print(f”开始阻塞任务 {name}”) time.sleep(3) print(f”结束阻塞任务 {name}”) return f”{name} done” async def main(): loop = asyncio.get_running_loop() # 把阻塞函数丢进默认线程池 result = await loop.run_in_executor(None, blocking_func, “任务A”) print(result) asyncio.run(main())
- 开始阻塞任务 任务A结束阻塞任务 任务A任务A done 注意:虽然 blocking_func() 是阻塞的,但不会阻塞 asyncio 主事件循环,因此你可以同时运行其他协程。
- def read_file(path): with open(path, ‘r’) as f: return f.read() data = await loop.run_in_executor(None, read_file, “bigfile.txt”)
- 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
目录
- ✅ 一、背景:为什么需要loop.run_in_executor
- 🧪 二、基本用法和执行流程
- 🧵 三、线程池 vs 进程池
- 1. ThreadPoolExecutor(默认)
- 2. ProcessPoolExecutor
- 示例(使用 ProcessPoolExecutor):
- ❌ 错误做法(会阻塞整个事件循环):
- ✅ 正确做法(使用run_in_executor):
- ⏳ 输出(并行):
- 示例 1:读取大文件(I/O 密集型)
- 示例 2:调用同步网络库(如requests)
- 📊 四、run_in_executor 与协程的差异对比
- 🔁 五、与其他异步写法对比
- 🔐 六、进阶用法:并行多个任务
- ⚠️ 七、注意事项与潜在陷阱
- 💼 八、真实场景举例
- 🧩 九、封装通用工具函数(推荐写法)
- ✅ 十、Python 3.9+ 新特性:asyncio.to_thread
- 📘 总结
在 Python 的异步编程(asyncio)中,协程可以并发运行,提高效率,但它们依赖于“非阻塞的 I/O”。如果你在协程中调用了一个阻塞的操作(比如 time.sleep()、requests.get() 等),它会阻塞整个事件循环,导致其他协程也无法继续执行。
为了解决这个问题,Python 提供了 loop.run_in_executor(),允许你把阻塞的同步代码“放在后台线程或进程中执行”,从而不影响主事件循环。
loop.run_in_executor(executor, func, *args)
executor: 指定使用的执行器(线程池或进程池),为 None 时使用默认线程池。
func: 要执行的同步阻塞函数。
*args: 传递给函数的参数。
executor: 指定使用的执行器(线程池或进程池),为 None 时使用默认线程池。func: 要执行的同步阻塞函数。*args: 传递给函数的参数。
import asyncio
import time
def blocking_func(name):
print(f"开始阻塞任务 {name}")
time.sleep(3)
print(f"结束阻塞任务 {name}")
return f"{name} done"
async def main():
loop = asyncio.get_running_loop()
# 把阻塞函数丢进默认线程池
result = await loop.run_in_executor(None, blocking_func, "任务A")
print(result)
asyncio.run(main())
开始阻塞任务 任务A
结束阻塞任务 任务A
任务A done
注意:虽然 blocking_func() 是阻塞的,但不会阻塞 asyncio 主事件循环,因此你可以同时运行其他协程。
开始阻塞任务 任务A
结束阻塞任务 任务A
任务A done
注意:虽然 blocking_func() 是阻塞的,但不会阻塞 asyncio 主事件循环,因此你可以同时运行其他协程。
- 用于 I/O 密集型任务(网络请求、文件 I/O 等)
- 启动快,线程共享内存,效率高
run_in_executor(None, ...) 就是使用默认线程池
run_in_executor(None, ...) 就是使用默认线程池
- 用于 CPU 密集型任务(图像处理、数据加密、科学计算等)
- 每个进程独立内存,更耗资源,但避免 GIL 限制
- 用于充分利用多核 CPU
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())
| 特性 | 协程 (async def) | run_in_executor |
|---|---|---|
| 是否阻塞事件循环 | 否 | 否 |
| 适用于 | 异步 I/O 操作 | 同步阻塞操作 |
| 是否需要线程或进程 | 否 | 是(线程或进程池) |
| 是否自动并发 | 是 | 是 |
| 是否可中断 | 可以使用 asyncio.CancelledError |
线程执行不能中断 |
import asyncio
import time
async def wrong():
time.sleep(2) # 阻塞整个事件循环!
print("完成")
asyncio.run(wrong())
async def correct():
loop = asyncio.get_running_loop()
await loop.run_in_executor(None, time.sleep, 2)
print("完成")
import asyncio
import time
def task(name, duration):
print(f"开始 {name}")
time.sleep(duration)
print(f"结束 {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
结束 C
结束 A
结束 B
全部完成: ['A', 'B', 'C']
开始 A
开始 B
开始 C
结束 C
结束 A
结束 B
全部完成: ['A', 'B', 'C']
| 问题 | 描述与解决 |
|---|---|
| 共享资源问题 | 多线程操作同一个变量可能会出错,考虑加锁或使用 asyncio.Queue |
| 线程池大小限制 | 默认线程池大小有限(通常为 CPU 核心数的 5 倍),可手动调整 |
| 异常处理 | 在线程中抛出的异常必须在主线程中 await 时捕获 |
| 进程池不能用 lambda | 进程池中的函数必须是可序列化的,不能是匿名函数或本地函数 |
| 不能中断线程任务 | 一旦 run_in_executor 开始执行函数,无法强制中断线程任务 |
def read_file(path):
with open(path, 'r') as f:
return f.read()
data = await loop.run_in_executor(None, read_file, "bigfile.txt")
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
实际建议:使用 httpx.AsyncClient 替代 requests
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() 来代替 run_in_executor(None, ...),更加简洁:
import asyncio
def blocking_func():
...
await asyncio.to_thread(blocking_func)
等价于:
await loop.run_in_executor(None, blocking_func)
| 特点 | 内容 |
|---|---|
| 功能 | 异步执行阻塞的同步函数 |
| 用法 | await loop.run_in_executor(executor, func, *args) |
| 默认线程池 | executor=None |
| 用于场景 | 文件 I/O、网络请求、同步数据库操作、CPU 密集计算 |
| 替代方案 | Python 3.9+: asyncio.to_thread() |
到此这篇关于Python中的loop.run_in_executor基本用法的文章就介绍到这了,更多相关Python loop.run_in_executor内容请搜索风君子博客以前的文章或继续浏览下面的相关文章希望大家以后多多支持风君子博客!
您可能感兴趣的文章:
- Python解决多线程运行异步代码报错"There is no current event loop"
- python uvloop事件循环库使用功能示例探究
- 简单理解Python中的事件循环EventLoop
- python定时任务timeloop库用法实例详解
- 详解python ThreadPoolExecutor异常捕获
- python中ThreadPoolExecutor线程池和ProcessPoolExecutor进程池
- python 多进程并行编程 ProcessPoolExecutor的实现
- Python线程池模块ThreadPoolExecutor用法分析