游乐游手机版
首页/业界动态/文章详情

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

时间:2026-04-14 21:30
1 什么是装饰器 要理解装饰器,我们得先看透它的本质。在Python世界里,装饰器就是个“高阶函数”——它可以把一个函数作为“原料”吃进去,加工后再吐出一个功能增强的新函数。整个过程,原始函数代码纹丝不动。 简单来说,它就是个能装进“通用功能”的工具箱。这个工具箱能发挥作用,全赖Python语言的

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函数装配了具备详细链路追踪和性能计时的异步装饰器,那么问题排查很可能就是几分钟的事:日志会清晰指出卡在支付、库存还是物流环节,耗时数据会直接定位到性能瓶颈。

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

来源:https://www.51cto.com/article/839866.html
上一篇如何在两个镜像仓库之间迁移 Docker 跨平台镜像 下一篇CISA 警告 F5 BIG-IP 漏洞正遭攻击者积极利用
本站内容用于信息整理与展示,如有侵权或内容问题请及时联系处理。

相关推荐

补充同频道和同主题内容,方便继续浏览更多相关内容。

同类最新

继续查看同栏目最近更新的文章。

更多
九号N1机甲风电动车发布 模拟声浪轻量化车架3499元起
业界动态 · 2026-05-29

九号N1机甲风电动车发布 模拟声浪轻量化车架3499元起

九号发布N1机甲风电动车系列,三款起售价3499元。N170极速47km h,轻量化车架;N185极速55km h,可选模拟声浪;旗舰N190极速60km h,标配模拟声浪及双通道ABS,7月上市。

九号2026新品发布会最强阵容连发4款新车重新定义好车标准
业界动态 · 2026-05-29

九号2026新品发布会最强阵容连发4款新车重新定义好车标准

九号公司发布2026年新品,推出N1、M1、M3及Fz5四款新车,覆盖电摩与电自领域。N1主打短轴距声光电酷玩体验,M1配备双通道ABS与100公里真续航,M3下放AXC车架技术,Fz5首搭载双向转把功能。同时推出3年原厂换新质保等用户权益。

世界超级摩托车锦标赛阿拉贡站张雪机车超级杆位赛获亚军
业界动态 · 2026-05-29

世界超级摩托车锦标赛阿拉贡站张雪机车超级杆位赛获亚军

5月29日,世界超级摩托车锦标赛(WSBK)阿拉贡站传来一则引人瞩目的消息——中国摩托车制造商“张雪机车”旗下的法国车手瓦伦丁·德比斯,在WorldSSP组别的超级杆位赛中成功夺得第二名。 先简要科普一下赛事背景:世界超级摩托车锦标赛(WSBK)是由国际摩托车联合会于1988年创立的顶级公路摩托车赛

英雄联盟海克斯大乱斗重大更新 移除羁绊新增技能符文
业界动态 · 2026-05-29

英雄联盟海克斯大乱斗重大更新 移除羁绊新增技能符文

英雄联盟海克斯大乱斗将在26 12版本移除羁绊系统,上线技能符文体系。该符文能重构技能释放逻辑,实现布里茨钩五人、拉克丝定全队等效果。部分原有羁绊效果转为独立专属符文,更新预计2026年6月中旬登陆国服。

领克10/10+正式上市限时价16.99-23.59万号称弯道之王
业界动态 · 2026-05-29

领克10/10+正式上市限时价16.99-23.59万号称弯道之王

```html 5月29日晚间,领克终于将其备受关注的中大型运动纯电轿车正式推向市场——领克10与领克10+同步上市,官方直接打出“弯道之王”的旗号。我们先不深究它是否真能“弯道超车”,单从价格来看,就已经颇具冲击力。 先奉上一张价格速览表,让大家心里有个底: 领克 10 701 长续航 Max:指