下面从架构视角系统讲解装饰器模式在 Python 中的作用、结构、使用场景与优缺点，并附上一个较为复杂但清晰的示例代码。
一、作用
- 在不修改原对象（组件）代码的前提下，动态地为其“按需叠加”功能。通过组合多个装饰器形成调用链，使行为横切关注（如缓存、重试、鉴权、度量、限流、熔断、压缩/加密等）以模块化方式复用。
- 遵循开放-封闭原则：对扩展开放，对修改封闭。
- 相比继承，更强调组合：避免“子类爆炸”，并可在运行时灵活重组功能。
二、结构（关键组件与角色）
- Component（组件接口/抽象类）：定义核心功能的统一接口。例如 Fetcher.fetch(url) -> str。
- ConcreteComponent（具体组件）：核心业务实现。例如真实的网络请求或数据加载。
- Decorator（抽象装饰器）：持有一个 Component 引用，并实现同样的接口；默认将调用委托给内部组件。
- ConcreteDecorator（具体装饰器）：在调用前后或异常路径上增加附加行为（缓存、重试、日志、鉴权等），同时保持接口一致。多个装饰器可嵌套形成调用链。
三、常见使用场景
- I/O 或服务访问增强：重试、超时、熔断、限流、回退、幂等。
- 性能与稳定性：缓存（LRU/TTL）、批处理、压缩。
- 可观测性：日志、度量（计数、延迟）、追踪（trace id）。
- 安全：鉴权、签名、加密/解密。
- 领域规则：输入校验、幂等键、数据脱敏。
四、优缺点
- 优点：
- 高内聚、低耦合，职责分离清晰。
- 组合灵活，可按需叠加，替代多层继承。
- 符合开放-封闭原则，扩展容易。
- 缺点：
- 对象数量增多，调用栈加深，调试复杂。
- 叠加顺序敏感（例如重试与熔断的相对位置影响行为）。
- 可能引入非显式的控制流与状态，学习成本较高。
- 若需改变接口（方法签名），装饰器难以处理。
五、代码示例（复杂示例：在数据获取组件上叠加缓存、重试、熔断、度量）
说明：
- 定义 Fetcher 接口与一个不稳定的具体实现（模拟网络/服务偶发失败）。
- 叠加装饰器：CircuitBreaker（熔断）、Retry（重试）、Cache（TTL+LRU）、Metrics（度量）。
- 展示顺序对行为的影响：Cache -> Retry -> CircuitBreaker -> ConcreteComponent，最外层再加 Metrics。
代码：
import time
import random
from abc import ABC, abstractmethod
from collections import OrderedDict
from typing import Dict, Optional, Tuple, Any
# =========================
# 领域与错误类型
# =========================
class TransientError(Exception):
"""可重试的瞬时错误（例如网络抖动、服务暂不可用）。"""
pass
class NotFoundError(Exception):
"""不可重试的错误（例如资源不存在）。"""
pass
class CircuitOpenError(Exception):
"""熔断器打开，拒绝调用。"""
pass
# =========================
# 组件接口与具体实现
# =========================
class Fetcher(ABC):
"""Component：数据获取接口"""
@abstractmethod
def fetch(self, url: str, headers: Optional[Dict[str, str]] = None) -> str:
pass
class UnstableFetcher(Fetcher):
"""
ConcreteComponent：不稳定的数据源，模拟真实服务的随机失败与延迟。
- failure_rate: 随机失败概率（TransientError）
- latency_range: 延迟范围（秒）
- data: 简单的URL->字符串数据映射
"""
def __init__(self, data: Dict[str, str], failure_rate: float = 0.3, latency_range: Tuple[float, float] = (0.05, 0.2)):
self.data = data
self.failure_rate = failure_rate
self.latency_range = latency_range
def fetch(self, url: str, headers: Optional[Dict[str, str]] = None) -> str:
# 模拟随机延迟
time.sleep(random.uniform(*self.latency_range))
# 模拟不可重试错误：不存在
if url not in self.data:
raise NotFoundError(f"URL not found: {url}")
# 模拟可重试错误：瞬时失败
if random.random() < self.failure_rate:
raise TransientError("Random transient failure.")
return self.data[url]
# =========================
# 抽象装饰器
# =========================
class FetcherDecorator(Fetcher):
"""Decorator：持有被装饰对象，并实现同样的接口"""
def __init__(self, inner: Fetcher):
self._inner = inner
def fetch(self, url: str, headers: Optional[Dict[str, str]] = None) -> str:
return self._inner.fetch(url, headers=headers)
# =========================
# Metrics 度量中心（供多个装饰器共享）
# =========================
class MetricsRegistry:
def __init__(self):
self.counters: Dict[str, int] = {}
self.timers: Dict[str, float] = {}
def inc(self, key: str, n: int = 1):
self.counters[key] = self.counters.get(key, 0) + n
def add_time(self, key: str, ms: float):
self.timers[key] = self.timers.get(key, 0.0) + ms
def snapshot(self) -> Dict[str, Any]:
return {
"counters": dict(self.counters),
"timers_ms": dict(self.timers),
}
# =========================
# ConcreteDecorator：熔断器
# =========================
class CircuitBreakerFetcher(FetcherDecorator):
"""
熔断器状态机：
- closed: 正常。累积失败达到阈值 -> open
- open: 拒绝请求，等待恢复时间 -> half_open
- half_open: 允许一次试探请求；成功 -> closed；失败 -> open
"""
def __init__(self, inner: Fetcher, failure_threshold: int = 3, recovery_time: float = 3.0, metrics: Optional[MetricsRegistry] = None):
super().__init__(inner)
self.failure_threshold = failure_threshold
self.recovery_time = recovery_time
self.metrics = metrics
self._state = "closed"
self._consecutive_failures = 0
self._opened_at = 0.0
self._half_open_trial_in_progress = False
def _to_open(self):
self._state = "open"
self._opened_at = time.time()
self._half_open_trial_in_progress = False
def _maybe_to_half_open(self):
if self._state == "open" and (time.time() - self._opened_at) >= self.recovery_time:
self._state = "half_open"
self._half_open_trial_in_progress = False
def fetch(self, url: str, headers: Optional[Dict[str, str]] = None) -> str:
# open -> half_open if冷却结束
self._maybe_to_half_open()
if self._state == "open":
if self.metrics:
self.metrics.inc("circuit_open_skips")
raise CircuitOpenError("Circuit breaker is OPEN")
if self._state == "half_open":
# 允许一次试探请求
if self._half_open_trial_in_progress:
# 有另一个试探中，拒绝（简单处理并发；单线程可忽略）
raise CircuitOpenError("Half-open trial in progress")
self._half_open_trial_in_progress = True
try:
result = self._inner.fetch(url, headers=headers)
# 成功：重置并转为closed（若half_open）
self._consecutive_failures = 0
if self._state == "half_open":
self._state = "closed"
self._half_open_trial_in_progress = False
return result
except Exception as e:
# 失败：计数并根据阈值转open
self._consecutive_failures += 1
if self._state == "half_open":
# half-open失败立即转open
self._to_open()
elif self._consecutive_failures >= self.failure_threshold:
self._to_open()
raise
# =========================
# ConcreteDecorator：重试（指数退避+抖动）
# =========================
class RetryFetcher(FetcherDecorator):
"""
对指定异常进行重试，支持指数退避与随机抖动。
- retry_exceptions: 需要重试的异常类型集合
- giveup_exceptions: 直接放弃的异常，不重试（如 CircuitOpenError、NotFoundError）
"""
def __init__(self, inner: Fetcher, retries: int = 3, backoff_base: float = 0.1,
retry_exceptions: Tuple[type, ...] = (TransientError,),
giveup_exceptions: Tuple[type, ...] = (CircuitOpenError, NotFoundError),
metrics: Optional[MetricsRegistry] = None):
super().__init__(inner)
self.retries = retries
self.backoff_base = backoff_base
self.retry_exceptions = retry_exceptions
self.giveup_exceptions = giveup_exceptions
self.metrics = metrics
def fetch(self, url: str, headers: Optional[Dict[str, str]] = None) -> str:
attempt = 0
while True:
try:
return self._inner.fetch(url, headers=headers)
except self.giveup_exceptions:
# 直接放弃，不做重试
raise
except self.retry_exceptions as e:
attempt += 1
if self.metrics:
self.metrics.inc("retries_attempted")
if attempt > self.retries:
raise
# 指数退避 + 随机抖动
delay = self.backoff_base * (2 ** (attempt - 1)) * random.uniform(0.8, 1.2)
time.sleep(delay)
# =========================
# ConcreteDecorator：TTL + LRU 缓存
# =========================
class TTLCache:
"""简单的TTL+LRU缓存实现，用于示例，不考虑并发。"""
def __init__(self, capacity: int = 128, ttl_seconds: float = 5.0):
self.capacity = capacity
self.ttl = ttl_seconds
self.store: OrderedDict[Any, Tuple[float, Any]] = OrderedDict()
def _evict_if_needed(self):
while len(self.store) > self.capacity:
self.store.popitem(last=False)
def _purge_expired(self):
now = time.time()
expired_keys = [k for k, (exp, _) in self.store.items() if exp < now]
for k in expired_keys:
del self.store[k]
def get(self, key: Any) -> Optional[Any]:
self._purge_expired()
if key in self.store:
exp, val = self.store.pop(key) # 触发LRU移动
self.store[key] = (exp, val)
return val
return None
def put(self, key: Any, value: Any):
exp = time.time() + self.ttl
if key in self.store:
del self.store[key]
self.store[key] = (exp, value)
self._evict_if_needed()
class CacheFetcher(FetcherDecorator):
def __init__(self, inner: Fetcher, cache: TTLCache, metrics: Optional[MetricsRegistry] = None):
super().__init__(inner)
self.cache = cache
self.metrics = metrics
def _key(self, url: str, headers: Optional[Dict[str, str]]) -> Tuple[str, Tuple[Tuple[str, str], ...]]:
headers_tuple = tuple(sorted((headers or {}).items()))
return (url, headers_tuple)
def fetch(self, url: str, headers: Optional[Dict[str, str]] = None) -> str:
key = self._key(url, headers)
val = self.cache.get(key)
if val is not None:
if self.metrics:
self.metrics.inc("cache_hits")
return val
if self.metrics:
self.metrics.inc("cache_misses")
val = self._inner.fetch(url, headers=headers)
self.cache.put(key, val)
return val
# =========================
# ConcreteDecorator：度量
# =========================
class MetricsFetcher(FetcherDecorator):
def __init__(self, inner: Fetcher, metrics: MetricsRegistry):
super().__init__(inner)
self.metrics = metrics
def fetch(self, url: str, headers: Optional[Dict[str, str]] = None) -> str:
self.metrics.inc("fetch_calls")
start = time.time()
try:
result = self._inner.fetch(url, headers=headers)
self.metrics.inc("fetch_success")
return result
except Exception:
self.metrics.inc("fetch_failure")
raise
finally:
elapsed_ms = (time.time() - start) * 1000.0
self.metrics.add_time("fetch_latency_ms_total", elapsed_ms)
# =========================
# 演示：构建装饰器调用链并运行
# =========================
if __name__ == "__main__":
# 准备示例数据与度量中心
data = {
"item/42": "The answer is 42.",
"item/7": "Lucky number is 7.",
}
metrics = MetricsRegistry()
# ConcreteComponent
base = UnstableFetcher(data=data, failure_rate=0.5, latency_range=(0.05, 0.15))
# 构建调用链（顺序很重要）：
# Cache -> Retry -> CircuitBreaker -> base，最外层再加 Metrics
cache = TTLCache(capacity=64, ttl_seconds=2.0)
chain = MetricsFetcher(
inner=CacheFetcher(
inner=RetryFetcher(
inner=CircuitBreakerFetcher(
inner=base,
failure_threshold=3, # 连续3次失败打开熔断
recovery_time=3.0, # 3秒后进入half-open
metrics=metrics
),
retries=4, # 最多4次补偿重试
backoff_base=0.08, # 初始退避
retry_exceptions=(TransientError,),
giveup_exceptions=(CircuitOpenError, NotFoundError),
metrics=metrics
),
cache=cache,
metrics=metrics
),
metrics=metrics
)
# 连续请求以观察：缓存、重试、熔断与恢复的行为
url = "item/42"
for i in range(1, 12):
try:
val = chain.fetch(url)
print(f"[{i:02d}] OK: {val}")
except CircuitOpenError as e:
print(f"[{i:02d}] CIRCUIT OPEN: {e}")
except NotFoundError as e:
print(f"[{i:02d}] NOT FOUND: {e}")
except TransientError as e:
print(f"[{i:02d}] TRANSIENT FAILURE (no more retries): {e}")
except Exception as e:
print(f"[{i:02d}] OTHER ERROR: {e}")
time.sleep(0.7) # 控制节奏，观察TTL与熔断恢复
print("\nMetrics snapshot:")
print(metrics.snapshot())
要点与实践建议：
- 装饰器顺序影响语义：
- 尽量让 Cache 处于外层以减少下游开销。
- Retry 通常放在 CircuitBreaker 外层，避免熔断后仍盲目重试；并配置 giveup_exceptions 包含 CircuitOpenError。
- 将横切关注的状态与指标统一收敛到 MetricsRegistry，有利于观测与测试。
- 在生产环境中：
- 封装可重试异常清单（网络超时、断开连接等），加入指数退避与最大等待。
- 熔断器需考虑并发安全、滑动窗口统计、半开并发限制。
- 缓存策略需结合业务一致性要求（TTL、失效、回源策略）。
- Python 中的函数/方法装饰器（@decorator）与该 OO 装饰器模式思想一致，都是在保持接口不变的前提下包装功能；本示例展示的是面向对象形式，适于需要多层组合与复杂状态的场景。
