Python案例如何实现装饰器模式?完整代码与实战解析
目录导读
- 什么是装饰器模式?核心概念与Python特色
- Python装饰器模式 vs 经典GoF装饰器模式差异
- 实战案例一:函数日志装饰器(入门级)
- 实战案例二:带参数的装饰器(进阶)
- 实战案例三:类装饰器实现性能监控(高阶)
- 常见问题FAQ
什么是装饰器模式?核心概念与Python特色
装饰器模式(Decorator Pattern)是一种结构型设计模式,允许在不修改原始对象代码的情况下,动态地为对象添加新功能,在Python中,装饰器模式被语言自身内化为语法糖,成为日常开发中高频使用的工具。

核心三要素:
- 被装饰对象(函数/类)
- 装饰器函数(实现附加逻辑)
- 返回的新函数(保持接口兼容)
Python的特殊优势在于:函数是一等公民,可以像普通变量一样传递和返回,这使得装饰器模式的实现比其他语言(如Java)简洁得多。
Python装饰器模式 vs 经典GoF装饰器模式差异
| 对比项 | 经典GoF装饰器模式(Java/C++) | Python装饰器模式 |
|---|---|---|
| 实现方式 | 需要定义抽象的Component类和ConcreteDecorator类 | 直接使用嵌套函数或类实现 |
| 调用方式 | new ConcreteDecorator(new Component()) |
@decorator语法 |
| 灵活性 | 需要预先规划接口 | 可装饰任意可调用对象 |
| 代码量 | 通常需要5-10个类 | 3-5行代码即可完成 |
关键区别:Python装饰器本质是一个高阶函数(接受函数作为参数并返回新函数),而GoF模式更强调通过对象组合实现扩展。
实战案例一:函数日志装饰器(入门级)
需求:为电商系统中的订单处理函数添加执行时间日志,不修改原有业务逻辑。
import time
import logging
def log_execution_time(func):
"""装饰器:记录函数执行时间"""
def wrapper(*args, **kwargs):
logger = logging.getLogger(func.__name__)
start_time = time.time()
try:
result = func(*args, **kwargs)
elapsed = time.time() - start_time
logger.info(f"{func.__name__} executed in {elapsed:.4f}s")
return result
except Exception as e:
elapsed = time.time() - start_time
logger.error(f"{func.__name__} failed after {elapsed:.4f}s: {e}")
raise
return wrapper
# 使用装饰器
@log_execution_time
def process_order(order_id):
# 模拟复杂业务逻辑
time.sleep(order_id % 10 * 0.1)
return f"Order {order_id} processed"
# 调用
print(process_order(1001)) # 自动输出日志
运行效果:
Order 1001 processed
INFO:process_order:process_order executed in 0.1012s
核心机制:
log_execution_time接收原函数func- 内部定义
wrapper函数,通过*args, **kwargs传递所有参数 - 在
wrapper中新增计时逻辑,返回原函数结果 - 最终语法将
process_order替换为wrapper
实战案例二:带参数的装饰器(进阶)
场景:需要为不同接口设置不同的缓存过期时间(TTL)。
import time
from functools import wraps
def cache_with_ttl(ttl_seconds: int = 300):
"""带参数的装饰器:为函数添加缓存功能"""
def decorator(func):
# 存储缓存数据的字典
cache = {}
@wraps(func) # 保持原函数的元信息
def wrapper(*args, **kwargs):
# 生成缓存键(使用参数组合)
key = (args, frozenset(kwargs.items()))
now = time.time()
# 检查缓存是否有效
if key in cache:
cached_time, cached_result = cache[key]
if now - cached_time < ttl_seconds:
print(f"[Cache HIT] {func.__name__}")
return cached_result
# 执行原函数
result = func(*args, **kwargs)
cache[key] = (now, result)
print(f"[Cache MISS] {func.__name__}")
return result
return wrapper
return decorator
# 使用:设置180秒过期时间
@cache_with_ttl(ttl_seconds=180)
def get_user_profile(user_id):
print(f"Fetching from database for user {user_id}")
return {"id": user_id, "name": "Alice", "visited": time.time()}
# 测试
print(get_user_profile(123)) # 第一次查询,缓存未命中
print(get_user_profile(123)) # 第二次查询,缓存命中
输出:
[Cache MISS] get_user_profile
Fetching from database for user 123
{'id': 123, 'name': 'Alice', 'visited': 1698912345.67}
[Cache HIT] get_user_profile
{'id': 123, 'name': 'Alice', 'visited': 1698912345.67}
关键设计:
- 使用三层嵌套:外层
cache_with_ttl接收参数,返回decorator @wraps(func)确保保留原函数__name__、__doc__等属性- 缓存采用字典+时间戳判断,避免内存泄漏可通过
functools.lru_cache优化
实战案例三:类装饰器实现性能监控(高阶)
需求:统计系统中所有API接口的调用次数和平均响应时间。
import time
import functools
from collections import defaultdict
class APIMonitor:
"""类装饰器:集中管理API性能指标"""
metrics = defaultdict(lambda: {"count": 0, "total_time": 0.0})
def __init__(self, api_name=None):
self.api_name = api_name
def __call__(self, func):
# 使用functools.wraps保持函数签名
@functools.wraps(func)
def wrapped_func(*args, **kwargs):
start = time.perf_counter()
result = func(*args, **kwargs)
elapsed = time.perf_counter() - start
# 更新指标
name = self.api_name or func.__name__
APIMonitor.metrics[name]["count"] += 1
APIMonitor.metrics[name]["total_time"] += elapsed
return result
return wrapped_func
@classmethod
def report(cls):
"""生成性能报告"""
print("\n========== API Performance Report ==========")
for api_name, data in cls.metrics.items():
avg_time = data["total_time"] / data["count"] if data["count"] else 0
print(f"{api_name}:")
print(f" - Calls: {data['count']}")
print(f" - Total time: {data['total_time']:.4f}s")
print(f" - Average time: {avg_time:.4f}s")
print("==============================================\n")
# 使用示例
@APIMonitor(api_name="create_order")
def create_user_order(user_id, item):
time.sleep(0.2)
return f"User {user_id} ordered {item}"
@APIMonitor(api_name="checkout")
def order_checkout(order_id):
time.sleep(0.5)
return f"Order {order_id} completed"
# 模拟调用
for _ in range(3):
create_user_order(101, "laptop")
for _ in range(2):
order_checkout(2001)
# 生成报告
APIMonitor.report()
输出:
========== API Performance Report ==========
create_order:
- Calls: 3
- Total time: 0.6024s
- Average time: 0.2008s
checkout:
- Calls: 2
- Total time: 1.0012s
- Average time: 0.5006s
==============================================
设计亮点:
- 类装饰器通过
__call__实现,支持实例化传参 - 使用类变量
metrics跨函数共享统计 time.perf_counter()提供高精度计时- 装饰器不仅可用于函数,也可用于类方法
常见问题FAQ
Q1: 多个装饰器的执行顺序是什么?
A: 从下往上装饰,从上往下执行。
@decorator_A
@decorator_B
def func():
pass
等价于func = decorator_A(decorator_B(func)),调用时,先执行装饰器A的wrapper逻辑,再进入B的wrapper,最后到原函数。
Q2: 装饰器是否可以装饰类?
A: 可以,有两种方式:
- 装饰器返回新类(类工厂模式)
- 使用类装饰器(如上面的
APIMonitor),通过@类名(参数)修饰类方法
Q3: 如何保留被装饰函数的元信息?
A: 使用functools.wraps,在wrapper定义前添加@wraps(func),它会复制__module__、__name__、__qualname__、__doc__、__dict__和__wrapped__属性。
Q4: 装饰器模式与代理模式有什么区别?
A: 代理模式通常控制对对象的访问(如延迟加载、权限验证),而装饰器模式侧重于动态添加功能,Python中两者实现技术类似,但语义不同。
Q5: 在实际项目中,何时使用装饰器?
A: 典型场景包括:日志记录、性能监控、权限校验、事务管理、缓存、重试机制、输入验证等,当需要横切多个函数添加相同功能时,装饰器是最优雅的解决方案。
延伸学习:Python官方文档的functools模块提供了wraps、lru_cache、singledispatch等实用装饰器,以及contextlib中的contextmanager装饰器,值得深入研究。