当前位置：首页>python>Python设计模式:装饰器与观察者

Python设计模式:装饰器与观察者

2026-07-01 21:05:18

Python设计模式:装饰器与观察者

副标题

: 90%的人不知道，装饰器模式能让你在不修改代码的情况下增强功能

痛点：为什么你的功能总是重复添加？

2025年某电商系统，每个API都要手动添加日志、权限验证、缓存。问题出在哪？工程师没有使用装饰器模式统一处理。

真相

：装饰器模式能在不修改原代码的情况下动态添加功能。

模式	核心目的	典型场景
装饰器	动态添加功能	日志、缓存、权限
观察者	一对多通知	事件系统、UI更新

一、装饰器模式

1.1 什么是装饰器模式

装饰器模式动态地将责任附加到对象上，比继承更灵活。

# 组件接口
class Coffee:
def cost(self):
pass
def description(self):
pass
具体组件
class SimpleCoffee(Coffee):
def cost(self):
return 5
def description(self):
return "普通咖啡"
装饰器基类
class CoffeeDecorator(Coffee):
def __init__(self, coffee):
self._coffee = coffee
def cost(self):
return self._coffee.cost()
def description(self):
return self._coffee.description()
具体装饰器
class MilkDecorator(CoffeeDecorator):
def cost(self):
return self._coffee.cost() + 2
def description(self):
return self._coffee.description() + ", 加奶"
class SugarDecorator(CoffeeDecorator):
def cost(self):
return self._coffee.cost() + 1
def description(self):
return self._coffee.description() + ", 加糖"
使用
coffee = SimpleCoffee()
print(f"{coffee.description()} - ${coffee.cost()}")
coffee = MilkDecorator(coffee)
print(f"{coffee.description()} - ${coffee.cost()}")
coffee = SugarDecorator(coffee)
print(f"{coffee.description()} - ${coffee.cost()}")
普通咖啡 - $5
普通咖啡, 加奶 - $7
普通咖啡, 加奶, 加糖 - $8

1.2 Python内置装饰器

# @staticmethod - 静态方法
class MyClass:
@staticmethod
def static_method():
return "静态方法"
@classmethod - 类方法
class MyClass:
@classmethod
def class_method(cls):
return f"类方法: {cls.__name__}"
@property - 属性访问器
class MyClass:
def __init__(self):
self._value = 0
@property
def value(self):
return self._value
@value.setter
def value(self, val):
if val < 0:
raise ValueError("不能为负")
self._value = val

1.3 自定义装饰器

def log_calls(func):
"""记录函数调用"""
def wrapper(args, *kwargs):
print(f"调用: {func.__name__}")
result = func(args, *kwargs)
print(f"返回: {result}")
return result
return wrapper
@log_calls
def add(a, b):
return a + b
add(3, 5)
调用: add
返回: 8

1.4 带参数的装饰器

def repeat(times):
"""重复执行"""
def decorator(func):
def wrapper(args, *kwargs):
for _ in range(times):
result = func(args, *kwargs)
return result
return wrapper
return decorator
@repeat(3)
def greet(name):
print(f"Hello, {name}")
greet("Alice")
Hello, Alice (3次)

1.5 类装饰器

def singleton(cls):
"""单例装饰器"""
instances = {}
def get_instance(args, *kwargs):
if cls not in instances:
instances[cls] = cls(args, *kwargs)
return instances[cls]
return get_instance
@singleton
class Database:
def __init__(self):
print("初始化数据库")
db1 = Database()
db2 = Database()
print(db1 is db2)  # True

二、观察者模式

2.1 什么是观察者模式

定义对象间的一对多依赖，当一个对象状态改变时，所有依赖者都会收到通知。

# 主题（被观察者）
class Subject:
def __init__(self):
self._observers = []
def attach(self, observer):
self._observers.append(observer)
def detach(self, observer):
self._observers.remove(observer)
def notify(self, message):
for observer in self._observers:
observer.update(message)
观察者
class Observer:
def update(self, message):
pass
class ConcreteObserver(Observer):
def __init__(self, name):
self.name = name
def update(self, message):
print(f"{self.name} 收到通知: {message}")
使用
subject = Subject()
observer1 = ConcreteObserver("观察者1")
observer2 = ConcreteObserver("观察者2")
subject.attach(observer1)
subject.attach(observer2)
subject.notify("状态改变!")
观察者1 收到通知: 状态改变!
观察者2 收到通知: 状态改变!

2.2 发布-订阅模式

class EventBus:
"""事件总线"""
def __init__(self):
self._subscribers = {}
def subscribe(self, event_type, callback):
if event_type not in self._subscribers:
self._subscribers[event_type] = []
self._subscribers[event_type].append(callback)
def unsubscribe(self, event_type, callback):
if event_type in self._subscribers:
self._subscribers[event_type].remove(callback)
def publish(self, event_type, data=None):
if event_type in self._subscribers:
for callback in self._subscribers[event_type]:
callback(data)
使用
bus = EventBus()
def on_user_created(data):
print(f"用户创建: {data}")
def on_user_deleted(data):
print(f"用户删除: {data}")
bus.subscribe('user.created', on_user_created)
bus.subscribe('user.deleted', on_user_deleted)
bus.publish('user.created', {'id': 1, 'name': 'Alice'})
bus.publish('user.deleted', {'id': 1})
用户创建: {'id': 1, 'name': 'Alice'}
用户删除: {'id': 1}

2.3 信号-槽机制

from functools import wraps
class Signal:
"""信号"""
def __init__(self):
self._receivers = []
def connect(self, receiver):
self._receivers.append(receiver)
return receiver  # 返回装饰器
def disconnect(self, receiver):
self._receivers.remove(receiver)
def send(self, sender, **kwargs):
for receiver in self._receivers:
receiver(sender, **kwargs)
使用
user_created = Signal()
user_deleted = Signal()
@user_created.connect
def log_user_created(sender, **kwargs):
print(f"日志: 用户 {kwargs.get('name')} 创建")
@user_deleted.connect
def log_user_deleted(sender, **kwargs):
print(f"日志: 用户 {kwargs.get('name')} 删除")
触发信号
user_created.send(None, name='Alice')
user_deleted.send(None, name='Alice')

三、实战案例

3.1 缓存装饰器

from functools import wraps
import hashlib
import time
def cache(ttl=60):
"""缓存装饰器"""
cache_store = {}
def decorator(func):
@wraps(func)
def wrapper(args, *kwargs):
# 生成缓存键
key_data = str(args) + str(sorted(kwargs.items()))
key = hashlib.md5(key_data.encode()).hexdigest()
# 检查缓存
if key in cache_store:
cached_time, result = cache_store[key]
if time.time() - cached_time < ttl:
print(f"命中缓存: {func.__name__}")
return result
# 执行函数
result = func(args, *kwargs)
cache_store[key] = (time.time(), result)
print(f"缓存结果: {func.__name__}")
return result
return wrapper
return decorator
@cache(ttl=10)
def expensive_calculation(x, y):
print("执行计算...")
time.sleep(1)
return x + y
expensive_calculation(1, 2)  # 执行计算...
expensive_calculation(1, 2)  # 命中缓存

3.2 权限验证装饰器

from functools import wraps
class PermissionError(Exception):
pass
def require_permission(permission):
"""权限验证装饰器"""
def decorator(func):
@wraps(func)
def wrapper(user, args, *kwargs):
if permission not in user.get('permissions', []):
raise PermissionError(f"用户缺少权限: {permission}")
return func(user, args, *kwargs)
return wrapper
return decorator
@require_permission('admin')
def delete_user(user, user_id):
print(f"用户 {user['name']} 删除用户 {user_id}")
使用
admin = {'name': 'Admin', 'permissions': ['admin', 'user:delete']}
guest = {'name': 'Guest', 'permissions': []}
delete_user(admin, 1)  # OK
delete_user(guest, 1)  # PermissionError

3.3 事件驱动系统

class Order:
def __init__(self, order_id, amount):
self.order_id = order_id
self.amount = amount
self.status = 'pending'
class EventBus:
def __init__(self):
self._handlers = {}
def on(self, event, handler):
if event not in self._handlers:
self._handlers[event] = []
self._handlers[event].append(handler)
def emit(self, event, data):
if event in self._handlers:
for handler in self._handlers[event]:
handler(data)
事件处理器
def send_confirmation_email(order):
print(f"发送确认邮件: 订单 {order.order_id}")
def update_inventory(order):
print(f"更新库存: 订单 {order.order_id}")
def log_order(order):
print(f"记录订单: {order.order_id} - ${order.amount}")
注册事件
event_bus = EventBus()
event_bus.on('order.created', send_confirmation_email)
event_bus.on('order.created', update_inventory)
event_bus.on('order.created', log_order)
触发事件
order = Order('ORD001', 99.99)
event_bus.emit('order.created', order)

3.4 性能监控装饰器

from functools import wraps
import time
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def monitor_performance(threshold=1.0):
"""性能监控装饰器"""
def decorator(func):
@wraps(func)
def wrapper(args, *kwargs):
start = time.perf_counter()
try:
result = func(args, *kwargs)
elapsed = time.perf_counter() - start
if elapsed > threshold:
logger.warning(
f"{func.__name__} 执行缓慢: {elapsed:.3f}s "
f"(阈值: {threshold}s)"
)
else:
logger.info(
f"{func.__name__} 执行完成: {elapsed:.3f}s"
)
return result
except Exception as e:
elapsed = time.perf_counter() - start
logger.error(
f"{func.__name__} 执行失败: {e} "
f"(耗时: {elapsed:.3f}s)"
)
raise
return wrapper
return decorator
@monitor_performance(threshold=0.5)
def slow_function():
time.sleep(0.3)
return "完成"
slow_function()

四、模式组合

4.1 装饰器 + 观察者

class Observable:
"""可观察对象"""
def __init__(self):
self._observers = []
self._value = 0
def attach(self, observer):
self._observers.append(observer)
def _notify(self):
for observer in self._observers:
observer.update(self._value)
@property
def value(self):
return self._value
@value.setter
def value(self, new_value):
old_value = self._value
self._value = new_value
if old_value != new_value:
self._notify()
class Observer:
def update(self, value):
print(f"观察者收到新值: {value}")
使用
observable = Observable()
observer1 = Observer()
observer2 = Observer()
observable.attach(observer1)
observable.attach(observer2)
observable.value = 10  # 通知所有观察者
observable.value = 20  # 再次通知

4.2 工厂 + 装饰器

class Logger:
def log(self, message):
pass
class FileLogger(Logger):
def log(self, message):
print(f"[文件] {message}")
class ConsoleLogger(Logger):
def log(self, message):
print(f"[控制台] {message}")
class LoggerDecorator(Logger):
def __init__(self, logger):
self._logger = logger
def log(self, message):
self._logger.log(message)
class TimestampLogger(LoggerDecorator):
def log(self, message):
from datetime import datetime
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
self._logger.log(f"[{timestamp}] {message}")
class LevelLogger(LoggerDecorator):
def __init__(self, logger, level='INFO'):
super().__init__(logger)
self._level = level
def log(self, message):
self._logger.log(f"[{self._level}] {message}")
工厂创建组合装饰器
def create_logger(logger_type, *decorators):
if logger_type == 'file':
logger = FileLogger()
else:
logger = ConsoleLogger()
for decorator_class in decorators:
logger = decorator_class(logger)
return logger
使用
logger = create_logger(
'console',
TimestampLogger,
lambda l: LevelLogger(l, 'ERROR')
)
logger.log("测试消息")