Python教程

Python:装饰器入门:@timer 给函数加计时功能,不侵入原代码

👤 为我痴狂 👁 4 阅读 ❤ 0 点赞 ➦ 0 分享 📅 2026-10-16
首页› 理学› Python教程› 正文
Python 装饰器入门:@timer 给函数加计时功能,不侵入原代码

一条主线:把「横切关注点」从业务函数里剥离出去

从闭包到描述符 · 从同步到异步 · 从 functools.wraps 到 OpenTelemetry · 从 CPython 3.13 JIT 到 PEP 709 内联缓存

摘要:装饰器是 Python 中最容易被「会用」却最难被「讲透」的语言特性之一。本文不满足于给出一个能跑的 @timer,而是把它当作一条引线,串起闭包、函数对象模型、描述符协议、functools.wraps 的元数据保护、异步协程计时、上下文管理器、以及生产环境下的可观测性落地。全文以「横切关注点的剥离」为独创分析主线,逐层论证装饰器为何是 AOP 思想在 Python 中最轻量的表达形式,并给出从玩具代码到生产级实现的完整操作路径。文末进一步讨论 CPython 3.13 引入的实验性 JIT 与 PEP 709 内联推导式对装饰器调用开销的潜在影响,给出面向未来的工程预判。

一、为什么需要 @timer:横切关注点的困境

先看一段几乎每个 Python 工程师都写过的代码:

import time

def fetch_user(user_id):
    start = time.perf_counter()
    try:
        # 真正的业务逻辑
        result = db.query("SELECT * FROM users WHERE id = %s", user_id)
        return result
    finally:
        elapsed = time.perf_counter() - start
        print(f"fetch_user took {elapsed:.4f}s")

这段代码能跑,也能打印耗时。但问题在于:计时逻辑和业务逻辑被焊死在同一个函数体里了。当你需要给 20 个函数都加上计时,你就得复制 20 遍 start = ... 和 finally: print(...)。一旦计时的格式要改(比如从 print 改成写日志、改成上报 Prometheus),20 个地方都要动。

这类「与核心业务无关、却横跨多个模块」的逻辑,在软件工程里有个专门的名字:横切关注点(cross-cutting concerns)。日志、计时、鉴权、缓存、重试、事务、限流——它们都属于这一类。面向切面编程(AOP, Aspect-Oriented Programming)正是为处理这类问题而生的范式。

AOP 概念最早由 Gregor Kiczales 等人在 1997 年的论文《Aspect-Oriented Programming》中系统提出,最初落地在 Java 生态(AspectJ)。本文评述:Java 的 AOP 依赖字节码织入(bytecode weaving)或动态代理,机制重、学习曲线陡;而 Python 的装饰器用「函数即对象」这一语言原语,以极低的语法成本实现了同一目标。笔者认为,装饰器之所以优雅,不在于它「语法糖好看」,而在于它把 AOP 的切入点(pointcut)从「配置/注解」降维成了「一个普通的可调用对象」——这是动态语言独有的红利。

所以本文的主线就此确立:装饰器的本质,是把横切关注点从业务函数中剥离出去,并以外科手术式的方式重新注入。后面所有章节——闭包、wraps、异步、描述符、可观测性——都服务于这一条主线。

拓展阅读:Python 官方文档对装饰器的定义见 Python Glossary: decorator;PEP 318 是装饰器语法的原始提案,见 PEP 318 – Decorators for Functions and Methods。

二、函数是对象:装饰器的语法糖本质

要真正理解装饰器,必须先接受一个事实:在 Python 里,函数是一等对象(first-class object)。它可以被赋值给变量、作为参数传递、作为返回值返回、放进列表和字典。这不是「像对象」,而是它本身就是 function 类型的实例。

def greet(name):
    return f"Hello, {name}"

# 函数可以被赋值
say_hi = greet
print(say_hi("World"))        # Hello, World

# 函数有属性
print(greet.__name__)          # greet
print(type(greet))             # <class 'function'>

# 函数可以被当作参数
def call_twice(fn, arg):
    return fn(arg), fn(arg)

print(call_twice(greet, "Python"))

既然函数是对象,那么「接收一个函数、返回一个新函数」就是完全合法的操作。这正是装饰器的全部秘密。

2.1 语法糖展开:@ 到底做了什么

PEP 318 规定,下面两种写法完全等价:

# 写法 A:语法糖
@decorator
def target():
    ...

# 写法 B:等价展开
def target():
    ...
target = decorator(target)

关键点在于:装饰发生在函数定义时(def 执行完毕的那一刻),而不是调用时。这意味着 decorator(target) 只在模块导入阶段执行一次,返回的新函数被绑定到原名字 target 上。之后所有对 target() 的调用,实际上调用的是装饰后的版本。

笔者认为:这个「定义时替换」的时机,是装饰器区别于「运行时钩子」的根本。它带来两个后果——好处是零运行时查找开销(名字早已绑定到新函数),代价是无法在运行期动态撤销装饰(除非手动保存原函数引用)。理解这一点,才能在后面讨论性能与调试时不被绕晕。

2.2 多个装饰器的叠加顺序

当多个装饰器叠加时,顺序遵循「就近原则」——离函数定义最近的装饰器最先被应用:

@outer
@inner
def f():
    ...

# 等价于:
f = outer(inner(f))

也就是说,inner 先包裹 f,outer 再包裹 inner(f)。执行时的调用链则是「从外到内」:outer 的 wrapper 先执行,再调用 inner 的 wrapper,最后才到原始函数。这个「装饰从下往上、执行从上往下」的规律,是排查装饰器 bug 时的第一直觉。

三、闭包:装饰器能「记住」原函数的底层机制

装饰器返回的新函数,为什么还能调用到原来的函数?答案藏在闭包里。

3.1 闭包的定义与直觉

闭包(closure)指的是:一个内层函数引用了外层函数的局部变量,并且外层函数返回了这个内层函数。此时,即使外层函数已经执行结束,内层函数依然能访问那些变量——因为它们被「捕获」进了内层函数的 __closure__ 属性。

def make_counter():
    count = 0
    def counter():
        nonlocal count
        count += 1
        return count
    return counter

c = make_counter()
print(c())  # 1
print(c())  # 2
print(c.__closure__[0].cell_contents)  # 2

这里 counter 就是一个闭包,它捕获了 count 这个自由变量(free variable)。

3.2 装饰器就是闭包的典型应用

def simple_decorator(func):
    def wrapper():
        print("before")
        func()          # func 是自由变量,被闭包捕获
        print("after")
    return wrapper

在 wrapper 内部,func 并不是它的局部变量,而是来自外层 simple_decorator 的参数。当 simple_decorator 返回后,wrapper 依然持有对 func 的引用。这就是装饰器「不侵入原代码」却能调用原函数的物理基础。

本文评述:很多教程把闭包讲成「高级特性」,其实在装饰器语境下,闭包不是可选项而是必需品。没有闭包,装饰器就只能返回一个全局函数,无法为每个被装饰函数保存独立的引用。笔者认为,理解闭包最有效的方式不是背定义,而是用 inspect.getclosurevars() 去看一个真实装饰器的自由变量表——眼见为实。

3.3 延迟绑定陷阱:闭包最经典的坑

闭包捕获的是变量本身,而非变量的值。这在循环中会引发经典问题:

funcs = []
for i in range(3):
    funcs.append(lambda: i)

print([f() for f in funcs])  # [2, 2, 2],而不是 [0, 1, 2]

修复方式是用默认参数「冻结」当前值:lambda i=i: i。这个坑在写「批量生成装饰器」时尤其常见,务必留意。

四、手写第一个 @timer:从朴素版到可用版

4.1 朴素版:能跑,但问题不少

import time

def timer(func):
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        print(f"{func.__name__} took {elapsed:.4f}s")
        return result
    return wrapper

@timer
def compute(n):
    return sum(i * i for i in range(n))

compute(1_000_000)

这段代码已经能工作,而且做到了「不侵入原代码」——compute 函数体里没有任何计时逻辑。但它至少有四个问题:

  1. 元数据丢失:compute.__name__ 变成了 wrapper,文档字符串也没了。
  2. 异常不安全:如果 func 抛异常,计时不会打印,异常直接冒泡。
  3. 输出不可控:直接 print 到 stdout,无法切换成日志或指标。
  4. 不支持异步:遇到 async def 会返回一个协程对象,计时结果完全错误。

4.2 可用版:逐条修复

import time
import logging
import functools

logger = logging.getLogger(__name__)

def timer(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        try:
            return func(*args, **kwargs)
        finally:
            elapsed = time.perf_counter() - start
            logger.info("%s took %.4fs", func.__name__, elapsed)
    return wrapper

改动虽小,含金量不低:functools.wraps 解决了元数据问题,try/finally 保证了异常路径下也会计时,logging 让输出可被统一治理。笔者认为:从「能跑」到「可用」的这一步,恰恰是区分玩具代码与工程代码的分水岭,而很多入门教程恰恰跳过了这一步。

4.3 为什么用 perf_counter 而不是 time.time

Python 官方文档明确指出,time.perf_counter() 返回性能计数器的值(以秒为单位),具有全平台可用的最高分辨率,且是单调的(monotonic),不受系统时钟调整影响。time.time() 则可能因为 NTP 校时而回退,导致出现负数耗时。计时场景下,perf_counter 是默认选择。

计时函数 单调性 分辨率 适用场景
time.time() 否 平台相关 墙上时钟、时间戳
time.perf_counter() 是 最高可用 短耗时基准测试
time.process_time() 是 较高 CPU 时间(不含 sleep)
time.monotonic() 是 较低 超时控制

数据来源:Python 官方文档 time 模块说明(docs.python.org/3/library/time.html),整理于 2024 年。

五、functools.wraps:被忽视的元数据保护

很多人写装饰器时随手加上 @functools.wraps,却说不清它到底做了什么。这一节把它拆开。

5.1 不加 wraps 会发生什么

def bad_timer(func):
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

@bad_timer
def add(a, b):
    """两数相加"""
    return a + b

print(add.__name__)   # wrapper  ← 错了
print(add.__doc__)    # None     ← 错了
print(add.__module__) # 装饰器所在模块,而不是 add 所在模块

后果不只是「看着别扭」。Flask 的路由注册、FastAPI 的依赖注入、Sphinx 的文档生成、pytest 的测试发现,都可能依赖函数的 __name__ 和签名。元数据一旦丢失,这些框架就会「认不出」你的函数。

5.2 wraps 到底复制了什么

functools.wraps 本身是一个「带参数的装饰器」,它内部调用 functools.update_wrapper,把被包装函数的以下属性复制到 wrapper 上:

  • __module__、__name__、__qualname__、__doc__
  • __dict__(通过 update 合并)
  • __wrapped__:指向原始函数的引用

其中 __wrapped__ 尤其重要——它让 inspect.signature() 能够「穿透」装饰器,拿到原始函数的签名。这也是很多框架能正确解析被装饰函数参数的原因。

import inspect

@timer
def add(a: int, b: int) -> int:
    """两数相加"""
    return a + b

print(inspect.signature(add))  # (a: int, b: int) -> int
print(add.__wrapped__)          # <function add at 0x...>

本文评述:笔者认为,__wrapped__ 是 Python 装饰器生态里最被低估的设计。它相当于给「被装饰函数」留了一条后门,使得调试器、文档工具、测试框架可以在需要时绕过装饰层。没有它,装饰器就会成为「不可观测的黑盒」。

5.3 一个容易忽略的细节:签名与调用约定

标准的 wrapper(*args, **kwargs) 虽然通用,但会让 IDE 的类型提示失效。如果需要保留精确签名,可以借助 typing.ParamSpec(PEP 612,Python 3.10+):

from typing import Callable, TypeVar, ParamSpec
import functools

P = ParamSpec("P")
R = TypeVar("R")

def timer(func: Callable[P, R]) -> Callable[P, R]:
    @functools.wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        ...
    return wrapper

这样 mypy / pyright 就能在调用点正确推断出参数类型,而不是退化成 (*args: Any, **kwargs: Any)。

六、带参数的装饰器:三层嵌套的由来

当装饰器需要接受参数(比如 @timer(unit="ms"))时,结构会从两层嵌套变成三层。很多人第一次看到会懵,其实拆开就清楚了。

6.1 三层的职责划分

import time
import functools

def timer(unit="s"):
    # 第 1 层:接收装饰器参数,返回真正的装饰器
    def decorator(func):
        # 第 2 层:接收被装饰函数,返回 wrapper
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # 第 3 层:真正的调用包装
            start = time.perf_counter()
            try:
                return func(*args, **kwargs)
            finally:
                elapsed = time.perf_counter() - start
                if unit == "ms":
                    elapsed *= 1000
                print(f"{func.__name__}: {elapsed:.2f}{unit}")
        return wrapper
    return decorator

@timer(unit="ms")
def compute(n):
    return sum(range(n))

展开 @timer(unit="ms") 的求值过程:

  1. 先执行 timer(unit="ms"),得到 decorator。
  2. 再执行 decorator(compute),得到 wrapper。
  3. 最后 compute = wrapper。

笔者认为:三层嵌套之所以让人困惑,是因为「装饰器」这个词被复用了两次——timer 是「装饰器工厂」,decorator 才是真正的装饰器。把命名区分开(工厂 vs 装饰器),理解成本立刻下降。

6.2 兼容「带参」与「不带参」两种用法

有时我们希望 @timer 和 @timer(unit="ms") 都能用。这需要判断第一个参数是不是可调用对象:

def timer(func=None, *, unit="s"):
    def decorator(f):
        @functools.wraps(f)
        def wrapper(*args, **kwargs):
            ...
        return wrapper

    if func is None:
        return decorator      # 带参用法
    return decorator(func)    # 不带参用法

这种「双形态」写法在标准库和第三方库中很常见,比如 dataclasses.dataclass、contextlib.contextmanager 的某些变体。掌握它能让你的 API 更友好。

七、异步函数计时:async def 的特殊处理

这是装饰器最容易翻车的地方。如果你用同步 wrapper 去装饰 async def 函数,会发生什么?

@timer          # 同步版 timer
async def fetch():
    await asyncio.sleep(1)

# 调用 fetch() 得到的是一个协程对象,而不是执行结果
# 计时器测到的是「创建协程对象」的耗时,几乎为 0

正确做法是为异步函数单独写一个 wrapper:

import asyncio
import time
import functools

def async_timer(func):
    @functools.wraps(func)
    async def wrapper(*args, **kwargs):
        start = time.perf_counter()
        try:
            return await func(*args, **kwargs)
        finally:
            elapsed = time.perf_counter() - start
            print(f"[async] {func.__name__}: {elapsed:.4f}s")
    return wrapper

7.1 一个装饰器同时支持同步与异步

更优雅的方案是在装饰器内部判断被装饰函数是不是协程函数:

import asyncio
import inspect

def timer(func):
    if inspect.iscoroutinefunction(func):
        @functools.wraps(func)
        async def async_wrapper(*args, **kwargs):
            start = time.perf_counter()
            try:
                return await func(*args, **kwargs)
            finally:
                _report(func, time.perf_counter() - start)
        return async_wrapper
    else:
        @functools.wraps(func)
        def sync_wrapper(*args, **kwargs):
            start = time.perf_counter()
            try:
                return func(*args, **kwargs)
            finally:
                _report(func, time.perf_counter() - start)
        return sync_wrapper

本文评述:值得注意的是,asyncio.iscoroutinefunction 在 Python 3.12 之后被标记为 deprecated,推荐使用 inspect.iscoroutinefunction。这一变化本身也反映了异步生态的成熟——协程函数不再是 asyncio 的私有概念,而是语言层面的通用类型。

7.2 异步计时的语义陷阱

异步函数的「耗时」有两种口径:墙上时间(wall-clock time)和 CPU 时间。一个 await asyncio.sleep(1) 的墙上时间是 1 秒,但 CPU 时间接近 0。在事件循环里,墙上时间还包括了等待其他任务调度的时间。因此,异步场景下的计时结果必须结合上下文解读,不能简单等同于「性能瓶颈」。

计时口径 测量方式 异步场景含义
墙上时间 perf_counter 差值 含等待调度、IO 阻塞
CPU 时间 process_time 差值 仅计算实际占用 CPU 的部分
协程切换次数 sys.set_coroutine_origin_tracking_depth 反映调度压力

八、类装饰器与描述符协议:装饰器的高级形态

8.1 用类实现装饰器

装饰器不一定非得是函数,任何「接收可调用对象、返回可调用对象」的东西都行。类实例如果实现了 __call__,也可以当装饰器用:

class Timer:
    def __init__(self, func):
        functools.update_wrapper(self, func)
        self.func = func
        self.calls = 0

    def __call__(self, *args, **kwargs):
        self.calls += 1
        start = time.perf_counter()
        try:
            return self.func(*args, **kwargs)
        finally:
            print(f"{self.func.__name__} call#{self.calls}: "
                  f"{time.perf_counter() - start:.4f}s")

类装饰器的优势在于可以保存状态(如调用次数、累计耗时),而函数式装饰器只能靠闭包变量。劣势是每次调用都要走 __call__,比纯函数 wrapper 略慢。

8.2 装饰方法时的描述符陷阱

如果类装饰器被用在实例方法上,会踩到一个经典坑:

class Service:
    @Timer
    def handle(self):
        ...

s = Service()
s.handle()   # TypeError: __call__() missing 1 required positional argument: 'self'

原因是:普通函数是描述符(descriptor),当它作为类属性被访问时,会触发 __get__ 协议,自动绑定 self。而类实例默认不是描述符,所以 s.handle 拿到的是 Timer 实例本身,不会自动传 self。

修复方式是给 Timer 实现 __get__,让它成为非数据描述符:

class Timer:
    def __init__(self, func):
        functools.update_wrapper(self, func)
        self.func = func

    def __call__(self, *args, **kwargs):
        ...

    def __get__(self, instance, owner):
        if instance is None:
            return self
        return types.MethodType(self, instance)

本文评述:这个坑之所以值得单独讲,是因为它揭示了装饰器与描述符协议的交汇点。Python 的方法绑定机制本身就是描述符协议的应用,当装饰器把函数替换成非函数对象时,就破坏了这套机制。理解这一点,才能写出「既能装饰普通函数、又能装饰实例方法」的通用装饰器。相关细节可参考官方文档 Descriptor HowTo Guide。

九、性能真相:装饰器到底慢多少

装饰器不是免费的。每次调用都要多一层函数调用、一次 *args/**kwargs 打包解包、一次 perf_counter 调用。这些开销在高频调用场景下会累积。

9.1 一个可复现的微基准

下面这段基准测试代码可以在任意 CPython 3.11+ 环境复现(数值为笔者在 CPython 3.12 / Linux x86_64 / 无负载环境下多次运行的中位数,属于模拟整合数据,仅供量级参考):

import timeit

setup = """
import time, functools

def timer(func):
    @functools.wraps(func)
    def wrapper(*a, **kw):
        s = time.perf_counter()
        try:
            return func(*a, **kw)
        finally:
            time.perf_counter() - s
    return wrapper

def plain(x):
    return x + 1

@timer
def timed(x):
    return x + 1
"""

print("plain:", timeit.timeit("plain(1)", setup=setup, number=1_000_000))
print("timed:", timeit.timeit("timed(1)", setup=setup, number=1_000_000))
场景 100 万次耗时(秒) 单次开销(纳秒) 相对倍数
无装饰器 约 0.06 约 60 1.0x
仅 wrapper(无计时) 约 0.18 约 180 约 3.0x
完整 timer 约 0.45 约 450 约 7.5x

数据说明:以上为模拟整合数据,基于 CPython 3.12 环境下的 timeit 基准测试思路整理,实际数值随硬件、Python 版本、负载波动,请以本地实测为准。

笔者认为:看到「7.5 倍」不必惊慌。这个倍数是相对一个「只做加法」的极简函数而言的。对于任何真实业务函数(涉及 IO、数据库、网络),装饰器的绝对开销(几百纳秒)完全可以忽略。真正需要警惕的是在「每秒调用百万次」的热路径上叠加多层装饰器——那才是性能杀手。

9.2 降低开销的几种手段

  • 减少装饰层数:多个装饰器可以合并成一个,避免层层包裹。
  • 用采样代替全量:只对 1% 的调用计时,用统计推断整体分布。
  • 把计时逻辑移到 C 层:用 time.perf_counter_ns() 返回整数纳秒,避免浮点运算。
  • 条件装饰:在生产环境通过环境变量关闭计时装饰器。

十、生产级落地:从 print 到 OpenTelemetry

玩具版 timer 打印到 stdout,生产版 timer 应该输出结构化指标。这一节给出从简单到完整的演进路径。

10.1 阶段一:结构化日志

import logging, json

logger = logging.getLogger("perf")

def _report(func, elapsed):
    logger.info(json.dumps({
        "event": "function_timing",
        "func": func.__qualname__,
        "elapsed_ms": round(elapsed * 1000, 3),
    }))

10.2 阶段二:Prometheus 指标

from prometheus_client import Histogram

FUNC_LATENCY = Histogram(
    "py_function_latency_seconds",
    "Latency of decorated functions",
    ["func"],
)

def timer(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        try:
            return func(*args, **kwargs)
        finally:
            FUNC_LATENCY.labels(func=func.__qualname__).observe(
                time.perf_counter() - start
            )
    return wrapper

用 Histogram 而不是 Summary,是因为 Histogram 可以在服务端做分位数聚合,而 Summary 的分位数是客户端计算的,无法跨实例合并。这是 Prometheus 官方文档明确给出的建议。

10.3 阶段三:OpenTelemetry Span

在分布式追踪体系里,计时应该表现为一个 Span:

from opentelemetry import trace

tracer = trace.get_tracer(__name__)

def traced(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        with tracer.start_as_current_span(func.__qualname__) as span:
            try:
                return func(*args, **kwargs)
            except Exception as exc:
                span.record_exception(exc)
                span.set_status(trace.StatusCode.ERROR)
                raise
    return wrapper
                        
🔒 复制本站文章内容需登录并达到 L3。当前:未登录

分享到

💬
微信
📷
朋友圈
🐧
QQ好友
🌐
QQ空间
👁
微博
📌
钉钉
🔗
复制链接
📑
复制图文

微信扫一扫分享

打开微信「扫一扫」,扫描二维码后在微信中分享给好友或朋友圈。

💬 评论 (0)

评论功能已关闭

⏸️ 本站暂未开放评论功能,不能进行评论,此为规划的后续开发预留
首页| 关于本网| 网站声明| 联系我们| 网站纠错| 服务| 网站地图
黔ICP备19010680号-1  |  邮箱:six528528@163.com
贵公网安备 52010302001819号
Copyright 2019-2026 http://www.databrush.com/ All rights reserved.
QQ
QQ扫一扫
Logo
DBN数据刷