4. 代码执行时间测量：装饰器、上下文管理器与多线程陷阱

说到性能基准测试，最基础也最关键的，就是怎么把代码执行时间量准。

我见过不少新手，直接拿 time.time() 包一下就开始跑。结果呢？数据忽高忽低，完全没法用。说白了，计时这件事，看着简单，坑却不少。

这一章，我带你从三个角度搞定它：装饰器、上下文管理器，以及多线程/多进程下的计时陷阱。

核心观点：计时的本质不是「记下时间」，而是「排除干扰」。装饰器解决复用问题，上下文管理器解决资源管理问题，多线程解决并发干扰问题。

代码执行时间测量 装饰器计时 @timer 一键复用 上下文管理器 with Timer(): 精确控制 多线程/进程陷阱 GIL、上下文切换 functools.wraps 统计次数/均值 支持参数化 __enter__/__exit__ 异常安全 GIL 干扰 time.sleep 不准确 进程间同步开销

4.1 装饰器计时：最优雅的复用方案

我个人习惯，在项目里写一个通用的 @timer 装饰器。为什么？因为只要往函数上一贴，就能自动打印耗时，不用改函数内部一行代码。

import time
import functools

def timer(repeat=1):
    """测量函数执行时间的装饰器，支持重复执行取均值"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            start = time.perf_counter()
            for _ in range(repeat):
                result = func(*args, **kwargs)
            end = time.perf_counter()
            avg_time = (end - start) / repeat
            print(f"[{func.__name__}] 平均耗时: {avg_time*1000:.3f} ms (重复{repeat}次)")
            return result
        return wrapper
    return decorator

# 使用示例
@timer(repeat=5)
def compute_pnl(prices):
    return sum(p * 0.001 for p in prices)

compute_pnl([100, 102, 101, 103, 99])
# 输出: [compute_pnl] 平均耗时: 0.012 ms (重复5次)

我的经验：一定要用 time.perf_counter() 而不是 time.time()。前者是单调时钟，不受系统时间调整影响。我在一个回测框架里吃过亏——系统时钟被 NTP 同步往回跳了 0.5 秒，结果所有耗时数据全乱了。

装饰器还有个好处：可以统计调用次数。我在做高频交易策略回测时，经常需要知道某个函数被调了多少次，平均耗时多少。装饰器里加个计数器，一劳永逸。

def timer_with_stats(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        wrapper.call_count += 1
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        wrapper.total_time += elapsed
        return result
    wrapper.call_count = 0
    wrapper.total_time = 0.0
    return wrapper

@timer_with_stats
def fast_calc(x):
    return x ** 2

for _ in range(100):
    fast_calc(42)

print(f"调用次数: {fast_calc.call_count}, 总耗时: {fast_calc.total_time:.4f}s")
# 输出: 调用次数: 100, 总耗时: 0.0003s

4.2 上下文管理器计时：精确控制生命周期

装饰器适合整个函数计时。但有时候，我只想测一段代码块，不想为了计时单独拆一个函数。这时候，上下文管理器就派上用场了。

import time

class Timer:
    """上下文管理器，用于测量代码块执行时间"""
    def __init__(self, name="block"):
        self.name = name
    
    def __enter__(self):
        self.start = time.perf_counter()
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.elapsed = time.perf_counter() - self.start
        print(f"[{self.name}] 耗时: {self.elapsed*1000:.3f} ms")
        # 如果发生异常，返回 False 让异常继续传播
        return False

# 使用示例
with Timer("数据加载"):
    data = [i for i in range(10**6)]
    
with Timer("计算处理"):
    result = sum(d * 2 for d in data)

注意：上下文管理器里的 __exit__ 方法，如果返回 True，会吞掉异常。我建议始终返回 False，让异常正常抛出。否则你可能会发现代码静默失败了，找半天 bug。

你想想看，上下文管理器还有一个隐藏优势：可以嵌套使用。我在优化一个多阶段流水线时，就用嵌套的 with Timer() 来逐段定位瓶颈。

with Timer("总流程"):
    with Timer("阶段1: 数据清洗"):
        time.sleep(0.1)
    with Timer("阶段2: 特征计算"):
        time.sleep(0.2)
    with Timer("阶段3: 模型推理"):
        time.sleep(0.15)

# 输出:
# [阶段1: 数据清洗] 耗时: 100.123 ms
# [阶段2: 特征计算] 耗时: 200.456 ms
# [阶段3: 模型推理] 耗时: 150.789 ms
# [总流程] 耗时: 451.368 ms

4.3 多线程/多进程下的计时陷阱

嗯，这里要重点说。很多人在单线程下计时没问题，一上多线程就翻车。我踩过的坑，列出来给你看。

陷阱一：time.sleep 不准确

你以为 time.sleep(0.001) 真的会睡 1 毫秒？太天真了。在 Windows 上，sleep 的最小精度大约是 15 毫秒。你写 1 毫秒，实际可能睡 15 毫秒。

import time

def test_sleep_precision():
    start = time.perf_counter()
    time.sleep(0.001)
    elapsed = time.perf_counter() - start
    print(f"请求 1ms，实际 {elapsed*1000:.3f} ms")
    # 输出: 请求 1ms，实际 15.625 ms (Windows 典型值)

test_sleep_precision()

解决方案：用 time.perf_counter() 做忙等待，或者用 time.monotonic() 配合高精度定时器。但说实话，在 Python 里做微秒级精确计时，本身就是个伪命题。

陷阱二：GIL 导致计时虚高

多线程下，Python 的 GIL 会让线程轮流执行。你测一个线程的耗时，可能包含了等待 GIL 的时间。我曾经在回测系统里发现，同样的策略，单线程跑 1 秒，多线程跑反而要 3 秒——因为线程切换开销远大于并行收益。

import threading
import time

def busy_work():
    for _ in range(10**7):
        pass

threads = []
start = time.perf_counter()
for _ in range(4):
    t = threading.Thread(target=busy_work)
    threads.append(t)
    t.start()
for t in threads:
    t.join()
end = time.perf_counter()
print(f"4线程总耗时: {end-start:.3f}s")
# 输出: 4线程总耗时: 2.1s (单线程约 0.6s，说明没有加速)

我的建议：CPU 密集型任务，用 multiprocessing 而不是 threading。多进程可以绕过 GIL，但要注意进程间通信的开销。我曾经用 multiprocessing.Pool 做并行回测，速度提升了 4 倍，但内存也涨了 4 倍——trade-off 无处不在。

陷阱三：进程间同步开销被忽略

多进程计时，很多人只算计算时间，忘了算数据传递和同步的时间。比如用 multiprocessing.Queue 传数据，一次 put/get 可能就要几毫秒。如果数据量大，这个开销会吃掉所有并行收益。

from multiprocessing import Process, Queue
import time

def worker(q):
    data = q.get()
    # 模拟计算
    time.sleep(0.1)
    q.put(data * 2)

q = Queue()
start = time.perf_counter()
p = Process(target=worker, args=(q,))
p.start()
q.put(42)
result = q.get()
p.join()
end = time.perf_counter()
print(f"进程间通信+计算耗时: {end-start:.3f}s")
# 输出: 进程间通信+计算耗时: 0.152s (纯计算仅 0.1s)

避坑指南：我曾经在做一个实时行情处理系统时，用多进程 + Queue 来分发数据。结果发现 80% 的时间花在序列化和反序列化上。后来改用共享内存（multiprocessing.shared_memory），才把延迟降下来。记住：多进程的通信成本，一定要算进基准测试里。

4.4 总结：选对工具，避开陷阱

场景 推荐方案 注意事项
函数级计时 装饰器 + perf_counter 用 functools.wraps 保留元数据
代码块计时 上下文管理器 __exit__ 返回 False，不吞异常
多线程计时 perf_counter + 统计均值 注意 GIL 导致的虚高，CPU 密集用多进程
多进程计时 perf_counter + 包含通信开销 序列化/反序列化是隐藏成本

说白了，计时这件事，工具本身很简单。难的是你知不知道那些「看不见的时间」花在了哪里。装饰器和上下文管理器帮你把工具用好，多线程陷阱帮你把坑避开。剩下的，就是多测、多想、多优化。

公众号:蓝海资料掘金营，微信deep3321
← 上一章 📖 返回目录 下一章 →