本文目录导读：
我来详细解释如何提升Python协程的执行效率,包括多个实用的优化策略：
使用合适的异步框架
uvloop - 更高效的事件循环
import asyncio
import uvloop
# 替换默认事件循环
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
async def main():
# 你的异步代码
pass
asyncio.run(main())
批量处理与并发控制
使用信号量限制并发
import asyncio
import aiohttp
async def fetch_with_semaphore(urls, max_concurrent=10):
semaphore = asyncio.Semaphore(max_concurrent)
async def fetch_one(url):
async with semaphore: # 控制并发数量
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
return await response.text()
tasks = [fetch_one(url) for url in urls]
return await asyncio.gather(*tasks)
避免阻塞操作
错误示例 - 阻塞调用
async def bad_example():
# 阻塞操作会阻塞整个事件循环
import time
time.sleep(1) # ❌ 阻塞所有协程
return "done"
正确示例 - 异步调用
import asyncio
async def good_example():
await asyncio.sleep(1) # ✅ 不会阻塞其他协程
return "done"
# 对于CPU密集型任务
async def cpu_intensive():
# 使用run_in_executor移到线程池
loop = asyncio.get_event_loop()
result = await loop.run_in_executor(
None, # 使用默认线程池
heavy_cpu_function, # CPU密集型函数
arg1, arg2
)
return result
合理使用任务组
asyncio.TaskGroup (Python 3.11+)
import asyncio
async def process_multiple():
async with asyncio.TaskGroup() as tg:
task1 = tg.create_task(task1_func())
task2 = tg.create_task(task2_func())
task3 = tg.create_task(task3_func())
# 所有任务完成后自动继续
results = [task1.result(), task2.result(), task3.result()]
return results
使用异步生成器
流式处理大数据
async def async_data_generator():
for i in range(10000):
# 模拟异步数据生成
await asyncio.sleep(0.001)
yield i
async def process_stream():
async for item in async_data_generator():
# 处理数据时不会阻塞其他协程
if item % 100 == 0:
print(f"Processing: {item}")
优化I/O操作
连接池复用
import aiohttp
class ConnectionPool:
def __init__(self):
self.session = None
async def __aenter__(self):
self.session = aiohttp.ClientSession(
connector=aiohttp.TCPConnector(limit=100) # 连接池大小
)
return self
async def __aexit__(self, *args):
await self.session.close()
async def fetch_many(self, urls):
tasks = [self.fetch_one(url) for url in urls]
return await asyncio.gather(*tasks)
async def fetch_one(self, url):
async with self.session.get(url) as response:
return await response.json()
使用异步上下文管理器
import asyncio
class AsyncResourceManager:
async def __aenter__(self):
# 初始化资源
self.resource = await self.acquire_resource()
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
# 清理资源
await self.release_resource()
async def acquire_resource(self):
await asyncio.sleep(0.1)
return "resource"
async def release_resource(self):
await asyncio.sleep(0.1)
# 使用
async def use_resource():
async with AsyncResourceManager() as manager:
# 使用资源
pass
监控和调试
使用asyncio调试模式
import asyncio
# 启用调试模式
asyncio.run(main(), debug=True)
# 或者在代码中设置
async def main():
loop = asyncio.get_event_loop()
loop.set_debug(True)
# 设置慢操作阈值
loop.slow_callback_duration = 0.05 # 50ms
性能对比示例
import asyncio
import time
async def slow_task(n):
await asyncio.sleep(0.1)
return n * 2
# 低效方式
async def inefficient():
results = []
for i in range(100):
result = await slow_task(i) # 串行执行
results.append(result)
return results
# 高效方式
async def efficient():
tasks = [slow_task(i) for i in range(100)]
return await asyncio.gather(*tasks) # 并发执行
# 测试
async def benchmark():
start = time.time()
await inefficient()
print(f"Inefficient: {time.time() - start:.2f}s")
start = time.time()
await efficient()
print(f"Efficient: {time.time() - start:.2f}s")
# 输出示例:
# Inefficient: 10.01s
# Efficient: 0.11s
最佳实践总结
- 选择合适的库：使用
uvloop、aiohttp等高性能库 - 控制并发度：使用
Semaphore避免资源耗尽 - 避免阻塞：所有阻塞操作都要使用异步版本
- 复用连接：使用连接池减少开销
- 批量处理：尽可能使用
gather或TaskGroup - 合理分割任务：将大任务拆分为小任务
- 使用队列：对于生产者-消费者模式
- 监控性能：启用调试模式，找出瓶颈
这些优化策略可以显著提升Python协程的执行效率,具体选择哪些策略取决于你的应用场景和业务需求。
