本文目录导读：
我来详细介绍优化Python异步任务执行效率的关键方法：
使用高性能异步框架
# 选择更高效的框架 # 比较：asyncio vs uvloop vs trio import asyncio import uvloop # 使用uvloop替代默认事件循环（性能提升2-4倍） asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
合理控制并发数量
import asyncio
from typing import List
async def process_task(task_id: int):
# 单个任务处理逻辑
await asyncio.sleep(0.1)
return task_id
async def optimized_concurrent_processing(tasks: List[int]):
"""使用信号量控制并发"""
semaphore = asyncio.Semaphore(10) # 限制并发数为10
async def limited_task(task_id):
async with semaphore:
return await process_task(task_id)
# 创建所有任务
coroutines = [limited_task(task) for task in tasks]
return await asyncio.gather(*coroutines)
# 批量处理，避免一次性创建过多协程
async def batch_processing(tasks: List[int], batch_size: int = 100):
results = []
for i in range(0, len(tasks), batch_size):
batch = tasks[i:i + batch_size]
batch_results = await asyncio.gather(*[
process_task(task) for task in batch
])
results.extend(batch_results)
return results
减少上下文切换开销
import asyncio
# 不好的实践：频繁的await
async def bad_practice():
for i in range(1000):
await asyncio.sleep(0) # 每个迭代都切换上下文
# 好的实践：批量处理
async def good_practice():
await asyncio.sleep(0) # 一次性让出控制权
for i in range(1000):
# 密集计算，不进行await
result = i * i
使用连接池和缓存
import asyncio
import aiohttp
from cachetools import TTLCache
class AsyncHttpClient:
def __init__(self):
# 使用连接池
self.session = aiohttp.ClientSession(
connector=aiohttp.TCPConnector(
limit=100, # 连接池大小
ttl_dns_cache=300, # DNS缓存时间
enable_cleanup_closed=True
)
)
# 使用缓存
self.cache = TTLCache(maxsize=1000, ttl=300)
async def fetch(self, url: str):
# 检查缓存
if url in self.cache:
return self.cache[url]
async with self.session.get(url) as response:
data = await response.text()
self.cache[url] = data
return data
async def close(self):
await self.session.close()
优化I/O操作
import asyncio
import aiofiles
# 异步文件操作
async def read_file_async(filepath: str):
async with aiofiles.open(filepath, 'r') as f:
content = await f.read()
return content
# 批量I/O操作
async def batch_io_operations(filepaths: list):
tasks = [read_file_async(path) for path in filepaths]
return await asyncio.gather(*tasks)
使用异步生成器
import asyncio
from typing import AsyncGenerator
# 使用异步生成器减少内存占用
async def async_data_stream() -> AsyncGenerator[int, None]:
for i in range(1000000):
yield i
# 每100个元素让出控制权
if i % 100 == 0:
await asyncio.sleep(0)
async def process_stream():
async for item in async_data_stream():
# 处理每个元素
pass
优化错误处理和超时
import asyncio
async def robust_task(task_id: int, timeout: float = 5.0):
try:
# 使用asyncio.wait_for设置超时
result = await asyncio.wait_for(
process_task(task_id),
timeout=timeout
)
return result
except asyncio.TimeoutError:
print(f"Task {task_id} timed out")
return None
except Exception as e:
print(f"Task {task_id} failed: {e}")
return None
# 优雅处理部分失败
async def handle_partial_failure(tasks: list):
done, pending = await asyncio.wait(
[robust_task(task) for task in tasks],
return_when=asyncio.FIRST_EXCEPTION
)
# 取消超时的任务
for task in pending:
task.cancel()
results = []
for completed in done:
try:
result = completed.result()
results.append(result)
except:
results.append(None)
return results
使用异步队列
import asyncio
from asyncio import Queue
async def producer(queue: Queue, items: list):
for item in items:
await queue.put(item)
await queue.put(None) # 发送结束信号
async def consumer(queue: Queue):
while True:
item = await queue.get()
if item is None:
break
# 处理item
await asyncio.sleep(0.1)
queue.task_done()
async def queue_based_processing(items: list):
queue = asyncio.Queue(maxsize=100) # 限制队列大小
# 启动多个consumer
consumers = [asyncio.create_task(consumer(queue)) for _ in range(5)]
# 启动producer
await producer(queue, items)
# 等待所有consumer完成
await asyncio.gather(*consumers)
性能监控和调优
import asyncio
import time
from functools import wraps
def async_timed(func):
@wraps(func)
async def wrapper(*args, **kwargs):
start = time.time()
try:
return await func(*args, **kwargs)
finally:
elapsed = time.time() - start
print(f"{func.__name__} took {elapsed:.3f} seconds")
return wrapper
@async_timed
async def monitored_async_function():
await asyncio.sleep(1)
# 使用asyncio.run的debug模式
asyncio.run(monitored_async_function(), debug=True)
完整优化示例
import asyncio
import uvloop
from typing import List, Any
import aiohttp
from cachetools import TTLCache
class OptimizedAsyncProcessor:
def __init__(self, max_concurrent: int = 50, cache_size: int = 1000):
# 使用uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
self.semaphore = asyncio.Semaphore(max_concurrent)
self.cache = TTLCache(maxsize=cache_size, ttl=300)
self.session = aiohttp.ClientSession(
connector=aiohttp.TCPConnector(
limit=max_concurrent,
ttl_dns_cache=300
)
)
async def process_item(self, item: Any) -> Any:
"""带信号量和缓存的异步处理"""
async with self.semaphore:
# 检查缓存
if item in self.cache:
return self.cache[item]
# 处理逻辑
result = await self._fetch_and_process(item)
# 更新缓存
self.cache[item] = result
return result
async def _fetch_and_process(self, item: Any) -> Any:
"""实际的获取和处理逻辑"""
async with self.session.get(f"http://api.example.com/{item}") as response:
data = await response.json()
# 处理数据...
return data
async def process_batch(self, items: List[Any], batch_size: int = 100) -> List[Any]:
"""批量处理，避免内存溢出"""
results = []
for i in range(0, len(items), batch_size):
batch = items[i:i + batch_size]
batch_results = await asyncio.gather(*[
self.process_item(item) for item in batch
])
results.extend(batch_results)
return results
async def cleanup(self):
"""清理资源"""
await self.session.close()
# 使用示例
async def main():
processor = OptimizedAsyncProcessor()
# 生成测试数据
items = list(range(1000))
# 处理数据
results = await processor.process_batch(items)
# 清理
await processor.cleanup()
print(f"Processed {len(results)} items")
if __name__ == "__main__":
asyncio.run(main())
关键优化要点
- 减少不必要的await：避免在循环中频繁await
- 控制并发数：使用信号量限制并发
- 使用连接池：复用网络连接
- 实现缓存：减少重复请求
- 批量处理：减少I/O操作次数
- 使用高效库：uvloop等高性能库
- 监控和调试：使用asyncio.run的debug模式
这些优化方法可以显著提升异步程序的执行效率,根据具体场景选择合适的优化策略。
