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

1. 什么是装饰器

装饰器的本质是Python中的高阶函数。它接收一个函数作为输入，并返回一个功能增强的新函数，整个过程无需修改原函数的内部代码。

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

装饰器是实现通用功能复用的核心工具，其可行性建立在Python语言的三个关键特性之上：

• 函数是一等公民：函数与基础数据类型地位相同，可作为参数传递或作为返回值。这是装饰器模式得以存在的语言基础。
• 闭包特性：装饰器利用闭包捕获并维持外部作用域的变量与函数引用，确保被装饰函数的状态在调用过程中持久且一致。
• 语法糖支持：Python的@decorator语法提供了直观的声明式使用方式，极大提升了代码的可读性与简洁性。

2. 装饰器的核心作用

装饰器在Python开发中广泛应用，主要解决以下四个核心问题：

• 代码复用：将日志记录、性能监控、权限校验等横切关注点抽象为独立装饰器，消除代码重复。
• 逻辑解耦：分离核心业务逻辑与辅助性功能，使代码结构更清晰，符合单一职责原则。
• 动态扩展：在不修改源代码的前提下，为函数或方法动态添加新行为，完美践行开闭原则。
• 非侵入性：对原有函数接口与调用方式完全透明，无需调用者感知装饰层的存在。

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

装饰器在工程实践中主要应用于以下场景，如同为函数装备可插拔的功能模块。

(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) 缓存机制

对计算密集型或IO密集型函数的结果进行缓存，避免重复计算，显著提升程序响应速度。

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/await语法。关键点在于包装函数需定义为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函数已装配具备全链路追踪与性能分析的异步装饰器，问题定位将变得极为高效：日志能立即指向支付、库存或物流中的阻塞环节，耗时数据可直接揭示性能瓶颈。

这正是装饰器的价值所在。它通过非侵入式的元编程手段，将日志、监控、安全、缓存等横切关注点封装为可复用的组件。从同步到异步，从基础装饰到参数化配置，装饰器体系为Python开发者提供了一套强大的功能增强工具箱。熟练运用装饰器，意味着你掌握了为代码库批量注入能力、同时保持架构整洁的关键技能。当你发现自己在多处复制相似代码时，便是考虑将其抽象为装饰器的最佳时机。