Python脚本如何统计协程任务执行耗时

本文目录导读：

1. 方法1：使用时间戳（最简单）
2. 方法2：使用装饰器统计单个协程耗时
3. 方法3：使用上下文管理器（推荐）
4. 方法4：统计所有协程的详细耗时
5. 方法5：使用asyncio的内置功能
6. 方法6：高级统计（包含异常处理）
7. 总结建议

在Python中统计协程任务执行耗时,有几种常见方法：

方法1：使用时间戳（最简单）

import asyncio
import time
async def my_task(task_id):
    print(f"任务 {task_id} 开始")
    await asyncio.sleep(2)  # 模拟耗时操作
    print(f"任务 {task_id} 结束")
async def main():
    start_time = time.time()
    # 执行多个协程
    tasks = [my_task(i) for i in range(3)]
    await asyncio.gather(*tasks)
    total_time = time.time() - start_time
    print(f"所有任务总耗时: {total_time:.2f} 秒")
asyncio.run(main())

方法2：使用装饰器统计单个协程耗时

import asyncio
import time
from functools import wraps
def async_timer(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        start_time = time.time()
        result = await func(*args, **kwargs)
        elapsed = time.time() - start_time
        print(f"{func.__name__} 执行耗时: {elapsed:.4f} 秒")
        return result
    return wrapper
@async_timer
async def my_task(task_id):
    await asyncio.sleep(task_id)
    return f"任务 {task_id} 完成"
async def main():
    tasks = [my_task(i) for i in range(1, 4)]
    results = await asyncio.gather(*tasks)
    print("结果:", results)
asyncio.run(main())

方法3：使用上下文管理器（推荐）

import asyncio
import time
from contextlib import asynccontextmanager
@asynccontextmanager
async def timer_context(name="任务"):
    start_time = time.time()
    yield
    elapsed = time.time() - start_time
    print(f"{name} 耗时: {elapsed:.2f} 秒")
async def my_task(task_id):
    await asyncio.sleep(task_id)
async def main():
    async with timer_context("总任务"):
        async with timer_context(f"子任务"):
            await my_task(2)
        await my_task(1)
asyncio.run(main())

方法4：统计所有协程的详细耗时

import asyncio
import time
class TaskTimer:
    def __init__(self):
        self.tasks = {}
    def start_task(self, task_id):
        self.tasks[task_id] = time.time()
    def end_task(self, task_id):
        if task_id in self.tasks:
            elapsed = time.time() - self.tasks[task_id]
            del self.tasks[task_id]
            return elapsed
        return None
    def get_stats(self):
        return self.tasks
async def my_task(task_id, timer):
    timer.start_task(task_id)
    print(f"任务 {task_id} 开始")
    await asyncio.sleep(task_id)
    elapsed = timer.end_task(task_id)
    print(f"任务 {task_id} 耗时: {elapsed:.2f} 秒")
    return task_id
async def main():
    timer = TaskTimer()
    tasks = [my_task(i, timer) for i in range(1, 4)]
    results = await asyncio.gather(*tasks)
    print("所有任务完成:", results)
asyncio.run(main())

方法5：使用asyncio的内置功能

import asyncio
import time
async def my_task(task_id):
    await asyncio.sleep(task_id)
    return f"任务 {task_id}"
async def main():
    # 创建任务，获取完成时间
    tasks = [asyncio.ensure_future(my_task(i)) for i in range(1, 4)]
    start = time.time()
    done, pending = await asyncio.wait(tasks)
    total = time.time() - start
    print(f"总耗时: {total:.2f} 秒")
    print(f"完成的任务: {len(done)}")
    print(f"未完成的任务: {len(pending)}")
    # 获取每个任务的结果
    for task in done:
        print(f"结果: {task.result()}")
asyncio.run(main())

方法6：高级统计（包含异常处理）

import asyncio
import time
from typing import Dict, List, Tuple
class AsyncTaskProfiler:
    def __init__(self):
        self.timings: Dict[str, float] = {}
    async def profile(self, coro, name=None):
        """分析单个协程"""
        name = name or coro.__name__
        start = time.time()
        try:
            result = await coro
            elapsed = time.time() - start
            self.timings[name] = elapsed
            return result
        except Exception as e:
            elapsed = time.time() - start
            self.timings[name] = elapsed
            raise e
    async def profile_all(self, coros: List) -> Tuple:
        """分析多个协程"""
        tasks = []
        for coro in coros:
            task = asyncio.create_task(self.profile(coro))
            tasks.append(task)
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return results
    def get_report(self) -> str:
        """生成统计报告"""
        if not self.timings:
            return "没有统计数据"
        total = sum(self.timings.values())
        report = ["任务执行报告:", "-" * 30]
        for name, timing in self.timings.items():
            percentage = (timing / total) * 100 if total > 0 else 0
            report.append(f"{name}: {timing:.3f}秒 ({percentage:.1f}%)")
        report.append("-" * 30)
        report.append(f"总计: {total:.3f}秒")
        return "\n".join(report)
async def main():
    profiler = AsyncTaskProfiler()
    # 定义几个异步任务
    async def task_a():
        await asyncio.sleep(1)
        return "A完成"
    async def task_b():
        await asyncio.sleep(2)
        return "B完成"
    async def task_c():
        await asyncio.sleep(0.5)
        return "C完成"
    # 分析所有任务
    coros = [task_a(), task_b(), task_c()]
    results = await profiler.profile_all(coros)
    # 输出报告
    print(profiler.get_report())
    print("结果:", results)
asyncio.run(main())

总结建议

1. 简单场景：使用方法1（时间戳）最直接
2. 需要复用：使用方法2（装饰器）或方法3（上下文管理器）
3. 需要详细统计：使用方法4（自定义类）或方法6（完整分析器）
4. 生产环境：考虑使用专门的性能分析库如 asyncio-profiler 或 yappi

推荐使用方法3（上下文管理器），因为它：

• 代码干净,易于阅读
• 可以精确控制计时范围
• 支持嵌套计时
• 易于扩展和维护