Python脚本如何提升协程执行效率

wen python案例 2026-06-27 240

本文目录导读：

Python脚本如何提升协程执行效率

使用合适的异步框架
批量处理与并发控制
避免阻塞操作
合理使用任务组
使用异步生成器
优化I/O操作
使用异步上下文管理器
监控和调试
性能对比示例
最佳实践总结

我来详细解释如何提升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