Python 设计模式
工厂模式
核心思想：集中管理对象创建，调用方只说"我要什么"。
简单工厂
from abc import ABC, abstractmethod
class Payment(ABC):
@abstractmethod
def pay(self, amount): pass
class Alipay(Payment):
def pay(self, amount): print(f"支付宝支付 {amount} 元")
class WechatPay(Payment):
def pay(self, amount): print(f"微信支付 {amount} 元")
class BankCard(Payment):
def pay(self, amount): print(f"银行卡支付 {amount} 元")
class PaymentFactory:
@staticmethod
def create(method: str) -> Payment:
if method == "alipay": return Alipay()
if method == "wechat": return WechatPay()
if method == "bank": return BankCard()
raise ValueError(f"未知支付方式: {method}")注册式工厂（推荐）
简单工厂每新增一种类型都要修改 create 方法，违反开闭原则。注册式工厂解决这个问题：
class PaymentFactory:
_registry: dict[str, type[Payment]] = {}
@classmethod
def register(cls, name: str):
"""装饰器形式注册，更优雅"""
def decorator(payment_cls):
cls._registry[name] = payment_cls
return payment_cls
return decorator
@classmethod
def create(cls, name: str) -> Payment:
if name not in cls._registry:
raise ValueError(f"未知支付方式: {name}")
return cls._registry[name]()
@PaymentFactory.register("alipay")
class Alipay(Payment):
def pay(self, amount): print(f"支付宝支付 {amount} 元")
@PaymentFactory.register("wechat")
class WechatPay(Payment):
def pay(self, amount): print(f"微信支付 {amount} 元")
# 新增支付方式：只需加这几行，不改工厂代码
@PaymentFactory.register("crypto")
class CryptoPay(Payment):
def pay(self, amount): print(f"加密货币支付 {amount} 元")
p = PaymentFactory.create("alipay")
p.pay(100) # 支付宝支付 100 元单例模式
核心思想：无论实例化多少次，都返回同一个实例。
基础实现：__new__ 拦截
class Singleton:
_instance = None
_initialized = False
def __new__(cls, *args, **kwargs):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self):
if self._initialized: # 防止 __init__ 每次都重新初始化
return
self._initialized = True
self.data = {}线程安全实现（生产环境推荐）
基础版本在多线程并发创建时有竞态条件，用 double-checked locking 解决：
import threading
class Singleton:
_instance = None
_lock = threading.Lock()
def __new__(cls, *args, **kwargs):
if cls._instance is None: # 第一次检查（无锁，快速）
with cls._lock:
if cls._instance is None: # 第二次检查（有锁，防竞态）
cls._instance = super().__new__(cls)
return cls._instance装饰器实现（更 Pythonic）
import functools
def singleton(cls):
instances = {}
@functools.wraps(cls)
def get_instance(*args, **kwargs):
if cls not in instances:
instances[cls] = cls(*args, **kwargs)
return instances[cls]
return get_instance
@singleton
class DatabasePool:
def __init__(self):
print("初始化连接池")
self.connections = []适用场景：配置中心、日志器、数据库连接池、全局缓存。
不适用：普通业务对象（User、Order 等天然需要多个实例的对象）。
Python 的模块天然是单例——模块级对象只会初始化一次，很多场景可以直接用模块级变量代替单例类。
用
__new__实现单例，见 python-oop-basics → 「__new__与__init__」
策略模式
核心思想：把可变行为抽象为独立对象，运行时可动态替换。
from abc import ABC, abstractmethod
# 1. 定义策略接口
class SortStrategy(ABC):
@abstractmethod
def sort(self, data: list) -> list: pass
# 2. 实现具体策略
class BubbleSort(SortStrategy):
def sort(self, data):
data = data.copy()
n = len(data)
for i in range(n):
for j in range(n - i - 1):
if data[j] > data[j+1]:
data[j], data[j+1] = data[j+1], data[j]
return data
class QuickSort(SortStrategy):
def sort(self, data):
if len(data) <= 1: return data
pivot = data[len(data) // 2]
left = [x for x in data if x < pivot]
mid = [x for x in data if x == pivot]
right = [x for x in data if x > pivot]
return self.sort(left) + mid + self.sort(right)
# 3. 上下文持有策略对象
class Sorter:
def __init__(self, strategy: SortStrategy):
self.strategy = strategy
def sort(self, data):
return self.strategy.sort(data)
data = [5, 2, 8, 1, 9]
sorter = Sorter(QuickSort())
print(sorter.sort(data)) # [1, 2, 5, 8, 9]
sorter.strategy = BubbleSort() # 运行时切换算法
print(sorter.sort(data)) # [1, 2, 5, 8, 9]触发信号：代码里出现大量 if/elif，且每个分支对应不同的"行为实现"时，考虑策略模式。
观察者模式
核心思想：被观察者（Subject）变化时，自动通知所有已注册的观察者，实现一对多的解耦通知。
from abc import ABC, abstractmethod
from typing import Callable
class EventEmitter:
"""更轻量的函数式观察者实现"""
def __init__(self):
self._handlers: dict[str, list[Callable]] = {}
def on(self, event: str, handler: Callable):
self._handlers.setdefault(event, []).append(handler)
def off(self, event: str, handler: Callable):
self._handlers.get(event, []).remove(handler)
def emit(self, event: str, *args, **kwargs):
for handler in self._handlers.get(event, []):
handler(*args, **kwargs)
class UserService(EventEmitter):
def register(self, username: str):
print(f"用户 {username} 注册成功")
self.emit("user_registered", username) # 触发事件，不关心谁来处理
svc = UserService()
# 观察者只需要注册回调函数
svc.on("user_registered", lambda u: print(f"[邮件] 发送欢迎邮件给 {u}"))
svc.on("user_registered", lambda u: print(f"[积分] 新用户 {u} 获得 100 积分"))
svc.on("user_registered", lambda u: print(f"[日志] 记录注册事件: {u}"))
svc.register("Alice")
# 用户 Alice 注册成功
# [邮件] 发送欢迎邮件给 Alice
# [积分] 新用户 Alice 获得 100 积分
# [日志] 记录注册事件: Alice触发信号：主逻辑完成后跟着一系列附加操作，且这些操作需要独立扩展时。
装饰器模式
核心思想：通过包装动态地为对象添加功能，包装后保持相同接口。
GoF 风格（对象组合）
from abc import ABC, abstractmethod
class TextProcessor(ABC):
@abstractmethod
def process(self, text: str) -> str: pass
class PlainText(TextProcessor):
"""被装饰的原始对象"""
def process(self, text: str) -> str:
return text
class TextDecorator(TextProcessor):
"""装饰器基类：持有一个 TextProcessor，并实现相同接口"""
def __init__(self, component: TextProcessor):
self._component = component
def process(self, text: str) -> str:
return self._component.process(text) # 默认透传
class UpperDecorator(TextDecorator):
def process(self, text: str) -> str:
return super().process(text).upper()
class TrimDecorator(TextDecorator):
def process(self, text: str) -> str:
return super().process(text).strip()
class ExclamationDecorator(TextDecorator):
def process(self, text: str) -> str:
return super().process(text) + "!!!"
# 装饰器可以任意叠加，顺序决定处理顺序
processor = ExclamationDecorator(
UpperDecorator(
TrimDecorator(PlainText())
)
)
print(processor.process(" hello world ")) # HELLO WORLD!!!Python 函数装饰器
import functools
def log(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
print(f"→ 调用 {func.__name__}")
result = func(*args, **kwargs)
print(f"← {func.__name__} 返回 {result!r}")
return result
return wrapper
def retry(times=3):
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
for i in range(times):
try:
return func(*args, **kwargs)
except Exception as e:
if i == times - 1: raise
print(f"第 {i+1} 次失败，重试...")
return wrapper
return decorator
@retry(times=3)
@log
def fetch_data(url):
return f"data from {url}"Python 函数装饰器语法见 python-core-mechanisms → 「装饰器」
适配器模式
核心思想：把一个类的接口转换成另一个，让接口不兼容的类协同工作。
from abc import ABC, abstractmethod
# 新系统期望的接口
class PaymentGateway(ABC):
@abstractmethod
def charge(self, amount: float, currency: str) -> bool: pass
# 已有的旧系统（无法修改）
class LegacyPaymentSDK:
def make_payment(self, amount_cents: int) -> str:
print(f"旧系统处理支付: {amount_cents} 分")
return "SUCCESS"
class ThirdPartyPayAPI:
def process(self, usd_amount: float) -> dict:
print(f"第三方 API 处理 ${usd_amount}")
return {"status": "ok"}
# 适配器：把旧接口包装成新接口
class LegacyAdapter(PaymentGateway):
def __init__(self):
self._sdk = LegacyPaymentSDK()
def charge(self, amount: float, currency: str) -> bool:
cents = int(amount * 100) # 接口转换：元 → 分
result = self._sdk.make_payment(cents)
return result == "SUCCESS"
class ThirdPartyAdapter(PaymentGateway):
def __init__(self):
self._api = ThirdPartyPayAPI()
def charge(self, amount: float, currency: str) -> bool:
usd = amount / 7.2 if currency == "CNY" else amount
result = self._api.process(usd)
return result["status"] == "ok"
# 新系统只知道 PaymentGateway 接口，不知道底层实现
def checkout(gateway: PaymentGateway, amount: float):
success = gateway.charge(amount, "CNY")
print("支付成功" if success else "支付失败")
checkout(LegacyAdapter(), 99.9)
checkout(ThirdPartyAdapter(), 99.9)工程场景：
- 集成第三方 SDK，其接口与系统期望不一致
- 新旧系统迁移过渡阶段
- 统一多个异构数据源为相同的读取接口
OOP 基础见 → python-oop-advanced