装饰器还能这么玩？手把手教你写出能“重试”的异步函数

1. 什么是装饰器

要理解装饰器，我们得先看透它的本质。在Python世界里，装饰器就是个“高阶函数”——它可以把一个函数作为“原料”吃进去，加工后再吐出一个功能增强的新函数。整个过程，原始函数代码纹丝不动。

免费影视、动漫、音乐、游戏、小说资源长期稳定更新！ 👉 点此立即查看 👈

简单来说，它就是个能装进“通用功能”的工具箱。这个工具箱能发挥作用，全赖Python语言的三项核心能力：

• 函数是一等公民：函数和其他数据类型（比如字符串、列表）平起平坐，可以被当成参数传递，也可以作为结果返回。这是装饰器得以实现的理论基石。
• 闭包特性：装饰器通常依靠闭包来“记住”外部作用域的变量和参数。这就像给工具箱上了锁，把原始函数的状态牢牢锁在里面，无论搬到哪里用都不会丢失。
• 语法糖支持：Python用那个简洁的@符号，把原本有点绕的调用过程变得无比优雅。没有它，装饰器虽能用，但代码的观感会大打折扣。

2. 装饰器的核心作用

装饰器这么受欢迎，到底能带来什么实在的好处？归根结底，是下面四条：

• 代码复用：谁也不想在不同的函数里重复写着同样的日志、计时代码吧？把这些“通用套路”抽象成一个装饰器，一次写好，随处复用。
• 逻辑解耦：想象一下，业务逻辑是主菜，而日志、权限、监控就像是调料。装饰器帮我们把“炒菜”和“调味”分离开，代码结构瞬间清晰，各司其职。
• 动态扩展：这恰好符合软件设计的“开闭原则”——对扩展开放，对修改关闭。给函数增加新能力，不必动它的源代码，运行时动态“戴上”装饰器就行。
• 非侵入性：这是最大的魅力所在。无论需要什么新功能，都无需入侵原有函数内部。原来的接口什么样，现在还是什么样，对调用者完全透明。

3. 装饰器的常见应用场景

明白了道理，就得看实战。下面这几个场景，几乎每个Python开发者都会遇到。我们可以把它想象成给游戏角色加“Buff”，让代码能力瞬间提升。

(1) 日志记录

自动记录函数何时被调用、何时结束，就像游戏里的“成就系统”，忠实地记录下每一次关键操作。

import functools
import datetime

def logger(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        print(f"[{datetime.datetime.now()}] 调用函数: {func.__name__}")
        result = func(*args, **kwargs)
        print(f"[{datetime.datetime.now()}] 函数 {func.__name__} 执行完成")
        return result
    return wrapper

@logger
def process_data():
    print(“正在处理数据…”)

(2) 性能计时

测量函数执行时间，好比“速度跑酷”的计时器，精准告诉你每个环节花了多少秒。性能优化的第一步，往往就从这里开始。

import time
import functools

def timer(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        end = time.time()
        print(f”函数 {func.__name__} 执行耗时: {end-start:.4f}秒")
        return result
    return wrapper

(3) 权限验证

控制哪些用户能访问特定函数，这功能就像游戏里的“区域权限”。普通玩家进不了管理员地图，在代码层面也是同样的逻辑。

def permission_check(required_role):
    def decorator(func):
        def wrapper(*args, **kwargs):
            user_role = getattr(args[0], ‘role’, ‘guest’) if args else ‘guest’
            if user_role != required_role:
                raise PermissionError(f"需要 {required_role} 权限")
            return func(*args, **kwargs)
        return wrapper
    return decorator

class Admin:
    def __init__(self):
        self.role = ‘admin’

    @permission_check(‘admin’)
    def delete_user(self):
        print(“删除用户操作成功”)

(4) 缓存机制

对于计算成本高昂或频繁读取相同数据的函数，加个缓存装饰器，能立竿见影地提升性能，避免无谓的重复劳动。

def cache(func):
    stored = {}
    @functools.wraps(func)
    def wrapper(*args):
        if args in stored:
            print(“从缓存获取结果”)
            return stored[args]
        result = func(*args)
        stored[args] = result
        return result
    return wrapper

@cache
def fibonacci(n):
    if n < 2:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

4. 装饰器的异步实现方法

进入异步编程的世界，装饰器的写法也需要“升级换代”。核心区别在于，异步装饰器的包装函数必须用async def来定义，并且在调用被装饰的异步函数时，必须加上await。这一步要是忘了，程序要么报错，要么就失去了异步的能力。

(1) 异步计时器

给异步函数计时的逻辑和同步版类似，但获取时间和等待都得用异步的方式。

import asyncio
import functools
import datetime

def async_timer(func):
    @functools.wraps(func)
    async def wrapper(*args, **kwargs):
        start = asyncio.get_event_loop().time()
        print(f"[{datetime.datetime.now()}] 异步函数 {func.__name__} 开始执行")
        result = await func(*args, **kwargs)
        end = asyncio.get_event_loop().time()
        print(f"[{datetime.datetime.now()}] 异步函数 {func.__name__} 执行完成，耗时: {end-start:.4f}秒")
        return result
    return wrapper

@async_timer
async def fetch_data():
    print(“正在获取数据…”)
    await asyncio.sleep(2)  # 模拟异步IO操作
    print(“数据获取完成”)

# 运行异步函数
asyncio.run(fetch_data())

(2) 异步重试机制

网络请求不稳定是常态。一个健壮的异步重试装饰器，能让你的程序在面对临时故障时更加从容。

import asyncio
import functools
import random

def async_retry(max_attempts=3, delay=1):
    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return await func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_attempts - 1:
                        raise e
                    print(f”第{attempt+1}次尝试失败: {e}，{delay}秒后重试…”)
                    await asyncio.sleep(delay)
        return wrapper
    return decorator

@async_retry(max_attempts=3, delay=1)
async def unreliable_api():
    # 模拟不稳定的API调用
    if random.random() < 0.7:  # 70%概率失败
        raise ConnectionError(“网络连接失败”)
    print(“API调用成功”)

# 运行异步函数
asyncio.run(unreliable_api())

(3) 异步缓存

异步函数的缓存逻辑同样关键。需要注意的是，缓存字典的访问在异步环境下也需要注意线程安全（本例为简单演示，在复杂并发场景下可能需要引入锁机制）。

import asyncio
import functools

def async_cache(func):
    stored = {}
    @functools.wraps(func)
    async def wrapper(*args):
        if args in stored:
            print(“从缓存获取异步结果”)
            return stored[args]
        result = await func(*args)
        stored[args] = result
        return result
    return wrapper

@async_cache
async def get_user_info(user_id):
    print(f”正在获取用户{user_id}信息…”)
    await asyncio.sleep(1)  # 模拟异步IO
    return {“id”: user_id, “name”: f"用户{user_id}"}

# 运行异步函数
async def main():
    # 第一次调用，会执行函数
    user1 = await get_user_info(1)
    print(user1)

    # 第二次调用，会从缓存获取
    user2 = await get_user_info(1)
    print(user2)

asyncio.run(main())

5. 高级用法

掌握了基础，再来看看那些能让装饰器更强大、更灵活的“进阶技巧”。

(1) 带参数的装饰器

有时候，我们想动态配置装饰器的行为，比如重试次数、延迟时间。这就需要让装饰器本身也能接收参数，其结构会多嵌套一层。

def retry(max_attempts=3, delay=1):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_attempts - 1:
                        raise e
                    print(f”第{attempt+1}次尝试失败，{delay}秒后重试…”)
                    time.sleep(delay)
        return wrapper
    return decorator

(2) 类装饰器

除了用函数实现，装饰器还可以是一个类。通过实现__call__方法，类的实例就能像函数一样被调用，同时还能利用类的属性来保存更复杂的状态。

class CountCalls:
    def __init__(self, func):
        self.func = func
        self.count = 0

    def __call__(self, *args, **kwargs):
        self.count += 1
        print(f”函数 {self.func.__name__} 被调用 {self.count} 次")
        return self.func(*args, **kwargs)

6. 小结

回头看文章开头的那个电商平台深夜故障。如果那位工程师提前为关键的process_order函数装配了具备详细链路追踪和性能计时的异步装饰器，那么问题排查很可能就是几分钟的事：日志会清晰指出卡在支付、库存还是物流环节，耗时数据会直接定位到性能瓶颈。

这就是装饰器的力量。它通过一种非侵入式的优雅方式，将那些横跨多个模块的共性需求——日志、监控、安全、缓存——封装起来。从同步到异步，从基础功能到动态参数，装饰器提供了一套强大的元编程工具。掌握它，意味着你掌握了为代码批量、动态赋能的能力。下次当你发现自己在不同函数中复制粘贴同一段代码时，就该考虑是不是该把它抽象成一个装饰器了。