设计模式

设计模式

设计模式

设计模式(Design Patterns)是软件设计中反复出现的问题的成熟解决方案,是前辈工程师经验的总结。它们不是代码库,而是"解决特定问题的思路模板"。

本教程内容

本教程系统介绍 23 个经典 GoF 设计模式(创建型、结构型、行为型)以及 10 个 J2EE 模式,每个模式包含:意图、结构、Python 代码示例、适用场景与优缺点。

为什么学习设计模式

• 沟通语言:说"单例"比解释十分钟更高效
• 复用经验:站在前人的肩膀上解决问题
• 规范设计:让代码可扩展、可维护
• 面试必备:高级岗位的常见考点

模式分类

# 创建型(5): 对象的创建
工厂 / 抽象工厂 / 单例 / 建造者 / 原型

# 结构型(7): 对象的组合
适配器 / 桥接 / 组合 / 装饰器 / 外观
享元 / 代理

# 行为型(11): 对象间的交互
责任链 / 命令 / 解释器 / 迭代器 / 中介者
备忘录 / 观察者 / 状态 / 策略 / 模板 / 访问者

# J2EE 模式(10): 表现层/服务层
MVC / 业务代表 / 组合实体 / DAO / 前端控制器
拦截过滤器 / 服务定位器 / 传输对象 等

学习建议

# 1. 先理解"解决什么问题",再记结构
# 2. 每个模式动手写一遍示例
# 3. 在真实项目中识别模式的使用场景
# 4. 不要滥用: 简单场景用简单方案
# 5. 模式是工具,不是目标

# 黄金法则:
# 简单问题不要为了"用模式"而用模式
# 当代码开始"重复/难以修改"时,
# 才考虑用模式重构

设计模式是工程师的"内功心法"。掌握这些套路,代码设计与沟通都会上一个台阶。

设计模式简介

设计模式简介

设计模式源于建筑学,由"四人组"(Gang of Four, GoF)于 1994 年在《Design Patterns》一书中系统化,奠定了面向对象设计模式的经典体系。

核心原则

# SOLID 原则(设计模式的基础):
# S - 单一职责: 一个类只做一件事
# O - 开闭原则: 对扩展开放,对修改关闭
# L - 里氏替换: 子类可替换父类
# I - 接口隔离: 最小接口
# D - 依赖倒置: 依赖抽象,不依赖具体

# 设计模式多数在实践这些原则:
# - 工厂/抽象工厂 → 开闭原则
# - 观察者 → 依赖倒置
# - 装饰器 → 开闭原则
# - 适配器 → 接口隔离

模式要素

# 每个模式包含四个要素:
# 1. 模式名称: 简洁的命名(沟通语言)
# 2. 问题: 何时使用(适用场景)
# 3. 解决方案: 结构与参与者
# 4. 效果: 优缺点与权衡

# 例: 单例模式
# 名称: Singleton
# 问题: 全局唯一实例(配置/连接池)
# 方案: 私有构造 + 静态获取
# 效果: 省资源,但难测试/难扩展

如何选择模式

# 按需求选择:
# 1. 创建对象复杂 → 工厂/建造者
# 2. 需要唯一实例 → 单例
# 3. 接口不匹配 → 适配器
# 4. 动态加功能 → 装饰器
# 5. 一对多通知 → 观察者
# 6. 算法可替换 → 策略
# 7. 行为随状态变 → 状态
# 8. 对象结构复杂 → 组合
# 9. 控制访问 → 代理
# 10. 解耦请求处理 → 责任链

# 误用信号:
# - 代码比不用模式更复杂
# - 为了"炫技"而套模式
# - 模式导致过度设计

模式的辩证看法

# 优点:
# - 经验沉淀,减少试错
# - 代码更易读(模式即文档)
# - 便于团队协作

# 缺点/风险:
# - 增加抽象与复杂度
# - 可能过度设计
# - 语言特性可简化部分模式
#   (Python 中单例可用模块、装饰器更简)

# 现代视角:
# - 函数式编程提供了新思路
# - 设计模式是"工具包"而非"教条"
# - 简单 > 模式: 优先可读性与简洁性

Python 中的设计模式

# Python 动态特性简化了部分模式:
# 单例 → 模块级对象
# 迭代器 → 生成器
# 命令 → 函数/闭包
# 策略 → 函数参数

# 但仍推荐学习经典模式:
# - 理解主流代码库(JAVA/C#)
# - 大型项目分层与扩展
# - 面试与设计能力

# 学习路径:
# 1. 掌握 SOLID 原则
# 2. 熟悉 23 个 GoF 模式(本教程)
# 3. 结合真实场景实践
# 4. 反思: 何时该用、何时不该用

设计模式是"经验结晶"。带着"解决什么问题"的视角学习,在实践中拿捏分寸,才是正确的打开方式。

工厂模式

工厂模式(Factory Pattern)

工厂模式:用一个"工厂"创建对象,客户端不直接 new,而是通过工厂获取。解耦"创建"与"使用"。

意图

# 问题: 客户端直接创建对象,耦合具体类
# 变化: 新增产品类型 → 修改所有创建处

# 解决: 将创建逻辑集中到工厂
# 客户端只依赖"产品接口"与"工厂"

# 优点:
# - 解耦创建与使用
# - 新增产品只需扩展工厂
# - 创建逻辑集中管理

# 缺点:
# - 每加一个产品要改工厂(简单工厂)
# - 增加类数量

结构

# 参与者:
# Product: 产品接口
# ConcreteProduct: 具体产品
# Factory: 工厂(负责创建)

# 关系:
# Client → Factory → Product

Python 示例

from abc import ABC, abstractmethod

# 产品接口
class Transport(ABC):
    @abstractmethod
    def deliver(self) -> str: ...

# 具体产品
class Truck(Transport):
    def deliver(self) -> str:
        return "用卡车运输"

class Ship(Transport):
    def deliver(self) -> str:
        return "用轮船运输"

# 简单工厂(静态方法)
class TransportFactory:
    @staticmethod
    def create(t_type: str) -> Transport:
        if t_type == "truck":
            return Truck()
        elif t_type == "ship":
            return Ship()
        raise ValueError(f"未知类型: {t_type}")

# 客户端
def client(factory, t_type: str):
    transport = factory.create(t_type)
    return transport.deliver()

print(client(TransportFactory, "truck"))
# 用卡车运输

# 扩展: 新增"火车"只需加产品类 + 工厂分支
class Train(Transport):
    def deliver(self) -> str:
        return "用火车运输"

工厂方法(Factory Method)

# 变体: 工厂方法(每个产品一个工厂)
class Logistics(ABC):
    @abstractmethod
    def create_transport(self) -> Transport: ...

    def plan_delivery(self) -> str:
        # 模板方法: 使用产品
        t = self.create_transport()
        return f"配送方案: {t.deliver()}"

class RoadLogistics(Logistics):
    def create_transport(self) -> Transport:
        return Truck()

class SeaLogistics(Logistics):
    def create_transport(self) -> Transport:
        return Ship()

# 客户端只与 Logistics 打交道
road = RoadLogistics()
print(road.plan_delivery())   # 用卡车运输

# 新增产品: 加产品类 + 加对应工厂,不改旧代码
# (符合开闭原则)

适用场景

# 1. 创建对象逻辑复杂(多参数/条件分支)
# 2. 运行时才确定具体类型(配置/参数驱动)
# 3. 需要统一管理对象的创建(日志/计数)

# 实际案例:
# - 数据库连接工厂(MySQL/PostgreSQL)
# - 支付工厂(支付宝/微信/银联)
# - 解析器工厂(JSON/XML/YAML)
# - 消息队列生产/消费者工厂

# Python 简化:
# 简单场景直接用 dict 映射
TRANSPORTS = {
    "truck": Truck,
    "ship": Ship,
}

def create(t_type: str):
    cls = TRANSPORTS.get(t_type)
    if not cls:
        raise ValueError(f"未知: {t_type}")
    return cls()

优缺点总结

# 优点:
# ✅ 解耦创建与使用
# ✅ 创建逻辑集中
# ✅ 符合开闭原则(工厂方法)

# 缺点:
# ❌ 类数量增加
# ❌ 简单场景过度设计
# ❌ 复杂产品层次时工厂复杂

# 决策:
# 简单创建 → 直接用构造器/dict 工厂
# 复杂/多变产品族 → 工厂方法/抽象工厂

工厂模式是"创建对象"的解耦利器。掌握简单工厂与工厂方法,新增产品类型不再牵一发动全身。

抽象工厂模式

抽象工厂模式(Abstract Factory)

抽象工厂:创建"一族相关对象"的工厂。与工厂方法的区别:工厂方法创建一种产品,抽象工厂创建一族(多个类型)产品。

意图

# 问题: 需要创建一组配套的对象
# 例: UI 主题(按钮+输入框+对话框必须风格一致)

# 解决: 抽象工厂接口 + 具体工厂(每族一个)
# 客户端通过工厂获取整族产品,保证配套

# 优点:
# - 保证产品族一致性
# - 客户端与具体类解耦
# - 换族只需换工厂

# 缺点:
# - 新增产品类型困难(改所有工厂)
# - 增加类数量

结构

# 参与者:
# AbstractFactory: 抽象工厂(创建族中每种产品)
# ConcreteFactory: 具体工厂(一种风格/平台)
# AbstractProduct: 抽象产品(每类产品的接口)
# ConcreteProduct: 具体产品

# 例:
# UIFactory(抽象)
#   ├── LightFactory(浅色族)
#   └── DarkFactory(深色族)
# 每族都生产 Button / Input / Dialog

Python 示例

from abc import ABC, abstractmethod

# 抽象产品
class Button(ABC):
    @abstractmethod
    def render(self) -> str: ...

class Dialog(ABC):
    @abstractmethod
    def show(self) -> str: ...

# 浅色族产品
class LightButton(Button):
    def render(self) -> str:
        return "浅色按钮"

class LightDialog(Dialog):
    def show(self) -> str:
        return "浅色对话框"

# 深色族产品
class DarkButton(Button):
    def render(self) -> str:
        return "深色按钮"

class DarkDialog(Dialog):
    def show(self) -> str:
        return "深色对话框"

# 抽象工厂
class UIFactory(ABC):
    @abstractmethod
    def create_button(self) -> Button: ...
    @abstractmethod
    def create_dialog(self) -> Dialog: ...

# 具体工厂(浅色族)
class LightFactory(UIFactory):
    def create_button(self) -> Button:
        return LightButton()
    def create_dialog(self) -> Dialog:
        return LightDialog()

# 深色族
class DarkFactory(UIFactory):
    def create_button(self) -> Button:
        return DarkButton()
    def create_dialog(self) -> Dialog:
        return DarkDialog()

# 客户端: 依赖抽象工厂
def render_ui(factory: UIFactory) -> str:
    btn = factory.create_button()
    dlg = factory.create_dialog()
    return f"{btn.render()} + {dlg.show()}"

print(render_ui(LightFactory()))
# 浅色按钮 + 浅色对话框

print(render_ui(DarkFactory()))
# 深色按钮 + 深色对话框

# 切换主题: 换工厂即可,客户端零改动

典型场景

# 1. UI 主题/风格体系
# 2. 跨平台组件(Windows/Mac/Linux 控件)
# 3. 数据库操作族(连接+游标+事务)
# 4. 支付体系(支付宝族: 支付+退款+查询)
# 5. 报告生成(PDF 族/Excel 族: 头+体+尾)

# Python 生态示例:
# - Django settings 多环境配置
# - 多后端缓存(cache backends)
# - 测试替身工厂(mock factory)

# 实际案例: 主题切换
theme = DarkFactory() if dark_mode else LightFactory()
ui = render_ui(theme)

工厂方法 vs 抽象工厂

# 工厂方法:
# 创建"一种"产品
# 通过继承扩展(子类重写工厂方法)

# 抽象工厂:
# 创建"一族"相关产品
# 通过组合扩展(注入具体工厂)

# 选择:
# 单一产品变化 → 工厂方法
# 产品族配套 → 抽象工厂

# 示例对比:
# 创建"运输工具"(一种) → 工厂方法
# 创建"UI 组件族"(多种配套) → 抽象工厂

优缺点总结

# 优点:
# ✅ 产品族一致性
# ✅ 解耦客户端
# ✅ 符合开闭原则(新增族易)

# 缺点:
# ❌ 新增产品类型难(违反开闭)
# ❌ 类爆炸(每族每产品一个类)

# 简化(Python):
# 用 dict 注册工厂
FACTORIES = {
    "light": LightFactory,
    "dark": DarkFactory,
}
factory = FACTORIES[theme_name]()

抽象工厂解决"产品族配套"问题。掌握族的组织方式,主题、跨平台、多后端等场景应对自如。

单例模式

单例模式(Singleton Pattern)

单例:确保一个类只有一个实例,并提供全局访问点。用于配置、日志、连接池等全局唯一对象。

意图

# 问题: 某些对象应该全局唯一
# - 配置中心(读取一份配置)
# - 日志器(统一写入)
# - 数据库连接池(复用连接)

# 解决: 控制构造,保证只创建一个实例

# 优点:
# - 全局唯一实例
# - 节省资源(懒加载)
# - 全局访问点

# 缺点:
# - 隐藏依赖(全局状态难测试)
# - 违反单一职责(创建+业务)
# - 并发初始化需注意

Python 实现方式

# 方式1: 模块级单例(最简单,推荐)
# config.py
class Config:
    def __init__(self):
        self.debug = False
        self.db_url = "sqlite:///app.db"

config = Config()   # 模块加载时创建一次

# 使用
from config import config
print(config.debug)

# 方式2: __new__ 控制
class Singleton:
    _instance = None

    def __new__(cls, *args, **kwargs):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

    def __init__(self):
        # 注意: __init__ 每次调用都会执行
        # 需自行控制初始化
        pass

a = Singleton()
b = Singleton()
print(a is b)   # True(同一实例)

线程安全单例

import threading

class ThreadSafeSingleton:
    _instance = None
    _lock = threading.Lock()

    def __new__(cls):
        # 双检锁: 先查后锁,减少锁开销
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:
                    cls._instance = super().__new__(cls)
        return cls._instance

# 验证多线程下唯一
def worker():
    s = ThreadSafeSingleton()
    assert s is ThreadSafeSingleton._instance

threads = [threading.Thread(target=worker)
           for _ in range(10)]
for t in threads: t.start()
for t in threads: t.join()
print("所有线程拿到同一实例")

元类单例(进阶)

# 用元类统一实现
class SingletonMeta(type):
    _instances = {}
    _lock = threading.Lock()

    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            with cls._lock:
                if cls not in cls._instances:
                    cls._instances[cls] =                         super().__call__(*args, **kwargs)
        return cls._instances[cls]

class Logger(metaclass=SingletonMeta):
    def __init__(self):
        self.logs = []

    def log(self, msg):
        self.logs.append(msg)
        print(f"[LOG] {msg}")

log1 = Logger()
log2 = Logger()
print(log1 is log2)   # True
log1.log("hello")
print(len(log2.logs)) # 1(共享状态)

装饰器单例

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):
        self.connected = False
        self._init_connection()

    def _init_connection(self):
        # 模拟连接
        print("创建数据库连接")
        self.connected = True

db1 = Database()   # 创建数据库连接
db2 = Database()   # (不再创建)
print(db1 is db2)  # True

适用场景与反模式

# 适用:
# ✅ 配置中心
# ✅ 日志器
# ✅ 连接池/线程池
# ✅ 缓存管理器

# 慎用/反模式:
# ❌ 隐藏依赖的全局状态(难测试)
# ❌ 业务对象强行单例
# ❌ 仅为了"方便访问"而单例

# 现代实践:
# - 依赖注入(DI)替代单例(可测试)
# - Python 模块天然单例(优先)
# - functools.lru_cache 做"值单例"

# 依赖注入示例
class App:
    def __init__(self, config):   # 注入
        self.config = config
# 测试时传入 mock 配置,无需单例

优缺点总结

# 优点:
# ✅ 唯一实例,省资源
# ✅ 全局访问方便
# ✅ 懒加载

# 缺点:
# ❌ 全局状态,测试困难
# ❌ 隐藏依赖关系
# ❌ 多线程需注意

# 决策:
# 简单场景 → 模块级对象
# 需要延迟/控制 → __new__ 或元类
# 团队项目 → 考虑依赖注入替代

单例是"唯一实例"的保障。掌握 Python 的多种实现(模块/__new__/元类/装饰器),按场景选择,避免滥用。

建造者模式

建造者模式(Builder Pattern)

建造者:将复杂对象的构建过程与表示分离,让同一构建过程可以创建不同表示。

意图

# 问题: 对象构造参数多、步骤复杂
# - 直接构造器: 参数顺序/可选参数难管理
# - 大量 setter: 忘记设置导致不完整

# 解决: 分步构建,一步一个方法
# 最后调用 build() 产出完整对象

# 优点:
# - 构建过程清晰(方法名即步骤)
# - 校验在 build() 统一做
# - 可复用构建流程
# - 参数多时比构造器更可读

# 缺点:
# - 类数量增加
# - 简单对象没必要

结构

# 参与者:
# Director: 指挥者(可选,编排构建步骤)
# Builder: 构建器接口(分步方法)
# ConcreteBuilder: 具体构建器
# Product: 最终产品

# 简化版(常用):
# Builder: 链式方法 + build()
# 无需 Director(客户端直接调用)

Python 示例: 链式建造者

class Pizza:
    def __init__(self):
        self.size = "中"
        self.cheese = False
        self.meat = []
        self.veggies = []
        self.sauce = "番茄"

    def __str__(self):
        return (f"Pizza({self.size}, cheese={self.cheese}, "
                f"meat={self.meat}, veggies={self.veggies}, "
                f"sauce={self.sauce})")

class PizzaBuilder:
    def __init__(self):
        self.pizza = Pizza()

    # 链式方法(返回 self)
    def set_size(self, size: str) -> "PizzaBuilder":
        self.pizza.size = size
        return self

    def add_cheese(self) -> "PizzaBuilder":
        self.pizza.cheese = True
        return self

    def add_meat(self, meat: str) -> "PizzaBuilder":
        self.pizza.meat.append(meat)
        return self

    def add_veggie(self, veggie: str) -> "PizzaBuilder":
        self.pizza.veggies.append(veggie)
        return self

    def set_sauce(self, sauce: str) -> "PizzaBuilder":
        self.pizza.sauce = sauce
        return self

    def build(self) -> Pizza:
        # 构建前校验
        if not self.pizza.meat and not self.pizza.veggies:
            raise ValueError("披萨至少需要一种配料")
        return self.pizza

# 使用: 链式调用,可读性强
pizza = (PizzaBuilder()
         .set_size("大")
         .add_cheese()
         .add_meat("培根")
         .add_veggie("蘑菇")
         .build())
print(pizza)
# Pizza(大, cheese=True, meat=['培根'], ...)

带 Director 的版本

# Director: 预定义常用构建流程
class PizzaDirector:
    def __init__(self, builder: PizzaBuilder):
        self.builder = builder

    def build_veggie_pizza(self) -> Pizza:
        return (self.builder
                .set_size("中")
                .add_veggie("青椒")
                .add_veggie("蘑菇")
                .build())

    def build_meat_lover(self) -> Pizza:
        return (self.builder
                .set_size("大")
                .add_cheese()
                .add_meat("培根")
                .add_meat("香肠")
                .build())

director = PizzaDirector(PizzaBuilder())
print(director.build_veggie_pizza())
print(director.build_meat_lover())

与直接构造对比

# 直接构造(参数多时混乱)
# Pizza(size="大", cheese=True, meat=["培根"],
#        veggies=[], sauce="番茄", extra=True, ...)

# 建造者(步骤清晰)
# PizzaBuilder().set_size("大").add_cheese()
#   .add_meat("培根").build()

# 选择:
# 参数 ≤3 且简单 → 直接构造
# 参数多/步骤多 → 建造者
# 或 Python dataclass + 默认值

# dataclass 简化(现代推荐)
from dataclasses import dataclass, field

@dataclass
class PizzaDC:
    size: str = "中"
    cheese: bool = False
    meat: list = field(default_factory=list)
    veggies: list = field(default_factory=list)
    sauce: str = "番茄"

pizza = PizzaDC(size="大", cheese=True,
                meat=["培根"])
# 参数化构造已够用,无需 Builder

适用场景

# 1. 对象参数多且可选(配置对象/请求对象)
# 2. 构建步骤固定但表示不同(同一流程多产物)
# 3. 构建过程需要校验与默认值

# 实际案例:
# - HTTP 请求构建(URL/头/体/超时)
# - 数据库查询构建(select/where/order)
# - 邮件构建(收件人/主题/附件)
# - 测试数据工厂

# 示例: 查询构建器
class QueryBuilder:
    def __init__(self, table):
        self._table = table
        self._where = []
        self._order = None

    def where(self, cond: str) -> "QueryBuilder":
        self._where.append(cond)
        return self

    def order_by(self, field: str) -> "QueryBuilder":
        self._order = field
        return self

    def build(self) -> str:
        sql = f"SELECT * FROM {self._table}"
        if self._where:
            sql += " WHERE " + " AND ".join(self._where)
        if self._order:
            sql += f" ORDER BY {self._order}"
        return sql

sql = (QueryBuilder("users")
       .where("age > 18")
       .where("active = 1")
       .order_by("created_at")
       .build())
print(sql)
# SELECT * FROM users WHERE age > 18 AND active = 1
#   ORDER BY created_at

优缺点总结

# 优点:
# ✅ 构建过程清晰
# ✅ 分步校验
# ✅ 复用构建流程(Director)
# ✅ 链式 API 体验好

# 缺点:
# ❌ 类数量增加
# ❌ 简单对象过度设计
# ❌ 链式 API 调试稍繁琐

# 决策:
# 简单参数 → dataclass/直接构造
# 复杂多步构建 → Builder
# Python 中链式 API 很自然(推荐)

建造者把"复杂构建"变得清晰有序。掌握链式调用与校验,配置类/查询类对象的构造优雅可读。

原型模式

原型模式(Prototype Pattern)

原型:通过"克隆已有对象"来创建新对象,而非重新构建。适合创建成本高或结构复杂的对象。

意图

# 问题: 创建对象成本高(数据库/复杂初始化)
# 或需要与现有对象"基本一致,略有不同"

# 解决: 以对象为原型,克隆出副本再修改

# 优点:
# - 创建成本低(复制而非重建)
# - 避免与创建逻辑耦合
# - 运行时动态创建相似对象

# 缺点:
# - 浅拷贝陷阱(嵌套对象共享)
# - 克隆接口实现需注意

# Python 中天然支持:
# copy.copy(浅拷贝) / copy.deepcopy(深拷贝)
# 或自定义 clone() 方法

Python 示例: 深浅拷贝

import copy

class Document:
    def __init__(self, title, content, tags=None):
        self.title = title
        self.content = content
        self.tags = tags or []

    def __str__(self):
        return (f"Document({self.title}, "
                f"tags={self.tags})")

# 原型对象
template = Document("模板", "默认内容",
                    tags=["通用"])

# 浅拷贝: 新对象,但嵌套对象共享
doc1 = copy.copy(template)
doc1.title = "文档A"
doc1.tags.append("A")        # 影响 template!

print(template.tags)   # ['通用', 'A'] ← 被改了!

# 深拷贝: 完全独立
doc2 = copy.deepcopy(template)
doc2.title = "文档B"
doc2.tags.append("B")

print(template.tags)   # ['通用'](不受影响)
print(doc2.tags)       # ['通用', 'B']

自定义 clone 方法

from abc import ABC, abstractmethod
import copy

class Prototype(ABC):
    @abstractmethod
    def clone(self): ...

class Report(Prototype):
    def __init__(self, title, data, config=None):
        self.title = title
        self.data = data
        self.config = config or {}

    def clone(self, **overrides):
        """克隆并支持覆盖属性"""
        new = copy.deepcopy(self)
        for key, value in overrides.items():
            setattr(new, key, value)
        return new

# 使用
base = Report("销售报告", {"total": 1000},
              {"format": "pdf"})

monthly = base.clone(title="三月销售报告",
                     data={"total": 1500})
print(monthly.title)    # 三月销售报告
print(monthly.data)     # {'total': 1500}

# base 不受影响
print(base.title)       # 销售报告

场景示例: 游戏对象

import copy

class Enemy:
    def __init__(self, name, health, damage,
                 skills=None):
        self.name = name
        self.health = health
        self.damage = damage
        self.skills = skills or []

    def clone(self, **mods):
        new = copy.deepcopy(self)
        for k, v in mods.items():
            setattr(new, k, v)
        return new

# 原型: 精英怪模板
elite_template = Enemy("精英守卫", 500, 30,
                       skills=["重击", "格挡"])

# 快速生成关卡怪物(每个略有不同)
enemy1 = elite_template.clone(health=600)
enemy2 = elite_template.clone(name="精英射手",
                              damage=40,
                              skills=["射击"])
enemy3 = elite_template.clone(health=450, damage=25)

# 大量相似对象: 克隆远比重新构建高效
spawns = [elite_template.clone()
          for _ in range(100)]

与工厂模式的对比

# 工厂模式: 从零创建(新对象)
# 原型模式: 克隆创建(基于已有对象)

# 选择:
# - 对象结构简单、创建便宜 → 工厂
# - 对象结构复杂、创建贵 → 原型
# - 需要"相似副本" → 原型

# 实际案例:
# - 文档/报表模板复制
# - 配置模板衍生
# - 游戏实体批量生成
# - 缓存对象快照

# Python 注意事项:
# 1. 浅拷贝共享嵌套对象(陷阱)
# 2. 深拷贝可能很慢/循环引用
# 3. 自定义 __copy__/__deepcopy__ 控制行为

# 自定义深拷贝行为
class MyObj:
    def __deepcopy__(self, memo):
        # 控制深拷贝逻辑
        new = MyObj()
        new.simple_field = self.simple_field
        return new  # 不复制大对象(按需)

优缺点总结

# 优点:
# ✅ 克隆比构建快
# ✅ 动态创建相似对象
# ✅ 与具体类解耦

# 缺点:
# ❌ 深浅拷贝陷阱
# ❌ 深拷贝性能成本
# ❌ 实现 __deepcopy__ 需小心

# 决策:
# 简单复制场景 → copy 模块
# 复杂对象克隆 → 自定义 clone 方法
# 需要覆盖属性 → clone(**overrides) 模式

原型模式用"复制"代替"重建"。掌握深浅拷贝的区别与自定义 clone,复杂对象的"复制改造"游刃有余。

适配器模式

适配器模式(Adapter Pattern)

适配器:将一个类的接口转换成客户端期望的另一个接口,让原本不兼容的类可以协作(类似充电器转接头)。

意图

# 问题: 已有类接口与客户端不匹配
# - 第三方库接口不同
# - 旧系统接口与新系统不一致

# 解决: 适配器包装不兼容接口
# 客户端调用适配器(目标接口)
# 适配器内部转发给被适配者

# 优点:
# - 复用已有类,无需修改
# - 解耦客户端与具体实现
# - 符合开闭原则(加适配器即可)

# 缺点:
# - 增加一层(轻微性能开销)
# - 过度使用让代码绕

结构(对象适配器)

# 参与者:
# Target: 目标接口(客户端期望)
# Adaptee: 被适配者(已有接口)
# Adapter: 适配器(实现 Target,包装 Adaptee)

# Client → Target ← Adapter → Adaptee

Python 示例: 充电器

from abc import ABC, abstractmethod

# 目标接口(客户端期望)
class UsbC(ABC):
    @abstractmethod
    def charge(self) -> str: ...

# 被适配者(旧接口)
class UsbACharger:
    def charge_with_usb_a(self) -> str:
        return "使用 USB-A 充电"

# 适配器(把 USB-A 转成 USB-C)
class UsbCAdapter(UsbC):
    def __init__(self, charger: UsbACharger):
        self.charger = charger

    def charge(self) -> str:
        # 转发并转换
        return f"{self.charger.charge_with_usb_a()} "
               f"(通过适配器转 USB-C)"

# 客户端: 只认 UsbC 接口
def charge_phone(charger: UsbC) -> str:
    return charger.charge()

old_charger = UsbACharger()
adapter = UsbCAdapter(old_charger)

print(charge_phone(adapter))
# 使用 USB-A 充电 (通过适配器转 USB-C)

类适配器(多继承)

# Python 支持多继承,可做类适配器
class UsbCAdapter2(UsbC, UsbACharger):
    def charge(self) -> str:
        # 直接调用父类方法
        return f"{self.charge_with_usb_a()} (类适配器)"

adapter2 = UsbCAdapter2()
print(charge_phone(adapter2))
# 使用 USB-A 充电 (类适配器)

# 对象适配器(组合) vs 类适配器(继承):
# 对象适配器: 灵活,适配多个对象
# 类适配器: 简单,但继承耦合

实际场景

# 1. 第三方库接口适配
# 例: 两个支付 SDK 接口不同 → 统一适配器

# 2. 旧系统集成
# 例: 旧数据访问接口 → 新 ORM 接口

# 3. 数据格式转换
# 例: XML 数据 → JSON 接口

# 4. 日志库适配
# 例: 统一日志接口 → logging/stdout/第三方

# 示例: 统一支付接口
class PaymentGateway(ABC):
    @abstractmethod
    def pay(self, amount): ...

class AlipaySDK:
    def alipay_pay(self, amount):
        return f"支付宝支付 {amount} 元"

class WechatSDK:
    def wechat_pay(self, amount):
        return f"微信支付 {amount} 元"

class AlipayAdapter(PaymentGateway):
    def __init__(self, sdk: AlipaySDK):
        self.sdk = sdk
    def pay(self, amount):
        return self.sdk.alipay_pay(amount)

class WechatAdapter(PaymentGateway):
    def __init__(self, sdk: WechatSDK):
        self.sdk = sdk
    def pay(self, amount):
        return self.sdk.wechat_pay(amount)

# 客户端统一调用
def checkout(gateway: PaymentGateway, amount):
    return gateway.pay(amount)

print(checkout(AlipayAdapter(AlipaySDK()), 100))
print(checkout(WechatAdapter(WechatSDK()), 100))

Python 动态适配(鸭子类型)

# Python 动态特性: 无需显式实现接口
# 只要方法名一致即可(鸭子类型)

class Duck:
    def quack(self): return "嘎嘎"

class Person:
    def quack(self): return "学鸭子叫"

# 直接兼容,无需适配器
def make_quack(obj):
    return obj.quack()

print(make_quack(Duck()))
print(make_quack(Person()))

# 但接口名不一致时仍需适配器
class Goose:
    def honk(self): return "昂昂"

class GooseAdapter:
    def __init__(self, goose: Goose):
        self.goose = goose
    def quack(self):
        return f"{self.goose.honk()} (鹅适配为鸭叫)"

print(make_quack(GooseAdapter(Goose())))

优缺点总结

# 优点:
# ✅ 复用已有代码
# ✅ 客户端接口统一
# ✅ 符合开闭原则

# 缺点:
# ❌ 增加抽象层
# ❌ 适配过多时系统复杂

# 决策:
# 接口方法名一致 → 鸭子类型直接用
# 接口不匹配 → 适配器
# 新旧系统长期共存 → 适配器过渡

# 反模式警示:
# 过度适配(每个类都套适配器)会让
# 代码难以理解。优先重构统一接口,
# 适配器作为过渡方案。

适配器是"接口翻译官"。掌握对象/类两种适配方式,配合 Python 鸭子类型,新旧代码无缝协作。

桥接模式

桥接模式(Bridge Pattern)

桥接:将抽象与实现分离,让两者可以独立变化。解决"多维变化"的组合爆炸问题。

意图

# 问题: 两个维度独立变化
# 例: 形状(圆形/方形) × 颜色(红/绿/蓝)
# 继承组合: 3×3=9 个类(爆炸!)

# 解决: 抽象(形状)持有实现(颜色)引用
# 两个维度各自独立扩展

# 优点:
# - 独立扩展两个维度
# - 减少类数量
# - 符合开闭原则(每维加新值只加一个类)

# 缺点:
# - 增加抽象与理解难度
# - 需要正确识别"两个维度"

结构

# 参与者:
# Abstraction: 抽象(高层,持实现引用)
# RefinedAbstraction: 扩展抽象
# Implementor: 实现接口
# ConcreteImplementor: 具体实现

# Shape → Color(桥)
#  ├─ Circle ──┐
#  └─ Square ──┤
#               ├─ Red / Green / Blue

Python 示例: 形状与颜色

from abc import ABC, abstractmethod

# 实现维度: 颜色
class Color(ABC):
    @abstractmethod
    def apply(self) -> str: ...

class Red(Color):
    def apply(self) -> str:
        return "红色"

class Green(Color):
    def apply(self) -> str:
        return "绿色"

# 抽象维度: 形状(持有颜色)
class Shape(ABC):
    def __init__(self, color: Color):
        self.color = color

    @abstractmethod
    def draw(self) -> str: ...

class Circle(Shape):
    def draw(self) -> str:
        return f"画{self.color.apply()}圆形"

class Square(Shape):
    def draw(self) -> str:
        return f"画{self.color.apply()}方形"

# 组合使用
circle_red = Circle(Red())
square_green = Square(Green())

print(circle_red.draw())    # 画红色圆形
print(square_green.draw())  # 画绿色方形

# 新增"蓝色": 只加 Blue 类,不动形状
# 新增"三角形": 只加 Triangle 类,不动颜色
class Blue(Color):
    def apply(self) -> str:
        return "蓝色"

print(Circle(Blue()).draw())   # 画蓝色圆形

应用: 跨平台渲染

# 场景: 窗口控件 × 操作系统
# 控件(抽象): 按钮/输入框
# 平台(实现): Windows/Linux/macOS

class Platform(ABC):
    @abstractmethod
    def render(self, widget) -> str: ...

class Windows(Platform):
    def render(self, widget) -> str:
        return f"Windows 渲染: {widget}"

class Linux(Platform):
    def render(self, widget) -> str:
        return f"Linux 渲染: {widget}"

class Widget(ABC):
    def __init__(self, platform: Platform):
        self.platform = platform
    @abstractmethod
    def draw(self) -> str: ...

class Button(Widget):
    def draw(self) -> str:
        return self.platform.render("按钮")

class Input(Widget):
    def draw(self) -> str:
        return self.platform.render("输入框")

# 组合: 2 控件 × 2 平台 = 4 种组合
# 无需 4 个类!
print(Button(Windows()).draw())
print(Input(Linux()).draw())

# 新增平台只加一个类
class Mac(Platform):
    def render(self, widget) -> str:
        return f"macOS 渲染: {widget}"

print(Button(Mac()).draw())   # macOS 渲染: 按钮

与适配器对比

# 适配器(Adapter):
# 让"已有接口"适配"期望接口"
# 侧重兼容(解决不匹配)

# 桥接(Bridge):
# 分离"抽象与实现"两个维度
# 侧重解耦(独立变化)

# 对比:
# 适配器: 事后补救(转换接口)
# 桥接: 事前设计(分离维度)

# 选择:
# 接口不兼容 → 适配器
# 多维变化 → 桥接
# 两维度独立扩展 → 桥接

Python 简化

# 动态语言中"抽象接口"可弱化
# 直接鸭子类型即可

class Shape2:
    def __init__(self, color):  # 任何有 apply 的对象
        self.color = color

    def draw(self):
        return f"画{self.color.apply()}图形"

# 只要 color 有 apply() 方法即可
print(Shape2(Red()).draw())

# 接口也可用 Protocol(类型标注)
from typing import Protocol

class ColorProtocol(Protocol):
    def apply(self) -> str: ...

def draw_shape(shape, color: ColorProtocol):
    return f"画{color.apply()}{shape}"

优缺点总结

# 优点:
# ✅ 两维度独立扩展(消灭组合爆炸)
# ✅ 减少类数量
# ✅ 开闭原则

# 缺点:
# ❌ 抽象理解成本
# ❌ 需正确识别变化维度

# 决策信号:
# - 组合类爆炸(N×M 类)
# - 两个维度会独立变化
# - 需要运行时切换实现

# 反模式:
# 只有一个变化维度 → 不要用桥接
# 简单场景直接组合即可

桥接是"多维变化"的解药。识别两个独立维度,用组合替代继承,组合爆炸迎刃而解。

过滤器模式

过滤器模式(Filter/Criteria Pattern)

过滤器:用标准(条件)对对象集合进行筛选,支持多个条件自由组合(与/或/非),避免用大量 if-else。

意图

# 问题: 按多个条件筛选集合
# - 条件组合爆炸(if-else 嵌套)
# - 条件复用困难

# 解决: 每个条件是一个"过滤器"
# 过滤器可组合(and/or/not)
# 客户端自由拼装筛选标准

# 优点:
# - 条件独立可复用
# - 组合灵活(标准模式)
# - 符合开闭原则(加条件=加过滤器)

# 缺点:
# - 类数量增加(小场景过度设计)

结构

# 参与者:
# Criteria: 过滤接口(meet(items))
# ConcreteCriteria: 具体条件
# AndCriteria / OrCriteria / NotCriteria: 组合

# Client: 拼装条件并过滤

Python 示例: 人员筛选

from abc import ABC, abstractmethod
from dataclasses import dataclass, field

@dataclass
class Person:
    name: str
    gender: str
    age: int
    married: bool = False

# 过滤接口
class Criteria(ABC):
    @abstractmethod
    def meet(self, items: list[Person]) -> list[Person]:
        ...

# 具体条件
class MaleCriteria(Criteria):
    def meet(self, items):
        return [p for p in items if p.gender == "男"]

class AgeCriteria(Criteria):
    def __init__(self, min_age, max_age=200):
        self.min_age = min_age
        self.max_age = max_age

    def meet(self, items):
        return [p for p in items
                if self.min_age <= p.age <= self.max_age]

# 组合条件
class AndCriteria(Criteria):
    def __init__(self, *criterias):
        self.criterias = criterias

    def meet(self, items):
        result = items
        for c in self.criterias:
            result = c.meet(result)
        return result

class OrCriteria(Criteria):
    def __init__(self, *criterias):
        self.criterias = criterias

    def meet(self, items):
        result = []
        for c in self.criterias:
            for item in c.meet(items):
                if item not in result:
                    result.append(item)
        return result

class NotCriteria(Criteria):
    def __init__(self, criteria):
        self.criteria = criteria

    def meet(self, items):
        excluded = set(self.criteria.meet(items))
        return [p for p in items if p not in excluded]

使用

people = [
    Person("张三", "男", 25),
    Person("李四", "女", 30, married=True),
    Person("王五", "男", 40, married=True),
    Person("赵六", "女", 22),
    Person("钱七", "男", 18),
]

male = MaleCriteria()
adult = AgeCriteria(20)
youth = AgeCriteria(18, 25)

# 组合使用
print("成年男性:")
for p in AndCriteria(male, adult).meet(people):
    print(" ", p.name)

print("男性或年轻人:")
for p in OrCriteria(male, youth).meet(people):
    print(" ", p.name)

print("非男性:")
for p in NotCriteria(male).meet(people):
    print(" ", p.name)

print("25-40 岁的男性:")
result = AndCriteria(
    male, AgeCriteria(25, 40)).meet(people)
print(" ", [p.name for p in result])

Python 简化: 函数过滤器

# Python 中函数式过滤更自然
from operator import and_, or_

def male(p): return p.gender == "男"
def adult(p): return p.age >= 20
def young(p): return p.age <= 25

# 直接组合(高阶函数)
result = [p for p in people
          if male(p) and adult(p)]

# 或用 all/any
def match_all(p, *preds):
    return all(pred(p) for pred in preds)

def match_any(p, *preds):
    return any(pred(p) for pred in preds)

# 使用
result = [p for p in people
          if match_all(p, male, adult)]
result2 = [p for p in people
           if match_any(p, male, young)]

# 或 functools 组合
import functools, operator
result3 = [p for p in people
           if functools.reduce(and_,
               [male(p), adult(p)])]

# 复杂组合才用类(标准)方式
# 简单条件用函数/Python 内置更清晰

适用场景

# 1. 多条件筛选(搜索/列表过滤)
# 2. 规则引擎(条件组合匹配)
# 3. 数据校验(多个校验规则组合)
# 4. 权限检查(多个条件)
# 5. 报表过滤(多维度筛选)

# 实际案例:
# - 商品筛选(价格/分类/品牌/评价)
# - 用户搜索(地区/年龄/活跃度)
# - 日志过滤(级别/来源/关键词)

# 与"职责链"区分:
# 过滤器: 筛选集合(输出子集)
# 责任链: 处理请求(依次处理)

# 与"策略"区分:
# 过滤器: 组合筛选条件
# 策略: 替换整个算法

优缺点总结

# 优点:
# ✅ 条件独立可复用
# ✅ 自由组合(与/或/非)
# ✅ 开闭原则
# ✅ 清晰可读

# 缺点:
# ❌ 类数量增加
# ❌ 简单过滤过度设计

# 决策:
# 简单条件 → 推导式/Python 内置
# 复杂组合且需复用 → 过滤器模式
# 动态组合规则 → 过滤器类(可序列化)

# 最佳实践:
# - 先函数式,再按需升级为类
# - 组合逻辑清晰命名(And/Or/Not)
# - 过滤条件保持纯函数(易测试)

过滤器把"条件组合"结构化。掌握与/或/非组合,配合 Python 函数式风格,筛选逻辑灵活清晰。

组合模式

组合模式(Composite Pattern)

组合:将对象组织成树形结构,让"部分-整体"的处理一致。客户端无需区分叶子与分支,统一调用。

意图

# 问题: 树形结构(目录/菜单/组织架构)
# - 处理叶子与容器逻辑不同
# - 客户端需判断类型

# 解决: 统一接口(叶子与容器都实现)
# 容器持有子节点,操作递归传播

# 优点:
# - 客户端统一处理(无需 if 判断)
# - 新增节点类型容易
# - 树形遍历自然递归

# 缺点:
# - 接口过度统一(叶子实现无关方法)
# - 可能破坏单一职责

结构

# 参与者:
# Component: 组件接口(叶子与容器共有)
# Leaf: 叶子(无子节点)
# Composite: 容器(含子节点列表)

# 例: 文件系统
# Folder(Composite) → File(Leaf)
# 删除文件夹 = 递归删除所有内容

Python 示例: 文件系统

from abc import ABC, abstractmethod

# 组件接口
class FileSystemNode(ABC):
    @abstractmethod
    def get_size(self) -> int: ...
    def display(self, indent: str = ""): ...

# 叶子: 文件
class File(FileSystemNode):
    def __init__(self, name, size):
        self.name = name
        self.size = size

    def get_size(self) -> int:
        return self.size

    def display(self, indent=""):
        print(f"{indent}📄 {self.name} ({self.size}B)")

# 容器: 文件夹
class Folder(FileSystemNode):
    def __init__(self, name):
        self.name = name
        self.children = []

    def add(self, node: FileSystemNode):
        self.children.append(node)
        return self

    def remove(self, node: FileSystemNode):
        self.children.remove(node)

    def get_size(self) -> int:
        # 递归求和(统一接口!)
        return sum(c.get_size() for c in self.children)

    def display(self, indent=""):
        print(f"{indent}📁 {self.name}/")
        for child in self.children:
            child.display(indent + "  ")

使用

# 构建目录树
root = Folder("root")

docs = Folder("docs")
docs.add(File("readme.md", 100))
docs.add(File("guide.md", 250))

src = Folder("src")
src.add(File("main.py", 800))
src.add(File("utils.py", 400))

tests = Folder("tests")
tests.add(File("test_main.py", 300))
src.add(tests)          # 嵌套

root.add(docs)
root.add(src)

# 统一接口: 递归显示
root.display()
# 📁 root/
#   📁 docs/
#     📄 readme.md (100B)
#     📄 guide.md (250B)
#   📁 src/
#     📄 main.py (800B)
#     📄 utils.py (400B)
#     📁 tests/
#       📄 test_main.py (300B)

# 统一获取大小(递归)
print(f"总大小: {root.get_size()}B")
# 总大小: 1850B

# 客户端无需判断类型!
def total_size(node: FileSystemNode):
    return node.get_size()

print(total_size(src))    # 1500
print(total_size(docs))   # 350

应用: 菜单系统

# 菜单项(叶子)与子菜单(容器)
class MenuItem(FileSystemNode):  # 复用结构
    ...

# 实际应用:
# - 菜单/导航树
# - 组织架构树
# - 表达式树
# - UI 组件树(容器与控件统一)
# - HTML DOM 树

# UI 组件示例
class UIComponent(ABC):
    @abstractmethod
    def render(self): ...

class Button(UIComponent):
    def render(self):
        return "按钮"

class Panel(UIComponent):
    def __init__(self):
        self.components = []

    def add(self, comp):
        self.components.append(comp)

    def render(self):
        inner = "".join(c.render()
                        for c in self.components)
        return f"
{inner}
"

page = Panel()
page.add(Button())
page.add(Panel().add(Button()))
print(page.render())
# 
按钮
按钮

安全与透明

# 透明方式: 接口统一(叶子也有 add 方法)
# - 客户端无感知,但叶子 add 应报错

# 安全方式: 只有容器有 add 方法
# - 客户端需判断类型(违背组合意图)

# Python 中权衡:
# 接口统一(get_size 等)是核心
# add/remove 可只在容器实现(鸭子类型)

# 避免循环引用:
# 添加子节点时检查是否自身
def add(self, node):
    if node is self:
        raise ValueError("不能添加自身")
    self.children.append(node)

优缺点总结

# 优点:
# ✅ 统一处理部分-整体
# ✅ 树形递归遍历
# ✅ 新增节点类型易

# 缺点:
# ❌ 接口过度统一
# ❌ 可能隐藏类型差异

# 决策:
# 树形结构且需统一操作 → 组合
# 简单层级 → 直接用列表嵌套

# Python 简化:
# 用 dict/list 嵌套即可表达树
tree = {
    "name": "root",
    "children": [
        {"name": "file1", "size": 100},
        {"name": "dir", "children": [...]},
    ],
}

# 递归函数处理(函数式替代)
def calc_size(node):
    return node.get("size", 0) + sum(
        calc_size(c)
        for c in node.get("children", []))

组合模式让"树形结构"处理统一。掌握递归接口设计,目录/菜单/组件树类场景自然优雅。

装饰器模式

装饰器模式(Decorator Pattern)

装饰器:动态地给对象添加职责,比继承更灵活。Python 的 @decorator 语法就是这一思想的语言级实现。

意图

# 问题: 需要动态添加功能
# - 继承方式: 功能组合爆炸(加糖+加冰+大杯...)
# - 修改原类: 违反开闭原则

# 解决: 包装器(装饰器)层层包裹
# 每个装饰器实现同一接口,内部调用被包装对象

# 优点:
# - 动态组合功能
# - 替代继承(组合优于继承)
# - 符合开闭原则

# 缺点:
# - 多层级包装调试难
# - 对象标识变化(== 比较可能失效)

结构

# 参与者:
# Component: 组件接口
# ConcreteComponent: 具体组件(被装饰)
# Decorator: 装饰器(持组件引用,实现接口)
# ConcreteDecorator: 具体装饰器(添加职责)

# Coffee → MilkDecorator → SugarDecorator
# 层层包裹,叠加功能

Python 示例: 咖啡订单

from abc import ABC, abstractmethod

# 组件接口
class Beverage(ABC):
    @abstractmethod
    def cost(self) -> float: ...
    @abstractmethod
    def description(self) -> str: ...

# 具体组件
class Espresso(Beverage):
    def cost(self) -> float:
        return 15.0

    def description(self) -> str:
        return "浓缩咖啡"

# 装饰器基类
class CondimentDecorator(Beverage):
    def __init__(self, beverage: Beverage):
        self.beverage = beverage

# 具体装饰器
class Milk(CondimentDecorator):
    def cost(self) -> float:
        return self.beverage.cost() + 3.0

    def description(self) -> str:
        return self.beverage.description() + ", 加奶"

class Sugar(CondimentDecorator):
    def cost(self) -> float:
        return self.beverage.cost() + 1.5

    def description(self) -> str:
        return self.beverage.description() + ", 加糖"

class Whip(CondimentDecorator):
    def cost(self) -> float:
        return self.beverage.cost() + 5.0

    def description(self) -> str:
        return self.beverage.description() + ", 加奶油"

使用: 动态组合

# 基础咖啡
coffee = Espresso()
print(coffee.description(), coffee.cost())
# 浓缩咖啡 15.0

# 加奶
coffee = Milk(coffee)
print(coffee.description(), coffee.cost())
# 浓缩咖啡, 加奶 18.0

# 再加糖和奶油
coffee = Sugar(Whip(coffee))
print(coffee.description(), coffee.cost())
# 浓缩咖啡, 加奶, 加奶油, 加糖 24.5

# 不同的组合随意搭配
another = Sugar(Espresso())   # 只要糖
print(another.description(), another.cost())
# 浓缩咖啡, 加糖 16.5

# 新增"豆浆": 只需一个类,无需修改咖啡类
class Soy(CondimentDecorator):
    def cost(self) -> float:
        return self.beverage.cost() + 2.0
    def description(self) -> str:
        return self.beverage.description() + ", 加豆浆"

Python 装饰器语法

# Python 语言级装饰器(函数装饰)
import time
from functools import wraps

def timer(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        print(f"{func.__name__} 耗时 {elapsed:.4f}s")
        return result
    return wrapper

@timer          # 等价 slow = timer(slow)
def slow():
    time.sleep(0.1)

slow()
# slow 耗时 0.1001s

# 多层装饰(叠加功能)
def log(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        print(f"调用 {func.__name__}")
        return func(*args, **kwargs)
    return wrapper

@timer
@log
def process(x):
    return x * 2

process(5)
# 调用 process
# process 耗时 0.0000s

实际应用

# 1. IO 流包装
# - BufferedInputStream → FileInputStream
# - gzip 压缩流 → 文件流

# 2. 中间件/拦截器(Web)
# - 认证 → 日志 → 缓存 → 业务处理

# 3. 函数增强(Python 装饰器)
# - @staticmethod / @classmethod
# - @property
# - @lru_cache(缓存)
from functools import lru_cache

@lru_cache(maxsize=128)
def fib(n):
    return n if n < 2 else fib(n-1) + fib(n-2)

# 4. 数据增强管道
# - 压缩 → 加密 → 序列化

# 5. 权限/校验包装

与代理模式对比

# 装饰器(Decorator):
# 添加新功能(增强)
# 接口不变,功能叠加

# 代理(Proxy):
# 控制访问(延迟/权限/日志)
# 关注访问控制,非增强

# 对比:
# 装饰器: 客户端直接构造包装链
# 代理: 客户端通常不知代理存在

# 选择:
# 动态加功能 → 装饰器
# 控制访问 → 代理

优缺点总结

# 优点:
# ✅ 动态组合功能
# ✅ 优于继承(组合)
# ✅ 开闭原则
# ✅ Python 语法原生支持

# 缺点:
# ❌ 多层包装调试困难
# ❌ 包装顺序影响行为(需注意)
# ❌ 对象比较/类型判断变复杂

# 最佳实践:
# - 装饰器保持"透明"(接口不变)
# - 单个装饰器职责单一
# - 注意装饰顺序(从内到外执行)
# - 用 functools.wraps 保留元数据

# Python 建议:
# 类装饰 → 继承 Decorator 基类
# 函数增强 → @decorator 语法
# 简单场景 → 手动包装函数

装饰器是"动态增强"的利器,Python 语法原生支持。掌握类装饰与 @decorator,功能叠加灵活轻量。

外观模式

外观模式(Facade Pattern)

外观:为复杂的子系统提供一个统一的高层接口,让客户端简单使用。相当于"前台/门面"。

意图

# 问题: 子系统接口复杂(多个类互相协作)
# - 客户端需要了解内部细节
# - 调用步骤繁琐易错

# 解决: 外观类封装复杂流程
# 客户端只调用外观的简单方法

# 优点:
# - 简化客户端(降低使用门槛)
# - 解耦客户端与子系统
# - 隐藏内部细节

# 缺点:
# - 外观可能成为"上帝对象"
# - 过度封装损失灵活性
# - 外观与子系统耦合

结构

# 参与者:
# Facade: 外观(提供简化接口)
# Subsystems: 子系统(复杂内部逻辑)

# Client → Facade → 各 Subsystem
# 客户端不直接碰子系统

Python 示例: 家庭影院

# 子系统
class Projector:
    def on(self): return "投影仪开机"
    def off(self): return "投影仪关机"
    def set_input(self, src): return f"切换到 {src}"

class SoundSystem:
    def on(self): return "音响开机"
    def set_volume(self, v): return f"音量调到 {v}"
    def off(self): return "音响关机"

class DVDPlayer:
    def play(self, movie): return f"播放 {movie}"
    def stop(self): return "停止播放"
    def eject(self): return "弹出光盘"

class Lights:
    def dim(self, level): return f"灯光调暗到 {level}%"
    def bright(self): return "灯光调亮"

# 外观
class HomeTheaterFacade:
    def __init__(self):
        self.projector = Projector()
        self.sound = SoundSystem()
        self.dvd = DVDPlayer()
        self.lights = Lights()

    def watch_movie(self, movie: str) -> list[str]:
        """一键看电影(封装复杂流程)"""
        steps = []
        steps.append(self.lights.dim(30))
        steps.append(self.projector.on())
        steps.append(self.projector.set_input("DVD"))
        steps.append(self.sound.on())
        steps.append(self.sound.set_volume(50))
        steps.append(self.dvd.play(movie))
        return steps

    def end_movie(self) -> list[str]:
        steps = []
        steps.append(self.dvd.stop())
        steps.append(self.dvd.eject())
        steps.append(self.projector.off())
        steps.append(self.sound.off())
        steps.append(self.lights.bright())
        return steps

# 客户端: 只需两步
theater = HomeTheaterFacade()
for step in theater.watch_movie("星际穿越"):
    print(step)
# 灯光调暗到 30%
# 投影仪开机
# 切换到 DVD
# 音响开机
# 音量调到 50
# 播放 星际穿越

for step in theater.end_movie():
    print(step)

应用: 数据访问门面

# 场景: 订单系统(用户+库存+支付+通知)
class UserService:
    def check(self, uid):
        return f"校验用户 {uid}"

class InventoryService:
    def deduct(self, pid, qty):
        return f"扣减库存 {pid} x{qty}"

class PaymentService:
    def charge(self, uid, amount):
        return f"扣款 {amount}"

class NotifyService:
    def send(self, uid, msg):
        return f"通知 {uid}: {msg}"

# 外观: 下单门面
class OrderFacade:
    def __init__(self):
        self.users = UserService()
        self.inventory = InventoryService()
        self.payment = PaymentService()
        self.notify = NotifyService()

    def place_order(self, uid, pid, qty, amount):
        """下单完整流程(封装)"""
        steps = []
        steps.append(self.users.check(uid))
        steps.append(self.inventory.deduct(pid, qty))
        steps.append(self.payment.charge(uid, amount))
        steps.append(self.notify.send(
            uid, f"订单成功,金额 {amount}"))
        return steps

# 客户端一行调用
facade = OrderFacade()
print(facade.place_order(1, "P001", 2, 199.9))

# 好处: 客户端不关心内部服务如何协作
# 修改内部流程不影响客户端

与适配器/门面对比

# 适配器(Adapter): 转换接口(1:1)
# 门面(Facade): 简化多个接口(1:N)

# 对比:
# 适配器: 让接口匹配(兼容)
# 门面: 让接口简单(封装)

# 应用场景区分:
# 接口不匹配 → 适配器
# 流程复杂 → 门面

# 门面 vs 中间件:
# 门面: 客户端主动调用
# 中间件: 请求自动经过

# Python 中门面常表现为:
# - 服务类的统一入口
# - 高层 API 封装
# - __init__.py 的导出接口

优缺点总结

# 优点:
# ✅ 简化客户端使用
# ✅ 解耦(客户端与子系统)
# ✅ 隐藏内部细节
# ✅ 降低学习成本

# 缺点:
# ❌ 外观可能臃肿
# ❌ 客户端失去底层灵活性
# ❌ 与子系统耦合

# 最佳实践:
# - 外观保持"薄"(只做编排)
# - 提供多个粒度的外观方法
# - 不把业务逻辑塞进外观
# - 必要时暴露子系统(高级用户)

# 决策:
# 子系统复杂且客户端频繁使用 → 门面
# 简单的几个类 → 不需要门面

外观是"复杂系统的前台"。掌握封装编排与粒度控制,复杂流程对客户端始终简单清晰。

享元模式

享元模式(Flyweight Pattern)

享元:共享对象以减少内存占用。将对象的"内在状态"(可共享)与"外在状态"(随场景变化)分离。

意图

# 问题: 大量相似对象占用内存
# 例: 文档中 10 万个字符对象

# 解决: 共享"可复用的部分"(享元)
# 每个对象只保存"独有部分"

# 内在状态: 可共享(字符的形状)
# 外在状态: 场景特有(字符的位置)

# 优点:
# - 大幅减少内存
# - 对象复用

# 缺点:
# - 复杂度增加
# - 需要区分内在外在状态
# - 线程安全需注意

结构

# 参与者:
# Flyweight: 享元接口
# ConcreteFlyweight: 具体享元(内在状态)
# FlyweightFactory: 享元工厂(缓存复用)
# Client: 使用享元(管理外在状态)

# 例: 字符渲染
# 字符 glyph(内在) + 位置/颜色(外在)

Python 示例: 字符渲染

class Glyph:   # 享元(字符形状,内在状态)
    def __init__(self, char: str, font: str):
        self.char = char
        self.font = font

    def render(self, x, y, color):
        # 外在状态(x, y, color)由调用方传入
        return (f"[{self.char}/{self.font} "
                f"在({x},{y}) 颜色{color}]")

# 享元工厂: 缓存复用
class GlyphFactory:
    def __init__(self):
        self._pool = {}

    def get_glyph(self, char: str,
                  font: str = "default") -> Glyph:
        key = (char, font)
        if key not in self._pool:
            self._pool[key] = Glyph(char, font)
        return self._pool[key]

    @property
    def pool_size(self):
        return len(self._pool)

# 使用
factory = GlyphFactory()
text = "Hello World Hello World"

positions = []
for i, ch in enumerate(text):
    glyph = factory.get_glyph(ch)
    positions.append(glyph.render(
        i % 10, i // 10, "black"))

print(f"字符数: {len(text)}")      # 22
print(f"享元数: {factory.pool_size}")  # 8(去重后)
# 只有 8 个唯一字符对象被创建!

应用: 粒子系统/地图标记

# 游戏粒子系统(大量粒子)
class ParticleType:   # 享元: 粒子外观(纹理等)
    def __init__(self, color, texture):
        self.color = color
        self.texture = texture

class ParticleFactory:
    _types = {}

    @classmethod
    def get_type(cls, color, texture):
        key = (color, texture)
        if key not in cls._types:
            cls._types[key] = ParticleType(
                color, texture)
        return cls._types[key]

# 单个粒子(外在状态多)
class Particle:
    def __init__(self, x, y, speed,
                 ptype: ParticleType):
        self.x, self.y = x, y
        self.speed = speed
        self.type = ptype   # 共享类型

# 1 万粒子,但只有几种类型
factory = ParticleFactory
particles = []
for i in range(10000):
    ptype = factory.get_type(
        "red" if i % 2 else "blue", "circle")
    particles.append(Particle(
        i % 100, i % 100, i % 10, ptype))

print(f"粒子数: {len(particles)}")
print(f"类型数: {len(ParticleFactory._types)}")
# 类型数: 2(共享,内存大幅节省)

# 其他应用:
# - 文本编辑器字符对象
# - 地图城市标记(共享图标)
# - 数据库连接池(复用连接)
# - 字符串驻留(interning)

与单例/对象池对比

# 单例(Singleton): 全局一个实例
# 享元(Flyweight): 一组可共享实例
# 对象池(Pool): 复用创建成本高的对象

# 对比:
# 享元: 按"内在状态"区分共享
# 对象池: 临时借用与归还(连接池)

# 选择:
# 对象重复且内在状态可共享 → 享元
# 对象创建贵但需独占使用 → 对象池

# Python 内置享元思想:
# - 小整数驻留(-5~256 缓存)
# - 字符串驻留(短字符串复用)
# - lru_cache(函数结果复用)

优缺点总结

# 优点:
# ✅ 大幅减少内存
# ✅ 提升性能(少创建)
# ✅ 复用对象

# 缺点:
# ❌ 复杂度增加(内外状态分离)
# ❌ 状态管理需谨慎
# ❌ 共享对象修改风险

# 最佳实践:
# - 内在状态只读(不可变)
# - 工厂集中管理共享对象
# - 外在状态不存享元内
# - 大数据量场景评估收益

# 决策:
# 对象数量大且重复 → 享元
# 数量小 → 直接创建
# Python: 优先用不可变数据结构
# (str/int/tuple 天然共享)

享元用"共享"对抗"重复"。掌握内外状态分离与工厂缓存,海量相似对象的内存压力迎刃而解。

代理模式

代理模式(Proxy Pattern)

代理:为对象提供一个替身(代理)来控制对它的访问。可在访问前后增加逻辑(权限/缓存/日志/延迟加载)。

意图

# 问题: 直接访问对象不安全/低效/不便
# - 大对象不想立即加载
# - 需要权限控制
# - 需要记录访问日志

# 解决: 代理实现同一接口,控制真实对象的访问

# 优点:
# - 控制访问(权限/日志)
# - 延迟加载(虚拟代理)
# - 远程访问(远程代理)
# - 符合开闭原则

# 缺点:
# - 增加一层
# - 响应可能延迟(代理逻辑)

类型

# 虚拟代理: 延迟创建昂贵对象
# 保护代理: 控制访问权限
# 缓存代理: 缓存结果
# 远程代理: 本地替身访问远程
# 日志代理: 记录访问

# 结构与装饰器相似
# 区别: 代理控制访问,装饰器增加功能

Python 示例: 虚拟代理(延迟加载)

from abc import ABC, abstractmethod

# 主题接口
class Image(ABC):
    @abstractmethod
    def display(self) -> str: ...

# 真实对象(创建昂贵)
class RealImage(Image):
    def __init__(self, filename):
        self.filename = filename
        self._load_from_disk()   # 昂贵操作

    def _load_from_disk(self):
        print(f"从磁盘加载: {self.filename}")

    def display(self) -> str:
        return f"显示图片: {self.filename}"

# 虚拟代理: 延迟到真正需要时创建
class ImageProxy(Image):
    def __init__(self, filename):
        self.filename = filename
        self._real = None      # 尚未加载

    def display(self) -> str:
        if self._real is None:
            self._real = RealImage(self.filename)
        return self._real.display()

# 客户端: 与使用 RealImage 一样
proxy = ImageProxy("photo.jpg")
# (此时未加载)

print(proxy.display())   # 首次: 触发加载
# 从磁盘加载: photo.jpg
# 显示图片: photo.jpg

print(proxy.display())   # 再次: 复用
# 显示图片: photo.jpg

保护代理(权限控制)

class Document:
    def __init__(self, content):
        self.content = content

    def read(self) -> str:
        return self.content

    def edit(self, new_content):
        self.content = new_content
        return "已编辑"

# 保护代理
class SecureDocumentProxy:
    def __init__(self, doc: Document,
                 role: str):
        self.doc = doc
        self.role = role

    def read(self) -> str:
        # 任何人可读
        return self.doc.read()

    def edit(self, new_content):
        # 只有管理员可编辑
        if self.role != "admin":
            raise PermissionError("无编辑权限")
        return self.doc.edit(new_content)

doc = Document("机密文档")
user_proxy = SecureDocumentProxy(doc, "user")
admin_proxy = SecureDocumentProxy(doc, "admin")

print(user_proxy.read())      # 机密文档
# user_proxy.edit("x")       # PermissionError!

print(admin_proxy.edit("新内容"))  # 已编辑

缓存代理

class ExpensiveService:
    def fetch(self, key):
        # 模拟昂贵查询
        print(f"查询: {key}")
        return {"key": key, "data": "..."}

# 缓存代理
class CachedProxy:
    def __init__(self, service):
        self.service = service
        self.cache = {}

    def fetch(self, key):
        if key not in self.cache:
            self.cache[key] = self.service.fetch(key)
        return self.cache[key]

service = ExpensiveService()
proxy = CachedProxy(service)

print(proxy.fetch("a"))   # 查询: a(实际查询)
print(proxy.fetch("a"))   # (缓存命中,不查询)
print(proxy.fetch("b"))   # 查询: b

# 注意: 过期策略/一致性
# Python 中 @lru_cache 即缓存代理
from functools import lru_cache

@lru_cache(maxsize=128)
def fetch(key):
    print(f"查询: {key}")
    return {"key": key}

Python 中的代理简化

# 函数代理(装饰器实现)
def cached(func):
    cache = {}
    def wrapper(*args):
        if args not in cache:
            cache[args] = func(*args)
        return cache[args]
    return wrapper

@cached
def heavy(n):
    print(f"计算: {n}")
    return n * n

# 属性访问代理
class LazyAttribute:
    def __get__(self, obj, owner):
        # 延迟计算的描述符(虚拟代理)
        print("计算属性值...")
        value = expensive_compute()
        obj.__dict__[self.name] = value
        return value

# 远程代理示例:
# requests 库/ORM 查询(惰性)
# 数据库 QuerySet 延迟求值

优缺点总结

# 优点:
# ✅ 控制访问(权限/日志)
# ✅ 延迟加载(性能)
# ✅ 缓存(性能)
# ✅ 解耦(客户端与真实对象)

# 缺点:
# ❌ 增加一层
# ❌ 代理逻辑复杂化
# ❌ 需要维护代理与真实的同步

# 决策:
# 延迟加载 → 虚拟代理
# 权限控制 → 保护代理
# 缓存 → 缓存代理/lru_cache
# 远程调用 → 远程代理(如 RPC 客户端)

# 反模式:
# 代理与真实对象不一致
# 无实际需求的代理(徒增复杂度)

代理是"对象的守门人"。掌握虚拟/保护/缓存代理,延迟加载、权限控制、性能优化一网打尽。

责任链模式

责任链模式(Chain of Responsibility)

责任链:将请求沿着处理者链传递,每个处理者决定处理或传给下一个。解耦"发送者"与"接收者"。

意图

# 问题: 请求处理逻辑分散且可能变化
# - 硬编码 if-else 判断处理者
# - 处理顺序/处理者变更困难

# 解决: 处理者链,请求依次传递
# 每个处理者: 能处理就处理,否则传给下一个

# 优点:
# - 解耦发送者与接收者
# - 动态调整链(顺序/增删)
# - 符合开闭原则

# 缺点:
# - 请求可能未被处理(需兜底)
# - 调试链式传递较难
# - 性能(长链开销)

结构

# 参与者:
# Handler: 处理者接口(设 next + handle)
# ConcreteHandler: 具体处理者
# Client: 构建链并发送请求

# 例: 审批流程
# 员工请假 → 组长 → 经理 → 总监 → 拒绝

Python 示例: 审批流程

from abc import ABC, abstractmethod

class Approver(ABC):
    def __init__(self):
        self.next = None

    def set_next(self, approver):
        self.next = approver
        return approver      # 链式设置

    @abstractmethod
    def approve(self, amount: float) -> str: ...

class TeamLeader(Approver):
    def approve(self, amount):
        if amount <= 1000:
            return f"组长批准: {amount} 元"
        return (f"组长无权,转交"
                f"{self.next.approve(amount)}")

class Manager(Approver):
    def approve(self, amount):
        if amount <= 5000:
            return f"经理批准: {amount} 元"
        return (f"经理无权,转交"
                f"{self.next.approve(amount)}")

class Director(Approver):
    def approve(self, amount):
        if amount <= 20000:
            return f"总监批准: {amount} 元"
        return f"超出权限,拒绝: {amount} 元"

# 构建链
leader = TeamLeader()
manager = Manager()
director = Director()
leader.set_next(manager).set_next(director)

# 发送请求(客户端只认识链头)
print(leader.approve(500))
# 组长批准: 500 元

print(leader.approve(3000))
# 组长无权,转交经理批准: 3000 元

print(leader.approve(8000))
# 组长无权,转交经理无权,转交总监批准: 8000 元

print(leader.approve(50000))
# ... 拒绝: 50000 元

# 动态调整链(加角色/改顺序)
class Boss(Approver):
    def approve(self, amount):
        return f"老板批准: {amount} 元"

leader.set_next(manager).set_next(
    director).set_next(Boss())
print(leader.approve(50000))
# ... 老板批准: 50000 元

应用: 中间件/过滤器链

# Web 中间件就是责任链!
# 请求 → 认证 → 日志 → 限流 → 业务

class Middleware:
    def __init__(self):
        self.next = None

    def set_next(self, mw):
        self.next = mw
        return mw

    def handle(self, request):
        if self.next:
            return self.next.handle(request)
        return "处理完成"

class AuthMiddleware(Middleware):
    def handle(self, request):
        if not request.get("token"):
            return "401 未认证"
        return super().handle(request)

class LoggerMiddleware(Middleware):
    def handle(self, request):
        print(f"日志: {request.get('path')}")
        return super().handle(request)

# 构建中间件链
chain = AuthMiddleware()
chain.set_next(LoggerMiddleware())

print(chain.handle({"path": "/api", "token": None}))
# 401 未认证

print(chain.handle({"path": "/api", "token": "abc"}))
# 日志: /api
# 处理完成

# 框架中的责任链:
# - Django 中间件(MIDDLEWARE 列表)
# - Express/Koa 中间件
# - 表单校验链
# - 事件冒泡(UI)

Python 简化

# 函数列表式责任链
handlers: list = []

def add_handler(fn):
    handlers.append(fn)
    return fn

@add_handler
def check_auth(req):
    if not req.get("token"):
        return "401"
    return None   # 不处理,继续

@add_handler
def check_rate(req):
    if req.get("burst"):
        return "429"
    return None

def process(req):
    for handler in handlers:
        result = handler(req)
        if result:
            return result
    return "OK"

print(process({"token": None}))        # 401
print(process({"token": "x"}))         # OK
print(process({"token": "x", "burst": True}))  # 429

# 或使用哨兵/异常控制流(谨慎)

优缺点总结

# 优点:
# ✅ 解耦发送与处理
# ✅ 动态调整链
# ✅ 开闭原则(加处理者即可)
# ✅ 职责单一(每个处理者一个职责)

# 缺点:
# ❌ 请求可能无人处理
# ❌ 调试困难(链式)
# ❌ 长链性能开销

# 最佳实践:
# - 链尾加兜底处理者
# - 处理者职责单一、顺序清晰
# - 控制链长度
# - 日志记录传递过程(调试)

# 决策:
# 多个条件依次处理 → 责任链
# 组合条件筛选 → 过滤器
# 简单 if-else → 不需要模式

责任链让"请求处理"灵活传递。掌握链式构建与兜底,审批流、中间件、校验链信手拈来。

命令模式

命令模式(Command Pattern)

命令:将请求封装为对象,从而支持参数化、排队、记录日志与撤销操作。

意图

# 问题: 请求直接绑定到执行者
# - 难以撤销/重做
# - 难以排队/记录
# - 发送者与接收者耦合

# 解决: 请求封装为命令对象
# 命令对象: 接收者 + 执行方法
# 调用方只发命令,不知接收者细节

# 优点:
# - 撤销/重做(保存命令历史)
# - 队列/延迟执行
# - 日志与审计
# - 发送与执行解耦

# 缺点:
# - 类数量增加
# - 简单调用过度设计

结构

# 参与者:
# Command: 命令接口(execute)
# ConcreteCommand: 具体命令(持接收者)
# Receiver: 接收者(实际执行)
# Invoker: 调用者(发命令)
# Client: 组装

# 例: 遥控器
# 按钮(Invoker) → 开灯命令(Command) → 灯(Receiver)

Python 示例: 文本编辑器

from abc import ABC, abstractmethod

# 接收者
class TextEditor:
    def __init__(self):
        self.text = ""

    def insert(self, text, pos=None):
        self.text += text

    def delete(self, count):
        removed = self.text[-count:]
        self.text = self.text[:-count]
        return removed

# 命令接口
class Command(ABC):
    @abstractmethod
    def execute(self): ...
    @abstractmethod
    def undo(self): ...

# 具体命令: 插入
class InsertCommand(Command):
    def __init__(self, editor, text):
        self.editor = editor
        self.text = text

    def execute(self):
        self.editor.insert(self.text)

    def undo(self):
        self.editor.delete(len(self.text))

# 具体命令: 删除
class DeleteCommand(Command):
    def __init__(self, editor, count):
        self.editor = editor
        self.count = count
        self._removed = ""

    def execute(self):
        self._removed = self.editor.delete(
            self.count)

    def undo(self):
        self.editor.insert(self._removed)

# 调用者(Invoker): 管理命令历史
class CommandHistory:
    def __init__(self):
        self.history = []
        self.editor = TextEditor()

    def execute(self, command: Command):
        command.execute()
        self.history.append(command)

    def undo(self):
        if self.history:
            cmd = self.history.pop()
            cmd.undo()

# 使用
app = CommandHistory()

app.execute(InsertCommand(app.editor, "Hello"))
app.execute(InsertCommand(app.editor, " World"))
print(app.editor.text)      # Hello World

app.execute(DeleteCommand(app.editor, 6))
print(app.editor.text)      # Hello

app.undo()                  # 撤销删除
print(app.editor.text)      # Hello World

app.undo()                  # 撤销插入
print(app.editor.text)      # Hello

app.undo()                  # 撤销插入
print(app.editor.text)      # (空)

# 撤销/重做栈:
# undo_stack / redo_stack 双栈实现重做

Python 简化: 函数命令

# Python 中函数即可作命令
class Editor:
    def __init__(self):
        self.text = ""
        self.history = []

    def execute(self, action, undo_action):
        action()
        self.history.append(undo_action)

    def undo(self):
        if self.history:
            self.history.pop()()

editor = Editor()
# 命令就是闭包/函数
editor.execute(
    lambda: editor.text.__add__("a"),
    # 简化示意
)
# 更清晰: 用函数封装

def insert(s):
    editor.text += s
    def undo():
        editor.text = editor.text[:-len(s)]
    return undo

# 命令 = 函数 + 撤销函数
cmds = []
cmds.append(insert("Hello"))

# 执行
for cmd in cmds:
    cmd()

# 用类型别名组织
from typing import Callable
CommandFn = Callable[[], None]

应用场景

# 1. 撤销/重做(编辑器/画板)
# 2. 任务队列(延迟/批处理)
# 3. 宏命令(组合多个命令)
# 4. 事务(全部成功或回滚)
# 5. UI 按钮/菜单(解耦)
# 6. 网络请求封装(可重试)

# 实际案例:
# - 数据库事务(提交/回滚)
# - IDE 操作历史(撤销栈)
# - 游戏操作(回放)
# - 分布式任务调度

# 宏命令(组合)
class MacroCommand(Command):
    def __init__(self, commands):
        self.commands = commands

    def execute(self):
        for c in self.commands:
            c.execute()

    def undo(self):
        for c in reversed(self.commands):
            c.undo()

优缺点总结

# 优点:
# ✅ 撤销/重做
# ✅ 队列/延迟/日志
# ✅ 解耦发送与执行
# ✅ 组合命令(宏)

# 缺点:
# ❌ 类数量增加
# ❌ 简单调用过度设计

# 决策:
# 需要撤销/队列 → 命令模式
# 简单调用 → 直接调用
# Python: 函数/闭包可作轻量命令

# 最佳实践:
# - 命令保持小且单一
# - 撤销逻辑与执行对称
# - 历史栈限长(防内存)
# - 组合命令(宏)复用

命令模式把"请求"变成"对象"。掌握执行/撤销与历史栈,编辑器、队列、事务类需求从容应对。

解释器模式

解释器模式(Interpreter Pattern)

解释器:为语言定义文法表示并解释语句。适合定义"规则/表达式"并求值的场景。

意图

# 问题: 需要解析并执行"规则/表达式"
# - 配置规则(如 "age > 18 AND active")
# - 简单脚本/查询语言

# 解决: 表达式树 + 解释执行
# 每个文法规则对应一个类

# 优点:
# - 易于扩展文法(加规则=加类)
# - 表达式树结构清晰

# 缺点:
# - 类数量膨胀(规则多时)
# - 复杂文法用解释器不划算
#   (应改用解析器库: lark/pyparsing)

结构

# 参与者:
# AbstractExpression: 抽象表达式(interpret)
# TerminalExpression: 终结符(变量/常量)
# NonterminalExpression: 非终结符(运算组合)
# Context: 上下文(变量值等)

# 例: 算术表达式
# 1 + 2 * 3 → 表达式树
#        +
#       / \
#      1   *
#         / \
#        2   3

Python 示例: 数学表达式

from abc import ABC, abstractmethod

# 抽象表达式
class Expression(ABC):
    @abstractmethod
    def interpret(self, context: dict) -> int: ...

# 终结符: 数字
class Number(Expression):
    def __init__(self, value: int):
        self.value = value

    def interpret(self, context):
        return self.value

# 终结符: 变量
class Variable(Expression):
    def __init__(self, name: str):
        self.name = name

    def interpret(self, context):
        return context.get(self.name, 0)

# 非终结符: 二元运算
class Add(Expression):
    def __init__(self, left, right):
        self.left, self.right = left, right
    def interpret(self, context):
        return (self.left.interpret(context)
                + self.right.interpret(context))

class Multiply(Expression):
    def __init__(self, left, right):
        self.left, self.right = left, right
    def interpret(self, context):
        return (self.left.interpret(context)
                * self.right.interpret(context))

# 构建表达式: (x + 5) * 2
expr = Multiply(
    Add(Variable("x"), Number(5)),
    Number(2),
)

# 解释执行
print(expr.interpret({"x": 10}))   # 30
print(expr.interpret({"x": 3}))    # 16

# 更复杂: ((x + 1) * (y - 2))
class Subtract(Expression):
    def __init__(self, left, right):
        self.left, self.right = left, right
    def interpret(self, context):
        return (self.left.interpret(context)
                - self.right.interpret(context))

expr2 = Multiply(
    Add(Variable("x"), Number(1)),
    Subtract(Variable("y"), Number(2)),
)
print(expr2.interpret({"x": 5, "y": 8}))  # (6)*(6)=36

# 新增运算: 除、幂 → 各加一个类
# 文法扩展容易,但类会变多

应用: 规则引擎

# 场景: 促销规则(满减/折扣)
# "满 100 减 20 且 VIP 双倍积分"

class And(Expression):
    def __init__(self, left, right):
        self.left, self.right = left, right
    def interpret(self, context):
        return (bool(self.left.interpret(context))
                and bool(self.right.interpret(context)))

class Or(Expression):
    def __init__(self, left, right):
        self.left, self.right = left, right
    def interpret(self, context):
        return (bool(self.left.interpret(context))
                or bool(self.right.interpret(context)))

class GreaterThan(Expression):
    def __init__(self, var, threshold):
        self.var, self.threshold = var, threshold
    def interpret(self, context):
        return context.get(self.var, 0) > self.threshold

class IsVIP(Expression):
    def interpret(self, context):
        return bool(context.get("vip", False))

# 规则: (金额 > 100) AND (是VIP)
rule = And(GreaterThan("amount", 100), IsVIP())

print(rule.interpret({"amount": 150, "vip": True}))
# True

print(rule.interpret({"amount": 150, "vip": False}))
# False

# 生产建议: 简单规则用解释器,
# 复杂 DSL 用 lark/pyparsing/rulegraph 库
# 或用 Python 表达式 eval(注意安全)

Python 替代方案

# 1. 简单表达式: eval/exec(注意安全!)
# 只对可信输入使用
expr = "x + 5 * 2"
result = eval(expr, {"__builtins__": {}}, {"x": 10})

# 安全: ast.literal_eval(仅字面量)
import ast
ast.literal_eval("[1, 2, 3]")   # 安全

# 2. 解析器库(正式 DSL)
# pip install lark
from lark import Lark, Transformer

grammar = """
start: expr
expr: expr "+" expr   -> add
    | expr "*" expr   -> mul
    | NUMBER          -> num
%import common.NUMBER
%ignore " "
"""
parser = Lark(grammar)

# 3. 规则库: rule-engine / simple-rules
# 4. 配置文件 + dict 映射(最简单)

优缺点总结

# 优点:
# ✅ 文法可扩展(加类)
# ✅ 表达式树清晰
# ✅ 符合开闭原则

# 缺点:
# ❌ 类数量膨胀
# ❌ 复杂文法维护难
# ❌ 性能(树遍历)

# 决策:
# 简单规则/表达式 → 解释器
# 复杂文法 → 解析器库(lark)
# 简单配置 → 直接逻辑判断

# 最佳实践:
# - 文法简单有限时用
# - 提供上下文对象
# - 配合工厂解析文本→表达式树
# - 复杂场景优先选库

解释器让"规则"可编程求值。掌握表达式树与上下文,简单 DSL 与规则引擎手到擒来。

迭代器模式

迭代器模式(Iterator Pattern)

迭代器:提供统一方式顺序访问集合元素,而不暴露集合内部结构。Python 的迭代协议是该模式的语言级实现。

意图

# 问题: 不同集合遍历方式不同
# - 数组/链表/树遍历方式各异
# - 客户端需要知道内部结构

# 解决: 统一迭代器接口(next/has_next)
# 客户端用相同方式遍历所有集合

# 优点:
# - 统一遍历接口
# - 隐藏集合内部
# - 支持多种遍历(正/反/跳跃)

# 缺点:
# - 简单集合过度设计
# - 迭代器失效(修改集合时)

Python 迭代协议

# Python 内置迭代协议:
# __iter__: 返回迭代器
# __next__: 返回下一个(无则 StopIteration)

# 所有内置容器都支持
for x in [1, 2, 3]: ...       # 列表
for x in "abc": ...           # 字符串
for k in {"a": 1}: ...        # 字典
for x in range(5): ...        # range

# 手动迭代
it = iter([1, 2, 3])
next(it)   # 1
next(it)   # 2
next(it)   # 3
# next(it) → StopIteration

# 生成器(最简迭代器)
def gen():
    yield 1
    yield 2

for x in gen(): ...

自定义迭代器

class MyCollection:
    def __init__(self, items):
        self._items = items

    def __iter__(self):
        return MyIterator(self)

class MyIterator:
    def __init__(self, collection):
        self._items = collection._items
        self._index = 0

    def __iter__(self):
        return self

    def __next__(self):
        if self._index >= len(self._items):
            raise StopIteration
        value = self._items[self._index]
        self._index += 1
        return value

col = MyCollection([10, 20, 30])
for x in col:
    print(x)      # 10 20 30

多种迭代方式

# 集合提供多个迭代器
class BookShelf:
    def __init__(self, books):
        self.books = books

    # 正向迭代
    def __iter__(self):
        return iter(self.books)

    # 反向迭代
    def reversed_iter(self):
        return reversed(self.books)

    # 奇数位迭代(自定义)
    def odd_iter(self):
        return (b for i, b in enumerate(self.books)
                if i % 2 == 0)

shelf = BookShelf(["A", "B", "C", "D"])

print(list(shelf))           # ['A','B','C','D']
print(list(shelf.reversed_iter()))  # ['D','C','B','A']
print(list(shelf.odd_iter()))       # ['A','C']

生成器迭代(惰性)

# 生成器 = 惰性迭代器(省内存)
def read_lines(filename):
    """逐行读取(大数据友好)"""
    with open(filename, encoding="utf-8") as f:
        for line in f:
            yield line.strip()

# 无限序列
def fibonacci():
    a, b = 0, 1
    while True:
        yield a
        a, b = b, a + b

# 只取前 10 个
import itertools
first10 = list(itertools.islice(fibonacci(), 10))
print(first10)   # [0,1,1,2,3,5,8,13,21,34]

# 迭代工具(itertools)
itertools.chain(a, b)      # 串联
itertools.cycle(items)     # 循环
itertools.repeat(x, 3)     # 重复
itertools.zip_longest(a,b) # 长短不一
itertools.groupby(items)   # 分组

实际应用

# 1. 自定义容器类(让客户端 for 遍历)
# 2. 惰性大数据(流式读取)
# 3. 树/图的遍历封装(DFS/BFS 迭代器)
# 4. 分页迭代器(自动翻页)

# 分页迭代器示例
class PaginatedIterator:
    def __init__(self, fetch_page, page_size=20):
        self.fetch = fetch_page
        self.page_size = page_size
        self.page = 0
        self.items = []
        self.index = 0

    def __iter__(self):
        return self

    def __next__(self):
        if self.index >= len(self.items):
            self.page += 1
            self.items = self.fetch(self.page)
            self.index = 0
            if not self.items:
                raise StopIteration
        item = self.items[self.index]
        self.index += 1
        return item

# 使用: 自动翻页加载
def api_page(page):
    # 模拟 API 分页
    data = [f"第{page}页-{i}" for i in range(3)]
    return data if page <= 2 else []

for item in PaginatedIterator(api_page):
    print(item)
# 第1页-0,1,2; 第2页-0,1,2

优缺点总结

# 优点:
# ✅ 统一遍历接口
# ✅ 隐藏内部结构
# ✅ 惰性求值(省内存)
# ✅ 多种遍历方式

# 缺点:
# ❌ 简单集合用不上
# ❌ 迭代中修改集合易出错

# Python 优势:
# 迭代协议内建,几乎零成本
# 生成器让惰性迭代极简
# itertools 提供丰富工具

# 最佳实践:
# - 自定义容器实现 __iter__
# - 大数据用生成器
# - 遍历中不修改集合(或操作副本)
# - 多迭代方式用独立方法返回新迭代器

迭代器是遍历的"统一语言",Python 内建支持。掌握 __iter__/__next__ 与生成器,集合遍历灵活高效。

中介者模式

中介者模式(Mediator Pattern)

中介者:用一个中介对象封装一组对象的交互,减少对象间的直接耦合(类似"交通指挥中心")。

意图

# 问题: 多个对象互相交互(网状耦合)
# - 每个对象要知道其他所有对象
# - 新增对象要改所有关联对象

# 解决: 对象只与中介者通信(星形)
# 中介者协调各对象间的交互

# 优点:
# - 减少对象间耦合(网状→星形)
# - 交互逻辑集中
# - 新增对象更简单

# 缺点:
# - 中介者可能膨胀(上帝对象)
# - 交互逻辑集中后难以复用

结构

# 参与者:
# Mediator: 中介者接口(notify)
# ConcreteMediator: 具体中介者(协调)
# Colleague: 同事接口(持中介者引用)
# ConcreteColleague: 具体同事

# 无中介: A→B, A→C, B→C(网状)
# 有中介: A→M, B→M, C→M(星形)

Python 示例: 聊天室

# 中介者: 聊天室
class ChatRoom:
    def __init__(self):
        self.users = {}

    def register(self, user):
        self.users[user.name] = user

    def send(self, message, sender):
        for name, user in self.users.items():
            if name != sender.name:
                user.receive(f"[{sender.name}] {message}")

    def send_to(self, message, sender, target):
        if target in self.users:
            self.users[target].receive(
                f"[{sender.name}->你] {message}")

# 同事: 用户(只与聊天室通信)
class User:
    def __init__(self, name, room: ChatRoom):
        self.name = name
        self.room = room
        room.register(self)

    def send(self, message):
        self.room.send(message, self)

    def send_to(self, message, target):
        self.room.send_to(message, self, target)

    def receive(self, message):
        print(f"{self.name} 收到: {message}")

# 使用: 对象之间不直接引用!
room = ChatRoom()
alice = User("Alice", room)
bob = User("Bob", room)
carol = User("Carol", room)

alice.send("大家好!")
# Bob 收到: [Alice] 大家好!
# Carol 收到: [Alice] 大家好!

bob.send_to("你好 Alice", "Alice")
# Alice 收到: [Bob->你] 你好 Alice

# 新增用户: 只需 register,无需改其他类
dave = User("Dave", room)

应用: 表单联动

# 场景: 表单组件联动
# 国家 → 城市 → 邮编; 开关 → 禁用

class FormMediator:
    def __init__(self):
        self.components = {}

    def register(self, name, component):
        component.mediator = self
        self.components[name] = component

    def notify(self, sender, event, data=None):
        if sender == "country" and event == "change":
            # 联动: 更新城市选项
            self.components["city"].set_options(
                cities_by_country[data])
            self.components["zip"].clear()
        elif sender == "toggle" and event == "change":
            # 联动: 开关控制输入禁用
            self.components["input"].set_disabled(
                not data)

# 组件基类
class Component:
    def __init__(self, name):
        self.name = name
        self.mediator = None

    def changed(self, event, data=None):
        if self.mediator:
            self.mediator.notify(
                self.name, event, data)

# 使用
mediator = FormMediator()
country = Component("country")
city = Component("city")
toggle = Component("toggle")
input_box = Component("input")
for c in [country, city, toggle, input_box]:
    mediator.register(c.name, c)

# 触发联动
country.changed("change", "CN")
toggle.changed("change", False)

# 对象解耦: 组件互相不认识!

与观察者的对比

# 观察者(Observer):
# 一对多通知(发布-订阅)
# 发布者不知道订阅者细节

# 中介者(Mediator):
# 多对多交互(集中协调)
# 对象间不直接通信

# 对比:
# 观察者: 单向通知流
# 中介者: 双向协调交互

# 选择:
# 一对多通知 → 观察者
# 复杂网状交互 → 中介者
# 事件总线常结合两者

# Python 中事件总线/信号槽:
# - blinker / pyee(事件库)
# - Django signals
# 都是中介者+观察者思想的实现

优缺点总结

# 优点:
# ✅ 解耦(网状→星形)
# ✅ 交互逻辑集中
# ✅ 新增对象影响小

# 缺点:
# ❌ 中介者膨胀(上帝对象)
# ❌ 交互逻辑难复用
# ❌ 集中后复杂度高

# 最佳实践:
# - 中介者保持职责聚焦
# - 交互协议清晰(事件/通知)
# - 大系统拆多个中介者
# - 配合观察者模式

# 决策:
# 3-4 个对象交互 → 直接引用即可
# 对象多/交互复杂 → 中介者
# 聊天室/表单联动/UI 协作 → 中介者

中介者把"网状耦合"变为"星形协调"。掌握注册/通知机制,聊天室、表单联动、UI 协作清晰解耦。

备忘录模式

备忘录模式(Memento Pattern)

备忘录:在不破坏封装的前提下,捕获并保存对象的内部状态,以便之后恢复到该状态(快照/存档)。

意图

# 问题: 需要保存/恢复对象状态
# - 撤销/重做
# - 存档/读档(游戏)
# - 直接保存内部字段破坏封装

# 解决: 备忘录对象保存状态快照
# 由"发起人"创建与恢复,外部不可直接访问

# 优点:
# - 不破坏封装
# - 简化发起人(状态管理分离)
# - 支持多级恢复

# 缺点:
# - 备忘录多时内存开销
# - 历史管理复杂

结构

# 参与者:
# Originator: 发起人(创建/恢复备忘录)
# Memento: 备忘录(保存状态,不透明)
# Caretaker: 管理者(保存备忘录列表)

# 关键: 备忘录内部状态对外不可见
# Python 约定用私有/下划线,靠约定

Python 示例: 文本编辑器

# 备忘录(状态快照)
class Memento:
    def __init__(self, state):
        # 保存不可变副本
        self._state = state.copy()

    def _get_state(self):
        return self._state.copy()

# 发起人
class Editor:
    def __init__(self):
        self.content = ""
        self.cursor = 0

    def type(self, text):
        self.content += text
        self.cursor += len(text)

    def save(self) -> Memento:
        # 创建快照(封装状态)
        return Memento({
            "content": self.content,
            "cursor": self.cursor,
        })

    def restore(self, memento: Memento):
        # 从快照恢复
        state = memento._get_state()
        self.content = state["content"]
        self.cursor = state["cursor"]

    def __str__(self):
        return f"Editor('{self.content}', cursor={self.cursor})"

# 管理者: 历史栈
class History:
    def __init__(self):
        self._stack = []

    def push(self, memento):
        self._stack.append(memento)

    def pop(self):
        return self._stack.pop() if self._stack                else None

# 使用
editor = Editor()
history = History()

# 输入并保存快照
editor.type("Hello")
history.push(editor.save())
print(editor)   # Editor('Hello', cursor=5)

editor.type(" World")
history.push(editor.save())
print(editor)   # Editor('Hello World', cursor=11)

editor.type("!")
print(editor)   # Editor('Hello World!', cursor=12)

# 撤销
editor.restore(history.pop())
print(editor)   # Editor('Hello World', cursor=11)

editor.restore(history.pop())
print(editor)   # Editor('Hello', cursor=5)

# 多级撤销 + 恢复
# 需要重做时: 双栈(undo/redo)

应用: 游戏存档

class GameState:
    """备忘录: 游戏状态快照"""
    def __init__(self, health, level,
                 inventory):
        self.health = health
        self.level = level
        self.inventory = inventory.copy()

class Game:
    def __init__(self):
        self.health = 100
        self.level = 1
        self.inventory = []

    def damage(self, amount):
        self.health -= amount

    def level_up(self):
        self.level += 1

    def add_item(self, item):
        self.inventory.append(item)

    # 创建快照
    def create_snapshot(self) -> GameState:
        return GameState(self.health, self.level,
                         self.inventory)

    # 恢复快照
    def restore(self, state: GameState):
        self.health = state.health
        self.level = state.level
        self.inventory = state.inventory.copy()

# 存档管理(每关自动存档)
game = Game()
saves = []

game.level_up()
game.add_item("剑")
saves.append(game.create_snapshot())  # 存档点1

game.damage(50)
game.add_item("药水")
saves.append(game.create_snapshot())  # 存档点2

# 挂了,读档
game.restore(saves[-1])   # 回到存档点2
print(game.health, game.level, game.inventory)

Python 简化

# 1. 用 copy 快照
import copy

class Editor:
    def save(self):
        return copy.deepcopy(self.__dict__)

    def restore(self, state):
        self.__dict__ = copy.deepcopy(state)

# 2. dataclasses.replace(不可变风格)
from dataclasses import dataclass, replace

@dataclass(frozen=True)
class State:
    content: str = ""
    cursor: int = 0

# 修改 = 生成新状态(不可变)
state = State("Hello", 5)
new_state = replace(state, content="Hi")
# 旧状态自动保留(天然快照)

# 3. 序列化存档(跨会话)
import json

def save_game(game):
    return json.dumps(game.__dict__)

def load_game(data):
    game = Game()
    game.__dict__ = json.loads(data)
    return game

优缺点总结

# 优点:
# ✅ 不破坏封装
# ✅ 状态管理分离
# ✅ 多级撤销/存档

# 缺点:
# ❌ 快照内存开销
# ❌ 历史管理复杂
# ❌ 大对象快照成本高

# 最佳实践:
# - 快照保存不可变副本
# - 历史栈限制长度
# - 大对象只保存差异(增量)
# - Python 用 copy/dataclass 简化

# 决策:
# 需要撤销/存档 → 备忘录
# 简单状态 → 直接变量/序列化
# 不可变风格 → dataclass + replace

备忘录是"时间旅行"的快照机制。掌握创建/恢复与历史管理,撤销、存档、状态回滚游刃有余。

观察者模式

观察者模式(Observer Pattern)

观察者:定义一对多依赖,当对象状态变化时,所有依赖者自动收到通知(发布-订阅)。

意图

# 问题: 对象状态变化需要通知多个对象
# - 直接引用: 耦合,新增观察者要改主题
# - 变化频繁: 忘记通知某对象

# 解决: 观察者注册到主题
# 主题变化时批量通知所有观察者

# 优点:
# - 解耦(主题不知道观察者细节)
# - 开闭原则(新增观察者容易)
# - 广播通知

# 缺点:
# - 通知顺序不确定
# - 观察者异常影响主题
# - 内存泄漏(未注销)

结构

# 参与者:
# Subject: 主题(维护观察者列表,通知)
# Observer: 观察者接口(update)
# ConcreteSubject: 具体主题
# ConcreteObserver: 具体观察者

# 例: 天气站(Subject) → 各显示屏(Observer)

Python 示例: 天气站

from abc import ABC, abstractmethod

# 观察者接口
class Observer(ABC):
    @abstractmethod
    def update(self, temperature, humidity,
               pressure): ...

# 主题
class WeatherStation:
    def __init__(self):
        self._observers = []
        self._temperature = 0
        self._humidity = 0

    # 注册/注销
    def attach(self, observer):
        self._observers.append(observer)

    def detach(self, observer):
        self._observers.remove(observer)

    # 设置新数据并通知
    def set_weather(self, temp, humidity):
        self._temperature = temp
        self._humidity = humidity
        self._notify()

    def _notify(self):
        for obs in self._observers:
            obs.update(self._temperature,
                       self._humidity, 1013)

# 具体观察者
class PhoneDisplay(Observer):
    def __init__(self, name):
        self.name = name
    def update(self, temp, hum, press):
        print(f"{self.name}: {temp}°C, "
              f"湿度{hum}%")

class WebDisplay(Observer):
    def update(self, temp, hum, press):
        print(f"网站显示: {temp}°C")

# 使用
station = WeatherStation()
phone = PhoneDisplay("手机")
web = WebDisplay()

station.attach(phone)
station.attach(web)

station.set_weather(25, 60)
# 手机: 25°C, 湿度60%
# 网站显示: 25°C

# 动态加/减观察者
station.detach(web)
station.set_weather(28, 55)
# 手机: 28°C, 湿度55%(网站不再收到)

事件总线(通用观察者)

# 事件总线: 按事件类型注册观察者
class EventBus:
    def __init__(self):
        self._listeners = {}   # 事件 → [回调]

    def on(self, event, callback):
        self._listeners.setdefault(
            event, []).append(callback)
        return callback

    def off(self, event, callback):
        if event in self._listeners:
            self._listeners[event].remove(callback)

    def emit(self, event, *args, **kwargs):
        for cb in self._listeners.get(event, []):
            cb(*args, **kwargs)

# 使用
bus = EventBus()

@bus.on("order_created")
def send_notification(order):
    print(f"通知: 新订单 {order['id']}")

@bus.on("order_created")
def deduct_stock(order):
    print(f"扣库存: {order['item']}")

bus.emit("order_created",
         {"id": 1, "item": "手机"})
# 通知: 新订单 1
# 扣库存: 手机

# 应用: 解耦业务模块
# 下单 → 通知/库存/积分/日志 各听各的

Python 内置信号/回调

# 1. Django signals(Web 项目常用)
from django.dispatch import Signal
order_created = Signal()

def on_order(sender, **kwargs):
    print("订单事件:", kwargs)

order_created.connect(on_order)
order_created.send(sender=None, order_id=1)

# 2. 属性监听(描述符)
class Observable:
    def __init__(self):
        self._value = None
        self._callbacks = []

    def bind(self, fn):
        self._callbacks.append(fn)

    @property
    def value(self):
        return self._value

    @value.setter
    def value(self, new):
        self._value = new
        for fn in self._callbacks:
            fn(new)

obs = Observable()
obs.bind(lambda v: print(f"变化: {v}"))
obs.value = 42    # 变化: 42

实际应用

# 1. UI 事件(按钮点击/输入变化)
# 2. 数据变化通知(视图更新)
# 3. 消息队列/事件驱动
# 4. 订阅推送(股票/天气/新闻)
# 5. 数据库触发器/监听

# 框架中的观察者:
# - RxJS Observable(响应式)
# - Vue/React 响应式数据
# - Django signals
# - 消息中间件(Redis pub/sub)

# 与中介者对比:
# 观察者: 一对多广播(单向)
# 中介者: 多对多协调(双向集中)

优缺点总结

# 优点:
# ✅ 解耦主题与观察者
# ✅ 广播通知
# ✅ 开闭原则

# 缺点:
# ❌ 通知顺序不确定
# ❌ 观察者异常需隔离
# ❌ 忘记注销(内存泄漏)

# 最佳实践:
# - 观察者回调捕获异常
# - 使用完 detach/off
# - 用弱引用(weakref)防泄漏
# - 复杂场景用事件总线/消息队列

# 决策:
# 一对多通知 → 观察者
# 系统级解耦 → 事件总线/消息队列
# 响应式链式 → RxJS

观察者让"状态变化"自动广播。掌握注册/通知机制与事件总线,通知、联动、事件驱动架构信手拈来。

状态模式

状态模式(State Pattern)

状态:对象的行为随内部状态变化而变化,把每个状态的行为封装为独立的类,消除大量 if-else。

意图

# 问题: 对象行为依赖状态(if-else 泛滥)
# - 状态多且行为各异
# - 状态转换逻辑分散

# 解决: 每个状态一个类
# 对象持有"当前状态",行为委托给状态对象
# 状态转换 = 替换状态对象

# 优点:
# - 消除状态 if-else
# - 状态行为内聚
# - 新增状态容易
# - 转换逻辑集中

# 缺点:
# - 类数量增加
# - 状态多时系统复杂

结构

# 参与者:
# Context: 上下文(持当前状态,委托行为)
# State: 状态接口
# ConcreteState: 具体状态(行为+转换)

# 例: 订单状态
# 待支付 → 已支付 → 已发货 → 已完成
# 每个状态可执行的行为不同

Python 示例: 订单状态机

from abc import ABC, abstractmethod

# 上下文(订单)
class Order:
    def __init__(self):
        # 初始状态: 待支付
        self.state = PendingState()

    def set_state(self, state):
        print(f"状态: {self} → {state}")
        self.state = state

    # 委托给当前状态
    def pay(self):
        self.state.pay(self)

    def ship(self):
        self.state.ship(self)

    def complete(self):
        self.state.complete(self)

    def __str__(self):
        return self.state.__str__()

# 状态接口
class OrderState(ABC):
    @abstractmethod
    def pay(self, order): ...
    @abstractmethod
    def ship(self, order): ...
    @abstractmethod
    def complete(self, order): ...

# 待支付
class PendingState(OrderState):
    def pay(self, order):
        print("支付成功")
        order.set_state(PaidState())

    def ship(self, order):
        print("错误: 未支付不能发货")

    def complete(self, order):
        print("错误: 未支付不能完成")

    def __str__(self):
        return "待支付"

# 已支付
class PaidState(OrderState):
    def pay(self, order):
        print("错误: 已支付,不能重复支付")

    def ship(self, order):
        print("发货成功")
        order.set_state(ShippedState())

    def complete(self, order):
        print("错误: 未发货不能完成")

    def __str__(self):
        return "已支付"

# 已发货
class ShippedState(OrderState):
    def pay(self, order):
        print("错误: 已发货,无需支付")

    def ship(self, order):
        print("错误: 已发货,不能重复发货")

    def complete(self, order):
        print("订单完成")
        order.set_state(CompletedState())

    def __str__(self):
        return "已发货"

# 已完成
class CompletedState(OrderState):
    def pay(self, order):
        print("错误: 订单已完成")

    def ship(self, order):
        print("错误: 订单已完成")

    def complete(self, order):
        print("错误: 已完成,不能重复")

    def __str__(self):
        return "已完成"

# 使用: 状态驱动的行为
order = Order()
print(order)          # 待支付

order.ship()          # 错误: 未支付不能发货
order.pay()           # 支付成功 → 已支付
order.pay()           # 错误: 已支付,不能重复支付
order.ship()          # 发货成功 → 已发货
order.complete()      # 订单完成 → 已完成
order.complete()      # 错误: 已完成,不能重复

# 非法操作被状态对象拦截,逻辑内聚!

状态机(更通用的实现)

# 用 dict 定义状态转换表(轻量状态机)
TRANSITIONS = {
    "pending": {
        "pay": ("paid", "支付成功"),
        "ship": (None, "错误: 未支付不能发货"),
    },
    "paid": {
        "ship": ("shipped", "发货成功"),
        "pay": (None, "错误: 已支付"),
    },
    "shipped": {
        "complete": ("completed", "订单完成"),
    },
}

class SimpleOrder:
    def __init__(self):
        self.state = "pending"

    def act(self, action):
        rule = TRANSITIONS[self.state].get(action)
        if not rule:
            print(f"错误: 状态[{self.state}] "
                  f"无操作[{action}]")
            return
        next_state, message = rule
        if next_state:
            self.state = next_state
        print(message)

o = SimpleOrder()
o.act("pay")     # 支付成功
o.act("ship")    # 发货成功
o.act("pay")     # 错误: 已支付
o.act("complete")  # 订单完成

# 或使用第三方库: transitions
# pip install transitions
# 适合复杂状态机(嵌套/动作/守卫)

实际应用

# 1. 订单/审批流程状态机
# 2. 播放器状态(播放/暂停/停止)
# 3. 网络连接状态(连接/断开/重试)
# 4. 游戏角色状态(站立/跑动/跳跃)
# 5. 业务流程引擎

# 状态模式 vs 状态机:
# 状态模式: 面向对象(状态类)
# 状态机: 数据驱动(dict/库)
# 简单状态机用 dict,复杂用库/状态类

# 与策略模式对比:
# 策略: 行为可替换(算法)
# 状态: 行为随状态变(自动切换)
# 状态 = 策略 + 自动状态转换

优缺点总结

# 优点:
# ✅ 消除状态 if-else
# ✅ 状态行为内聚
# ✅ 新增状态容易
# ✅ 非法操作拦截

# 缺点:
# ❌ 类数量增加
# ❌ 状态类间共享上下文需注意
# ❌ 状态转换逻辑分散在状态类

# 最佳实践:
# - 上下文只做委托
# - 状态转换集中在各状态类
# - 非法操作明确报错
# - 简单状态机用 dict/库
# - 状态类尽量无状态(用上下文数据)

状态模式把"状态逻辑"结构化。掌握状态类与委托,订单流、播放器、连接管理等场景告别 if-else 泥潭。

空对象模式

空对象模式(Null Object Pattern)

空对象:用"什么都不做"的对象替代 None,消除客户端大量的 None 判断(null check)。

意图

# 问题: 返回值可能为 None
# - 客户端到处都是 if x is not None
# - 忘记判断 → AttributeError

# 解决: 定义空对象(实现相同接口,但空操作)
# 客户端统一调用,无需判空

# 优点:
# - 消除判空代码
# - 统一客户端逻辑
# - 默认行为优雅(空操作)

# 缺点:
# - 空对象可能隐藏错误
# - 接口需支持空实现
# - 调试时不易发现"为什么没效果"

结构

# 参与者:
# AbstractObject: 抽象接口
# RealObject: 真实对象
# NullObject: 空对象(空操作实现)

# Client: 统一调用,不判空

# 例: 用户查询
# get_user(id) → 真实用户 或 NullUser

Python 示例: 用户系统

from abc import ABC, abstractmethod

# 抽象接口
class User(ABC):
    @abstractmethod
    def get_name(self) -> str: ...
    @abstractmethod
    def get_role(self) -> str: ...
    @abstractmethod
    def is_active(self) -> bool: ...

# 真实用户
class RealUser(User):
    def __init__(self, name, role):
        self.name = name
        self.role = role

    def get_name(self): return self.name
    def get_role(self): return self.role
    def is_active(self): return True

# 空对象
class NullUser(User):
    def get_name(self): return "游客"
    def get_role(self): return "guest"
    def is_active(self): return False

# 工厂: 返回真实或空对象(绝不返回 None)
class UserService:
    def __init__(self):
        self._users = {"1": RealUser("张三", "admin")}

    def get_user(self, user_id) -> User:
        # 找不到时返回空对象,而非 None!
        return self._users.get(user_id, NullUser())

# 使用: 无需判空!
service = UserService()

for uid in ["1", "999"]:
    user = service.get_user(uid)
    # 直接调用,不用 if user is None
    print(f"{user.get_name()} "
          f"({user.get_role()}, "
          f"active={user.is_active()})")
# 张三 (admin, active=True)
# 游客 (guest, active=False)

# 对比: 传统 None 方式
user = service.get_user("999")
if user is not None:      # 到处判空
    print(user.get_name())
else:
    print("用户不存在")

# 空对象方式更统一

应用: 日志/命令/配置

# 1. 空日志器
class Logger(ABC):
    @abstractmethod
    def log(self, msg): ...

class RealLogger(Logger):
    def log(self, msg):
        print(f"[LOG] {msg}")

class NullLogger(Logger):
    def log(self, msg):
        pass   # 什么都不做

# 配置可关闭日志
logger = RealLogger() if debug else NullLogger()
logger.log("启动")   # 统一调用

# 2. 空命令(菜单项无操作)
class Command(ABC):
    @abstractmethod
    def execute(self): ...

class NoopCommand(Command):
    def execute(self):
        pass   # 空操作

# 3. 默认配置
class Config(ABC):
    @abstractmethod
    def get(self, key): ...

class NullConfig(Config):
    def get(self, key):
        return None   # 或默认值

# 4. 空集合(替代 None 列表)
EMPTY_LIST = []
def get_items():
    return found_items or EMPTY_LIST
# 客户端直接遍历,无需判空

Python 简化

# 1. 默认参数/可调用
def get_user(user_id):
    return self._users.get(
        user_id, None)

# 客户端用 getattr 默认
user = get_user("999")
name = getattr(user, "name", "游客")
role = getattr(user, "role", "guest")

# 2. 返回默认对象
DEFAULT_USER = RealUser("游客", "guest")
def get_user(user_id):
    return self._users.get(user_id,
                           DEFAULT_USER)
# 注意: 共享默认对象需不可变

# 3. dict 默认值
self._users = defaultdict(lambda: RealUser(
    "游客", "guest"))

# 4. 可选属性用 None-safe 工具
# 或 pydantic 默认值
from dataclasses import dataclass, field

@dataclass
class UserDC:
    name: str = "游客"
    role: str = "guest"
    active: bool = False

优缺点总结

# 优点:
# ✅ 消除判空
# ✅ 客户端逻辑统一
# ✅ 默认行为清晰

# 缺点:
# ❌ 空对象可能掩盖错误
# ❌ 接口需支持空实现
# ❌ 调试困难(无日志无报错)

# 最佳实践:
# - 空对象保持"语义明确"(游客/默认)
# - 关键路径慎用(可能隐藏 bug)
# - 空对象记录可观测性(可选)
# - 配合工厂返回(客户端无感)

# 决策:
# 频繁判空且默认行为明确 → 空对象
# 需要区分"不存在"场景 → 保持 None
# 错误需要暴露 → 用异常/None

空对象用"空操作"替代判空。掌握空对象设计与默认行为,客户端代码干净统一,少一堆 if 判断。

策略模式

策略模式(Strategy Pattern)

策略:定义一系列算法并封装,让它们可以互相替换。客户端在运行时选择策略,消除大量条件分支。

意图

# 问题: 同一操作有多种算法实现
# - if-else 选择算法,难扩展
# - 算法代码混杂在业务中

# 解决: 每个算法一个策略类
# 客户端持策略引用,运行时切换

# 优点:
# - 消除算法选择 if-else
# - 算法独立扩展(开闭原则)
# - 运行时动态切换
# - 算法可复用/可测试

# 缺点:
# - 类数量增加
# - 客户端需了解策略差异

结构

# 参与者:
# Strategy: 策略接口
# ConcreteStrategy: 具体策略(算法)
# Context: 上下文(持策略,委托执行)

# Client: 选择策略注入 Context

# 例: 排序算法
# Context(排序器) + Bubble/Quick/Merge 策略

Python 示例: 支付方式

from abc import ABC, abstractmethod

# 策略接口
class PaymentStrategy(ABC):
    @abstractmethod
    def pay(self, amount) -> str: ...

# 具体策略
class AlipayStrategy(PaymentStrategy):
    def pay(self, amount):
        return f"支付宝支付 {amount} 元"

class WechatStrategy(PaymentStrategy):
    def pay(self, amount):
        return f"微信支付 {amount} 元"

class BankCardStrategy(PaymentStrategy):
    def pay(self, amount):
        return f"银行卡支付 {amount} 元"

# 上下文(持策略)
class ShoppingCart:
    def __init__(self, strategy=None):
        self.items = []
        self.strategy = strategy

    def add_item(self, name, price):
        self.items.append((name, price))

    def set_strategy(self, strategy):
        self.strategy = strategy

    def checkout(self):
        total = sum(price for _, price in self.items)
        if not self.strategy:
            raise ValueError("请选择支付方式")
        return self.strategy.pay(total)

# 使用: 运行时切换策略
cart = ShoppingCart()
cart.add_item("手机", 3999)
cart.add_item("耳机", 599)

# 选择支付方式
cart.set_strategy(AlipayStrategy())
print(cart.checkout())   # 支付宝支付 4598 元

# 换策略(无需改购物车代码!)
cart.set_strategy(WechatStrategy())
print(cart.checkout())   # 微信支付 4598 元

cart.set_strategy(BankCardStrategy())
print(cart.checkout())   # 银行卡支付 4598 元

# 新增支付方式: 加一个策略类即可
class CryptoStrategy(PaymentStrategy):
    def pay(self, amount):
        return f"加密货币支付 {amount} 元"

cart.set_strategy(CryptoStrategy())
print(cart.checkout())

Python 简化: 函数策略

# 策略 = 函数(简单场景)
from typing import Callable

def alipay(amount): return f"支付宝 {amount}"
def wechat(amount): return f"微信 {amount}"
def bankcard(amount): return f"银行卡 {amount}"

class Cart:
    def __init__(self, pay_func: Callable = None):
        self.pay_func = pay_func
        self.total = 0

    def checkout(self):
        return self.pay_func(self.total)

cart = Cart(alipay)
cart.total = 100
print(cart.checkout())   # 支付宝 100

cart.pay_func = wechat   # 换策略
print(cart.checkout())   # 微信 100

# 或字典选择
PAYMENTS = {
    "alipay": alipay,
    "wechat": wechat,
}
pay_func = PAYMENTS.get("alipay")
pay_func(100)

应用场景

# 1. 支付/配送/优惠策略
# 2. 排序/搜索算法选择
# 3. 压缩/加密算法
# 4. 校验规则(不同场景不同规则)
# 5. 价格计算(会员/促销)

# 框架应用:
# - 排序 key 函数
# sorted(items, key=len)
# - 校验器(Validator 模式)
# - Django 存储后端/缓存后端

# 与状态模式对比:
# 策略: 算法替换(客户端主动)
# 状态: 行为随状态(自动切换)
# 策略无状态转换,状态有

# 与模板方法对比:
# 策略: 整体算法替换
# 模板: 固定骨架,子类填步骤

优缺点总结

# 优点:
# ✅ 消除算法 if-else
# ✅ 开闭原则(加策略即可)
# ✅ 运行时切换
# ✅ 策略独立测试

# 缺点:
# ❌ 类数量增加
# ❌ 客户端需了解策略
# ❌ 策略间有隐性联系时注意

# 最佳实践:
# - 策略接口最小化
# - 简单策略用函数/dict
# - 策略工厂统一创建
# - 无状态策略可共享实例

# 决策:
# 算法多变/可选 → 策略
# 简单分支 → dict 映射
# 参数化行为 → 函数参数

策略模式让"算法"可插拔。掌握策略注入与切换,支付、排序、校验类场景灵活替换。

模板模式

模板模式(Template Method Pattern)

模板:在父类中定义算法的骨架(步骤顺序),将部分步骤延迟到子类实现。复用骨架,定制细节。

意图

# 问题: 多个流程结构相同,细节不同
# - 复制粘贴 → 修改一处忘改他处
# - 流程顺序必须一致

# 解决: 父类模板方法定义步骤骨架
# 子类只实现差异步骤(钩子)

# 优点:
# - 复用骨架(代码复用)
# - 保证流程一致性
# - 控制扩展点(钩子方法)

# 缺点:
# - 父类与子类耦合
# - 骨架修改影响所有子类

结构

# 参与者:
# AbstractClass: 抽象类(模板方法 + 抽象步骤)
# ConcreteClass: 具体子类(实现差异步骤)

# 模板方法: 固定调用顺序
# def template_method(self):
#     step1()
#     step2()   # 子类实现
#     step3()

# 钩子方法: 可选覆盖(默认空/默认行为)

Python 示例: 数据导入流程

from abc import ABC, abstractmethod

# 抽象类: 定义流程骨架
class DataImporter(ABC):
    # 模板方法(骨架,不可覆盖)
    def import_data(self, source):
        data = self.fetch(source)      # 1
        cleaned = self.clean(data)     # 2
        self.validate(cleaned)         # 3
        result = self.save(cleaned)    # 4
        self.notify(result)            # 5
        return result

    # 公共步骤(可直接实现)
    def fetch(self, source):
        print(f"读取: {source}")
        return [{"raw": source, "n": 1}]

    def notify(self, result):
        print(f"完成: {len(result)} 条")

    # 差异步骤(子类实现)
    @abstractmethod
    def clean(self, data): ...

    @abstractmethod
    def validate(self, data): ...

    @abstractmethod
    def save(self, data): ...

# 具体子类: CSV 导入
class CsvImporter(DataImporter):
    def clean(self, data):
        print("清洗 CSV 数据")
        return [{"n": d["n"] * 1.0} for d in data]

    def validate(self, data):
        print("校验 CSV: 无空值")
        assert all(d["n"] >= 0 for d in data)

    def save(self, data):
        print("保存到 MySQL")
        return data

# 具体子类: API 导入
class ApiImporter(DataImporter):
    def fetch(self, source):
        print(f"API 请求: {source}")
        return [{"n": i} for i in range(3)]

    def clean(self, data):
        print("清洗 API 数据")
        return [d for d in data if d["n"] != 0]

    def validate(self, data):
        print("校验 API: 限 5 条")
        assert len(data) <= 5

    def save(self, data):
        print("写入 MongoDB")
        return data

# 使用: 流程骨架一致,细节各异
CsvImporter().import_data("data.csv")
# 读取: data.csv
# 清洗 CSV 数据
# 校验 CSV: 无空值
# 保存到 MySQL
# 完成: 1 条

ApiImporter().import_data("https://api.x.com")
# API 请求: https://api.x.com
# 清洗 API 数据
# 校验 API: 限 5 条
# 写入 MongoDB
# 完成: 3 条

# 骨架固定: 顺序不可乱(父类控制)
# 差异: 每步实现不同(子类定制)

钩子方法

# 钩子: 可选覆盖(默认行为)
class ReportGenerator(ABC):
    def generate(self):
        header = self.get_header()
        body = self.get_body()
        footer = self.get_footer()
        return f"{header}\n{body}\n{footer}"

    # 公共默认实现(钩子)
    def get_header(self):
        return "=== 报告 ==="

    def get_footer(self):
        return "=== 结束 ==="

    @abstractmethod
    def get_body(self): ...

    # 是否包含页脚(钩子,可覆盖)
    def include_footer(self):
        return True

class SalesReport(ReportGenerator):
    def get_body(self):
        return "本月销售额: 100万"

    def get_footer(self):
        # 覆盖页脚
        return f"{super().get_footer()} | 内部机密"

class SummaryReport(ReportGenerator):
    def get_body(self):
        return "汇总: 正常"

    def include_footer(self):
        return False   # 不要页脚

print(SalesReport().generate())
print(SummaryReport().generate())

实际应用

# 1. 数据导入/导出流程
# 2. 报表生成(头/体/尾)
# 3. 框架生命周期(init/start/stop)
# 4. 测试用例(SetUp/Test/TearDown)
# 5. 构建流程(编译/测试/打包)

# 框架中的模板方法:
# - pytest 夹具(setup/teardown)
# - Django Form 校验(clean 系列)
# - 算法骨架(排序的对比器)

# 与策略模式对比:
# 模板: 骨架固定,子类填步骤(继承)
# 策略: 算法整体替换(组合)
# 模板适合"流程固定细节多变"

优缺点总结

# 优点:
# ✅ 复用流程骨架
# ✅ 保证流程一致
# ✅ 扩展点清晰(抽象/钩子)

# 缺点:
# ❌ 继承耦合
# ❌ 骨架修改影响全局
# ❌ 子类太多时层级深

# 最佳实践:
# - 模板方法控制步骤(不要子类覆盖)
# - 抽象步骤明确(abstractmethod)
# - 钩子给默认行为(可覆盖)
# - 骨架稳定后再抽象

# Python 注意:
# 模板方法不要被装饰器/子类意外覆盖
# 用 final(3.12)或约定保护骨架

模板方法固定"骨架"复用,子类定制"细节"。掌握抽象步骤与钩子,导入、报表、生命周期类流程统一有序。

访问者模式

访问者模式(Visitor Pattern)

访问者:在不修改元素类的前提下,为对象结构增加新操作。将"操作"与"数据结构"分离。

意图

# 问题: 对结构中的对象执行多种操作
# - 每加一个操作要改所有元素类
# - 操作逻辑分散在各元素中

# 解决: 操作封装为访问者
# 元素接受访问者,调用对应 visit 方法

# 优点:
# - 新增操作容易(加访问者)
# - 操作逻辑集中
# - 元素类稳定(不修改)

# 缺点:
# - 新增元素类型要改所有访问者
# - 访问者需了解元素内部
# - 双向分派实现复杂

# 适用: 结构稳定、操作多变

结构

# 参与者:
# Visitor: 访问者接口(visit(元素))
# ConcreteVisitor: 具体访问者(操作)
# Element: 元素接口(accept(访问者))
# ConcreteElement: 具体元素
# ObjectStructure: 对象结构(遍历)

# 例: 文件系统
# 文件/目录(结构稳定) + 大小/压缩/备份(操作多变)

Python 示例: 报表系统

from abc import ABC, abstractmethod

# 元素接口
class Employee(ABC):
    @abstractmethod
    def accept(self, visitor): ...

# 具体元素
class Developer(Employee):
    def __init__(self, name, salary, lines):
        self.name = name
        self.salary = salary
        self.code_lines = lines

    def accept(self, visitor):
        return visitor.visit_developer(self)

class Manager(Employee):
    def __init__(self, name, salary, reports):
        self.name = name
        self.salary = salary
        self.report_count = reports

    def accept(self, visitor):
        return visitor.visit_manager(self)

# 访问者接口
class Visitor(ABC):
    @abstractmethod
    def visit_developer(self, emp): ...
    @abstractmethod
    def visit_manager(self, emp): ...

# 具体访问者: 薪资报告
class SalaryReport(Visitor):
    def visit_developer(self, emp):
        return (f"开发 {emp.name}: "
                f"{emp.salary}元/月")

    def visit_manager(self, emp):
        return (f"经理 {emp.name}: "
                f"{emp.salary}元/月")

# 具体访问者: 绩效评分
class PerformanceVisitor(Visitor):
    def visit_developer(self, emp):
        score = min(emp.code_lines // 500, 10)
        return f"开发 {emp.name}: {score}分"

    def visit_manager(self, emp):
        score = min(emp.report_count * 2, 10)
        return f"经理 {emp.name}: {score}分"

# 使用: 新增操作 = 新增访问者,不改员工类
employees = [
    Developer("张三", 20000, 3000),
    Manager("李四", 30000, 4),
    Developer("王五", 18000, 1200),
]

for emp in employees:
    print(emp.accept(SalaryReport()))
    print(emp.accept(PerformanceVisitor()))

# 新增"培训需求"访问者...
class TrainingVisitor(Visitor):
    def visit_developer(self, emp):
        return f"{emp.name}: 需要 AI 课程"
    def visit_manager(self, emp):
        return f"{emp.name}: 需要管理课"

# 元素类零修改,操作无限扩展!

Python 简化: 单分派/模式匹配

# Python 可用模式匹配简化(3.10+)
from dataclasses import dataclass

@dataclass
class Dev:
    name: str
    salary: float

@dataclass
class Mgr:
    name: str
    salary: float

# 操作函数(无访问者类)
def salary_report(emp):
    match emp:
        case Dev(name=n, salary=s):
            return f"开发 {n}: {s}"
        case Mgr(name=n, salary=s):
            return f"经理 {n}: {s}"
        case _:
            return "未知"

# 或单分派
from functools import singledispatch

@singledispatch
def report(emp):
    return f"未知 {emp}"

@report.register
def _(emp: Dev):
    return f"开发 {emp.name}: {emp.salary}"

@report.register
def _(emp: Mgr):
    return f"经理 {emp.name}: {emp.salary}"

print(report(Dev("张三", 20000)))
print(report(Mgr("李四", 30000)))

# 新增类型: 注册新函数,元素类零修改
# 简单场景用 singledispatch 更 Pythonic

实际应用

# 1. 编译器: AST 分析(语义检查/优化)
# 2. 报表: 对不同实体生成不同报表
# 3. 序列化: 对不同对象序列化
# 4. 统计: 遍历结构做各类统计
# 5. 文档导出: HTML/PDF/JSON 导出器

# 条件:
# ✅ 对象结构稳定(少变)
# ✅ 操作频繁增加
# ❌ 结构常变 → 不要用访问者
# ❌ 操作少 → 普通方法即可

# 注意:
# 访问者模式在 Python 中较少必要
# singledispatch / match 更简洁
# 大型类型层次时访问者有优势

优缺点总结

# 优点:
# ✅ 新增操作容易
# ✅ 操作逻辑集中
# ✅ 元素类稳定

# 缺点:
# ❌ 新增元素类型需改所有访问者
# ❌ 破坏封装(访问者需了解内部)
# ❌ 双向分派复杂

# Python 决策:
# 简单层次 → singledispatch
# 结构稳定+操作多变 → 访问者
# 复杂类型系统 → 考虑访问者或 match

# 反模式警示:
# 元素类型频繁变化时,访问者
# 会导致大规模修改,避免使用

访问者把"操作"从"结构"中分离。掌握 accept/visit 双向分派,结构稳定而操作多变的场景游刃有余。

MVC 模式

MVC 模式(Model-View-Controller)

MVC 是经典的架构模式:将应用分为模型(Model)、视图(View)、控制器(Controller)三层,实现关注点分离。

意图

# 问题: 业务逻辑、数据、界面混杂
# - 代码难以维护
# - 修改界面影响业务
# - 难以测试

# 解决: 三层分离
# Model: 数据与业务逻辑
# View: 界面展示
# Controller: 处理输入,协调 Model 与 View

# 优点:
# - 关注点分离
# - 可独立修改与测试
# - 代码复用(多 View 共享 Model)

# 缺点:
# - 学习曲线
# - 简单应用过度分层

结构

# 请求流:
# 用户 → Controller(处理输入)
#              ↓ 更新
#           Model(数据/业务)
#              ↓ 通知
#           View(展示)

# 依赖方向:
# Controller → Model + View
# View → Model(只读)
# Model 不依赖 View/Controller

Python 示例: 任务管理

# Model: 数据与业务
class TaskModel:
    def __init__(self):
        self.tasks = []
        self._observers = []

    def add(self, title):
        self.tasks.append(
            {"title": title, "done": False})
        self.notify()

    def toggle(self, index):
        self.tasks[index]["done"] =             not self.tasks[index]["done"]
        self.notify()

    def get_tasks(self):
        return self.tasks

    def attach(self, observer):
        self._observers.append(observer)

    def notify(self):
        for obs in self._observers:
            obs.update(self.tasks)

# View: 展示
class TaskView:
    def update(self, tasks):
        print("\n=== 任务列表 ===")
        for i, t in enumerate(tasks):
            status = "✓" if t["done"] else "□"
            print(f"{i}. [{status}] {t['title']}")

# Controller: 输入处理
class TaskController:
    def __init__(self, model, view):
        self.model = model
        self.view = view
        model.attach(view)   # 数据变化 → 视图更新

    def add_task(self, title):
        self.model.add(title)

    def complete_task(self, index):
        self.model.toggle(index)

# 使用
model = TaskModel()
view = TaskView()
controller = TaskController(model, view)

controller.add_task("学习 Python")
controller.add_task("写报告")
# === 任务列表 ===
# 0. [□] 学习 Python
# 1. [□] 写报告

controller.complete_task(0)
# === 任务列表 ===
# 0. [✓] 学习 Python
# 1. [□] 写报告

# 数据与展示分离:
# - 换 View(网页/终端/APP)不影响 Model
# - Model 可独立单元测试
# - Controller 可单独测输入逻辑

框架中的 MVC

# Django: MVT(视图=Controller)
# Model → ORM 模型
# View(函数/类) → 控制器逻辑
# Template → 视图层

# Flask/FastAPI: MVC 风格
# 路由函数 = Controller
# SQLAlchemy 模型 = Model
# 模板/JSON 响应 = View

# 前端框架:
# Angular/React/Vue = MVVM(变体)
# ViewModel 绑定视图与模型

# 对比:
# MVC: Controller 处理输入
# MVVM: ViewModel 自动绑定(数据驱动)

# 命令/事件驱动:
# 简单 MVC 可以不用观察者
# 用 Controller 手动刷新 View 也行

MVC 变体

# MVP(Model-View-Presenter):
# Presenter 完全控制 View(接口)
# View 被动,适合单元测试
# Android 常用

# MVVM(Model-View-ViewModel):
# ViewModel 双向绑定
# View 声明式(绑定语法)
# WPF/Angular/Vue

# 选择:
# 简单应用 → 分层即可(MVC 简化)
# 需要强测试 → MVP
# 数据绑定框架 → MVVM

# 架构层次:
# MVC 是"模式",微服务是"架构"
# MVC 可在微服务内部使用

优缺点总结

# 优点:
# ✅ 关注点分离
# ✅ 可测试性
# ✅ 复用与并行开发
# ✅ 维护性

# 缺点:
# ❌ 学习成本
# ❌ 简单应用过度设计
# ❌ Controller 可能臃肿

# 最佳实践:
# - Model 纯业务(不依赖 UI)
# - Controller 薄(只编排)
# - View 只展示(无业务)
# - 数据流单向(清晰可追踪)
# - 简单 CRUD 用框架默认分层

# Python 建议:
# - 小工具: 函数式分层即可
# - Web: 用框架自带结构(不硬套)
# - 桌面: MVVM 更适合

MVC 是软件架构的"地基"。理解三层职责与数据流,无论 Web/桌面/移动,项目结构清晰可维护。

业务代表模式

业务代表模式(Business Delegate Pattern)

业务代表:客户端与业务服务之间的中间层,隐藏服务查找与访问细节,减少客户端与业务层耦合。

意图

# 问题: 客户端直接访问业务服务
# - 需要知道服务查找逻辑(JNDI/注册)
# - 服务地址/接口变化影响客户端

# 解决: 业务代表封装查找与调用
# 客户端只与代表交互

# 优点:
# - 解耦客户端与服务
# - 隐藏服务查找
# - 统一访问点(缓存/降级)

# 缺点:
# - 增加一层
# - 简单应用过度设计

结构

# 参与者:
# Client: 客户端(调代表)
# BusinessDelegate: 业务代表(封装)
# BusinessService: 业务服务接口
# ServiceLookup: 服务查找(定位实现)
# ConcreteService: 具体服务

# 例: 客户端下单 → 代表 → 查找并调用订单服务

Python 示例: 业务服务访问

from abc import ABC, abstractmethod

# 业务服务接口
class BusinessService(ABC):
    @abstractmethod
    def do_processing(self) -> str: ...

# 具体服务
class OrderService(BusinessService):
    def do_processing(self):
        return "订单服务处理中"

class PaymentService(BusinessService):
    def do_processing(self):
        return "支付服务处理中"

# 服务查找
class ServiceLookup:
    def __init__(self):
        self._services = {
            "order": OrderService(),
            "payment": PaymentService(),
        }

    def get_service(self, name) -> BusinessService:
        service = self._services.get(name)
        if not service:
            raise ValueError(f"未知服务: {name}")
        return service

# 业务代表
class BusinessDelegate:
    def __init__(self):
        self.lookup = ServiceLookup()
        self.service_type = "order"

    def set_service(self, service_type):
        self.service_type = service_type

    def do_task(self) -> str:
        service = self.lookup.get_service(
            self.service_type)
        # 可加缓存/日志/重试
        return service.do_processing()

# 客户端: 只认识代表
class Client:
    def __init__(self, delegate):
        self.delegate = delegate

    def run(self):
        return self.delegate.do_task()

delegate = BusinessDelegate()
client = Client(delegate)

delegate.set_service("order")
print(client.run())      # 订单服务处理中

delegate.set_service("payment")
print(client.run())      # 支付服务处理中

# 客户端不知道服务如何查找/实例化
# 换服务实现不影响客户端

# 实际应用:
# - RPC/微服务客户端封装
# - 业务网关(统一入口)
# - 服务代理(缓存/降级/限流)

Python 简化

# 1. 函数式代表
SERVICES = {
    "order": lambda: "订单处理",
    "payment": lambda: "支付处理",
}

def business_delegate(name):
    # 封装查找与调用
    service = SERVICES.get(name)
    if not service:
        raise ValueError(f"未知: {name}")
    return service()

print(business_delegate("order"))

# 2. 依赖注入(现代替代)
# 客户端直接注入服务(无需代表)
class OrderService:
    def process(self): return "订单"

class App:
    def __init__(self, order_service):
        self.orders = order_service

app = App(OrderService())
app.orders.process()

# 3. 服务定位器(见 J2EE 模式章节)
# 与代表区别: 代表封装"使用",定位器封装"查找"

优缺点总结

# 优点:
# ✅ 解耦客户端与服务
# ✅ 隐藏查找细节
# ✅ 统一访问点(可加横切逻辑)
# ✅ 服务变更影响小

# 缺点:
# ❌ 增加一层
# ❌ 简单应用过度设计

# 最佳实践:
# - 代表保持薄(只转发)
# - 横切逻辑(日志/重试/缓存)放代表
# - 现代项目用依赖注入/服务层
# - 微服务场景用 API 网关

# 决策:
# 客户端多且服务查找复杂 → 业务代表
# 现代 Python → 依赖注入
# 分布式 → 网关/服务发现

业务代表是"客户端与服务的中间人"。掌握查找封装与统一访问,服务调用层解耦灵活。

组合实体模式

组合实体模式(Composite Entity Pattern)

组合实体:用粗粒度实体(聚合对象)表示一组细粒度对象,简化持久化与客户端交互。

意图

# 问题: 业务对象由多个相关对象组成
# - 客户端需分别操作多个对象
# - 持久化时多次读写

# 解决: 组合实体聚合多个"依赖对象"
# 客户端操作组合实体,统一读写

# 优点:
# - 简化客户端(一个对象)
# - 统一事务边界
# - 依赖对象自动管理

# 缺点:
# - 粒度粗,更新可能过度
# - 组合对象大时内存开销

结构

# 参与者:
# CompositeEntity: 组合实体(聚合依赖)
# CoarseGrainedObject: 粗粒度对象(生命周期)
# DependentObject: 依赖对象(细粒度)
# Client: 客户端(操作组合实体)

# 例: 用户资料
# UserEntity(组合) → Profile + Address

Python 示例: 用户资料

# 依赖对象(细粒度)
class ProfileData:
    def __init__(self):
        self.nickname = ""
        self.email = ""

    def set_data(self, nickname, email):
        self.nickname = nickname
        self.email = email

    def get_data(self):
        return (self.nickname, self.email)

class AddressData:
    def __init__(self):
        self.city = ""
        self.street = ""

    def set_data(self, city, street):
        self.city = city
        self.street = street

    def get_data(self):
        return (self.city, self.street)

# 粗粒度对象: 管理依赖对象生命周期
class UserCoarseGrained:
    def __init__(self):
        self.profile = ProfileData()
        self.address = AddressData()

    def set_data(self, nickname, email,
                 city, street):
        self.profile.set_data(nickname, email)
        self.address.set_data(city, street)

    def get_data(self):
        return {
            "nickname": self.profile.nickname,
            "email": self.profile.email,
            "city": self.address.city,
            "street": self.address.street,
        }

# 组合实体: 对外统一接口
class UserCompositeEntity:
    def __init__(self):
        self.coarse = UserCoarseGrained()

    def set_data(self, nickname, email,
                 city, street):
        # 统一设置(内部协调多个依赖)
        self.coarse.set_data(
            nickname, email, city, street)

    def get_data(self):
        # 统一读取
        return self.coarse.get_data()

    # 模拟持久化
    def save(self):
        data = self.get_data()
        print(f"保存用户: {data}")

# 客户端: 只与组合实体交互
class Client:
    def __init__(self):
        self.entity = UserCompositeEntity()

    def update_user(self, info):
        self.entity.set_data(
            info["nickname"], info["email"],
            info["city"], info["street"])
        self.entity.save()

client = Client()
client.update_user({
    "nickname": "小王",
    "email": "w@e.com",
    "city": "北京",
    "street": "长安街1号",
})
# 保存用户: {'nickname': '小王', ...}

# 客户端无需关心 Profile/Address 内部
# 依赖对象可独立复用,也可被组合

实际应用

# 1. 用户资料聚合(基本信息+地址+偏好)
# 2. 订单聚合(订单+明细+收货信息)
# 3. 文档聚合(内容+样式+元数据)
# 4. 表单聚合(多区块表单统一提交)

# 现代对应: 聚合根(DDD)
# 领域驱动设计中的 Aggregate Root:
# 聚合根统一管理内部实体
# 事务边界 = 聚合边界

# 对比:
# 组合实体: J2EE 经典模式
# 聚合根: DDD 现代演进
# 核心一致: 粗粒度对象管理细粒度

# Python 中常表现为:
# - dataclass 组合
# - Pydantic 嵌套模型
# - ORM 关系(joined 加载)

Python 简化: dataclass/Pydantic

from dataclasses import dataclass, field

@dataclass
class Profile:
    nickname: str = ""
    email: str = ""

@dataclass
class Address:
    city: str = ""
    street: str = ""

@dataclass
class User:
    # 组合实体(聚合)
    profile: Profile = field(default_factory=Profile)
    address: Address = field(default_factory=Address)

    def to_dict(self):
        return {
            "nickname": self.profile.nickname,
            "email": self.profile.email,
            "city": self.address.city,
            "street": self.address.street,
        }

user = User()
user.profile.nickname = "小王"
user.address.city = "北京"
print(user.to_dict())

# 或 Pydantic(校验+序列化)
# from pydantic import BaseModel
# class UserModel(BaseModel):
#     profile: Profile
#     address: Address

优缺点总结

# 优点:
# ✅ 简化客户端
# ✅ 统一事务/持久化
# ✅ 依赖对象自动管理
# ✅ 结构清晰

# 缺点:
# ❌ 粒度粗(更新过度)
# ❌ 大对象内存开销
# ❌ 依赖对象复用受限

# 最佳实践:
# - 组合实体边界清晰(内聚)
# - 依赖对象可测试
# - 按业务聚合(非任意组合)
# - 现代用 dataclass/Pydantic/DDD 聚合

# 决策:
# 多对象统一操作 → 组合实体
# 简单数据 → dataclass
# 复杂业务 → DDD 聚合根

组合实体用"粗粒度"简化"细粒度"。掌握聚合与统一接口,用户、订单、文档类复合对象管理简洁。

数据访问对象模式

数据访问对象模式(DAO Pattern)

DAO(Data Access Object):将数据访问逻辑(数据库/API/文件)封装为独立对象,业务层不直接接触数据源细节。

意图

# 问题: 业务代码直接写 SQL/ORM 调用
# - 数据源变化(MySQL→PostgreSQL)改业务
# - 查询逻辑分散,难复用/测试

# 解决: DAO 集中数据访问
# 业务层只依赖 DAO 接口

# 优点:
# - 业务与数据源解耦
# - 数据访问复用与测试
# - 换数据源只改 DAO

# 缺点:
# - 增加抽象层
# - 简单 CRUD 略显繁琐

结构

# 参与者:
# DAO: 数据访问接口
# ConcreteDAO: 具体实现(MySQL/API/内存)
# Model/Entity: 数据对象
# Client(业务层): 使用 DAO

# 例: UserDAO
# 接口: get/find/add/update/delete
# 实现: MySQLDAO / MockDAO / APIDAO

Python 示例: 用户 DAO

from abc import ABC, abstractmethod
from dataclasses import dataclass

# 数据对象
@dataclass
class User:
    id: int
    name: str
    email: str

# DAO 接口
class UserDAO(ABC):
    @abstractmethod
    def get(self, user_id: int) -> User: ...
    @abstractmethod
    def find_all(self) -> list[User]: ...
    @abstractmethod
    def add(self, user: User) -> User: ...
    @abstractmethod
    def update(self, user: User) -> User: ...
    @abstractmethod
    def delete(self, user_id: int) -> bool: ...

# 内存实现(测试/演示用)
class InMemoryUserDAO(UserDAO):
    def __init__(self):
        self._data = {}
        self._next_id = 1

    def get(self, user_id):
        return self._data.get(user_id)

    def find_all(self):
        return list(self._data.values())

    def add(self, user):
        user.id = self._next_id
        self._next_id += 1
        self._data[user.id] = user
        return user

    def update(self, user):
        if user.id not in self._data:
            return None
        self._data[user.id] = user
        return user

    def delete(self, user_id):
        return self._data.pop(user_id, None) is not None

# 业务层(服务): 只依赖 DAO 接口
class UserService:
    def __init__(self, dao: UserDAO):
        self.dao = dao

    def register(self, name, email):
        # 业务逻辑 + DAO 调用
        if not email or "@" not in email:
            raise ValueError("邮箱无效")
        return self.dao.add(User(0, name, email))

    def get_profile(self, user_id):
        user = self.dao.get(user_id)
        if not user:
            raise ValueError("用户不存在")
        return user

    def delete_account(self, user_id):
        return self.dao.delete(user_id)

# 使用: 注入任意 DAO 实现
service = UserService(InMemoryUserDAO())

user = service.register("张三", "z@e.com")
print(user.id, user.name)     # 1 张三

print(service.get_profile(1).email)   # z@e.com
service.delete_account(1)

# 换数据源: 实现 MySQLDAO 注入即可
# 业务层零修改!

MySQL 实现

import pymysql

class MysqlUserDAO(UserDAO):
    def __init__(self, conn):
        self.conn = conn

    def find_all(self):
        with self.conn.cursor() as cur:
            cur.execute(
                "SELECT id, name, email FROM users")
            rows = cur.fetchall()
        return [User(*row) for row in rows]

    def add(self, user):
        with self.conn.cursor() as cur:
            cur.execute(
                "INSERT INTO users(name, email) "
                "VALUES(%s, %s)",
                (user.name, user.email))
            user.id = cur.lastrowid
        self.conn.commit()
        return user

    def delete(self, user_id):
        with self.conn.cursor() as cur:
            n = cur.execute(
                "DELETE FROM users WHERE id=%s",
                (user_id,))
        self.conn.commit()
        return n > 0

# 参数化查询防注入!
# 业务层不变,只换 DAO
# service = UserService(MysqlUserDAO(conn))

Python 生态: ORM 即 DAO

# SQLAlchemy(声明式 DAO)
from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session, declarative_base
from sqlalchemy import Column, Integer, String

Base = declarative_base()

class UserORM(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    email = Column(String(100))

# 使用(Session 类似 DAO)
engine = create_engine("sqlite:///app.db")
Base.metadata.create_all(engine)

with Session(engine) as session:
    session.add(UserORM(name="张三", email="z@e.com"))
    session.commit()

    users = session.scalars(
        select(UserORM)).all()

# 仓储模式(Repository, DDD):
# DAO 聚焦数据访问
# Repository 更聚焦领域聚合
# 现代 Python 用 ORM + 服务层即可

实际应用

# 1. 数据库访问封装(DAO)
# 2. 第三方 API 封装(远程数据)
# 3. 缓存访问封装
# 4. 测试替身(内存/模拟 DAO)

# 分层:
# Controller → Service(业务) → DAO → DB
# 业务层不出现 SQL/ORM 调用

# 好处:
# - 单元测试用 Mock DAO(快)
# - 换库/换 API 影响小
# - 查询优化集中一处

# 反模式:
# 业务逻辑写在 DAO 里(DAO 应纯数据)
# 每表一个 DAO 而无业务抽象
# 过度封装简单 CRUD

优缺点总结

# 优点:
# ✅ 数据源解耦
# ✅ 复用与测试
# ✅ 集中管理(连接/事务/缓存)
# ✅ 参数化查询(安全)

# 缺点:
# ❌ 抽象层成本
# ❌ 简单 CRUD 繁琐

# 最佳实践:
# - DAO 纯数据访问(无业务)
# - 接口抽象(可换实现)
# - 参数化查询防注入
# - 事务边界清晰
# - 现代用 ORM + Repository

# 决策:
# 多数据源/需要测试 → DAO 接口
# 单数据源简单 → ORM 直接用
# 大型项目 → 分层 + 仓储

DAO 让"数据访问"可替换、可测试。掌握接口抽象与实现分离,数据层稳定,业务层清爽。

前端控制器模式

前端控制器模式(Front Controller)

前端控制器:所有请求先经过统一入口(控制器),在入口完成公共处理(认证/日志/路由),再分发到具体处理器。

意图

# 问题: 每个请求处理都要重复公共逻辑
# - 认证/日志/权限散落各处
# - 路由管理分散

# 解决: 统一前端控制器
# 所有请求 → 前端控制器 → 公共处理 → 分发

# 优点:
# - 公共逻辑集中(一处修改全局生效)
# - 统一入口(安全/审计)
# - 路由集中管理

# 缺点:
# - 单点故障/性能瓶颈
# - 控制器可能臃肿

结构

# 参与者:
# FrontController: 前端控制器(入口)
# Dispatcher: 分发器(路由到处理者)
# Handler/View: 具体处理者

# 例: Web 框架路由
# 请求 → 中间件(公共) → 路由 → 视图函数

Python 示例: Web 请求

# 处理器
class HomeHandler:
    def get(self):
        return "首页"

class UserHandler:
    def get(self, user_id):
        return f"用户 {user_id}"

class NotFoundHandler:
    def get(self):
        return "404 页面", 404

# 分发器(路由)
class Dispatcher:
    def __init__(self):
        self.routes = {
            "/": HomeHandler(),
            "/user": UserHandler(),
        }

    def dispatch(self, path):
        handler = self.routes.get(path)
        if not handler:
            return NotFoundHandler().get()
        return handler.get()

# 前端控制器(统一入口)
class FrontController:
    def __init__(self):
        self.dispatcher = Dispatcher()

    # 公共处理: 认证
    def _auth(self, token):
        return token == "valid-token"

    # 公共处理: 日志
    def _log(self, path):
        print(f"[访问] {path}")

    def handle(self, path, token):
        # 1. 公共逻辑(认证)
        if not self._auth(token):
            return "401 未授权", 401

        # 2. 公共逻辑(日志)
        self._log(path)

        # 3. 分发
        return self.dispatcher.dispatch(path)

# 使用: 客户端只面对一个入口
controller = FrontController()

print(controller.handle("/", "invalid"))
# (401 未授权, 401)

print(controller.handle("/", "valid-token"))
# [访问] /
# 首页

print(controller.handle("/user", "valid-token"))
# [访问] /user
# 用户

# 新增页面: 加路由即可
# 公共逻辑(认证/日志)统一,零重复!

框架中的前端控制器

# Flask: 请求钩子
@app.before_request
def auth():
    if not check_token():
        return "401", 401

@app.before_request
def logging():
    print(f"访问: {request.path}")

# 所有视图自动经过钩子(前端控制器思想)

# FastAPI: 中间件
from starlette.middleware.base import BaseHTTPMiddleware

class AuthMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        if not valid(request.headers.get("token")):
            return JSONResponse({"err": "401"}, 401)
        return await call_next(request)

app.add_middleware(AuthMiddleware)

# Django: MIDDLEWARE 列表
# 认证/日志/CSRF 都是"前端控制器"的一环

# 对比:
# 前端控制器 = 手动实现中间件
# 现代框架用中间件(更解耦)

手动实现 vs 中间件

# 前端控制器:
# 优点: 集中、直观
# 缺点: 需自己维护分发

# 中间件(现代推荐):
# 优点: 可插拔、顺序控制
# 缺点: 抽象、需理解执行顺序

# 选择:
# 简单应用 → 前端控制器(直观)
# 框架项目 → 中间件(标准)
# 需要灵活组合 → 中间件

# 公共处理清单(集中放):
# - 认证/授权
# - 日志/审计
# - 限流/防抖
# - 请求日志(耗时)
# - 统一错误处理
# - 国际化/头处理

优缺点总结

# 优点:
# ✅ 公共逻辑集中
# ✅ 统一入口(安全)
# ✅ 路由集中管理
# ✅ 易于添加横切功能

# 缺点:
# ❌ 单点瓶颈
# ❌ 控制器臃肿风险
# ❌ 需维护分发逻辑

# 最佳实践:
# - 公共逻辑放控制器/中间件
# - 控制器薄(只编排,不写业务)
# - 按需拆分中间件链
# - 现代框架用中间件机制

# 决策:
# 框架项目 → 中间件
# 自研小型服务 → 前端控制器
# 需要统一入口 → 网关(微服务)

前端控制器是"统一入口"的经典方案。掌握公共处理与分发,认证、日志、路由统一管理,现代框架里就是中间件。

拦截过滤器模式

拦截过滤器模式(Intercepting Filter)

拦截过滤器:在请求处理前后插入一系列过滤器,执行公共横切关注点(认证、日志、压缩、加密),且可按需组合。

意图

# 问题: 请求处理需要附加处理
# - 认证/日志/压缩等横切逻辑
# - 顺序与组合需要灵活配置

# 解决: 过滤器链
# 请求 → 过滤器1 → 过滤器2 → 目标处理

# 与前端控制器区别:
# 前端控制器: 统一入口(集中)
# 拦截过滤器: 可插拔处理链(灵活)

# 优点:
# - 横切关注点分离
# - 可插拔/可组合
# - 顺序可控
# - 复用过滤器

# 缺点:
# - 链管理复杂
# - 顺序错误影响功能

结构

# 参与者:
# Filter: 过滤器接口
# ConcreteFilter: 具体过滤器
# FilterChain: 过滤器链(管理顺序)
# Target: 目标处理器
# FilterManager: 过滤器管理器(组装)

# 例: Web 请求
# 认证 → 日志 → 压缩 → 业务

Python 示例: 请求过滤链

from abc import ABC, abstractmethod

# 过滤器接口
class Filter(ABC):
    @abstractmethod
    def execute(self, request) -> bool: ...
    # 返回 False 中断链

# 具体过滤器
class AuthFilter(Filter):
    def execute(self, request):
        if not request.get("token"):
            print("认证失败,拦截")
            return False
        print("认证通过")
        return True

class LogFilter(Filter):
    def execute(self, request):
        print(f"日志: {request.get('path')}")
        return True

class RateLimitFilter(Filter):
    def __init__(self, limit=100):
        self.limit = limit
        self.count = 0
    def execute(self, request):
        self.count += 1
        if self.count > self.limit:
            print("限流拦截")
            return False
        return True

# 过滤器链
class FilterChain:
    def __init__(self):
        self.filters = []

    def add(self, filter: Filter):
        self.filters.append(filter)
        return self

    def execute(self, request) -> bool:
        for f in self.filters:
            if not f.execute(request):
                return False   # 中断
        return True

# 目标处理
class Target:
    def execute(self, request):
        return f"执行业务: {request.get('path')}"

# 管理器
class FilterManager:
    def __init__(self, target: Target):
        self.chain = FilterChain()
        self.target = target

    def add_filter(self, f: Filter):
        self.chain.add(f)
        return self

    def process(self, request):
        if self.chain.execute(request):
            return self.target.execute(request)
        return "请求被拦截"

# 使用: 灵活组装过滤器
manager = FilterManager(Target())
manager.add_filter(AuthFilter())
manager.add_filter(RateLimitFilter(limit=2))
manager.add_filter(LogFilter())

print(manager.process({"path": "/api", "token": None}))
# 认证失败,拦截
# 请求被拦截

print(manager.process({"path": "/api", "token": "ok"}))
# 认证通过
# 日志: /api
# 执行业务: /api

# 换组合(去掉限流,换顺序)
manager2 = FilterManager(Target())
manager2.add_filter(LogFilter())
manager2.add_filter(AuthFilter())
print(manager2.process({"path": "/x", "token": "ok"}))
# 日志: /x(先日志)
# 认证通过
# 执行业务: /x

# 过滤器可插拔、可复用、顺序可配!

框架中的过滤器

# Django 中间件(过滤器思想)
MIDDLEWARE = [
    "django.middleware.security.SecurityMiddleware",
    "django.contrib.auth.middleware.AuthenticationMiddleware",
    "django.middleware.common.CommonMiddleware",
]

# Flask 钩子
@app.before_request   # 认证/限流
@app.after_request    # 压缩/头处理

# FastAPI 中间件
@app.middleware("http")
async def add_header(request, call_next):
    response = await call_next(request)
    response.headers["X-Custom"] = "1"
    return response

# 与责任链模式的关系:
# 拦截过滤器 = 责任链 + 横切过滤
# 责任链: 处理者依次尝试
# 过滤器: 前置/后置处理(不替代目标)

常用过滤器

# 前置过滤器(请求前):
# - 认证/授权
# - 限流/防爬
# - 参数校验
# - 请求日志
# - 解密/解压

# 后置过滤器(响应后):
# - 响应日志/耗时
# - 压缩/加密
# - 头注入(CORS/缓存)
# - 统一错误格式

# 实现要点:
# - 过滤器无状态(可复用)
# - 失败返回明确结果
# - 顺序敏感(认证在前)
# - 可配置(白名单/开关)

优缺点总结

# 优点:
# ✅ 横切逻辑分离
# ✅ 可插拔组合
# ✅ 顺序可控
# ✅ 复用过滤器

# 缺点:
# ❌ 链管理复杂
# ❌ 顺序错误难排查
# ❌ 过滤器间隐式依赖

# 最佳实践:
# - 过滤器单一职责
# - 顺序文档化/配置化
# - 提供跳过/白名单机制
# - 日志记录链的执行
# - 框架中用中间件机制

# 决策:
# 框架项目 → 中间件
# 自定义管道 → 过滤器链
# 复杂规则 → 责任链+过滤器

拦截过滤器让"横切关注点"可插拔。掌握链的组装与顺序,认证、日志、限流如积木般灵活组合。

服务定位器模式

服务定位器模式(Service Locator)

服务定位器:集中注册与查找服务,客户端通过定位器获取服务实例,减少直接依赖(服务注册表)。

意图

# 问题: 客户端直接创建/查找服务
# - 服务依赖关系分散
# - 换实现要改多处

# 解决: 服务定位器注册所有服务
# 客户端向定位器请求服务

# 优点:
# - 集中管理服务(注册/生命周期)
# - 客户端解耦(只需定位器)
# - 懒加载(需要时创建)

# 缺点:
# - 隐藏依赖(服务变全局状态)
# - 难测试(需重置定位器)
# - 与依赖注入相比不透明

# 现代建议:
# 依赖注入(DI)通常更优
# 服务定位器用于特殊情况

结构

# 参与者:
# ServiceLocator: 服务定位器(注册+查找)
# Service: 服务接口
# ConcreteService: 具体服务
# Client: 客户端(通过定位器获取服务)

# 例:
# locator.get("payment") → 支付服务实例

Python 示例

from abc import ABC, abstractmethod

# 服务接口
class NotificationService(ABC):
    @abstractmethod
    def send(self, msg): ...

class EmailService(NotificationService):
    def send(self, msg):
        return f"[邮件] {msg}"

class SmsService(NotificationService):
    def send(self, msg):
        return f"[短信] {msg}"

# 服务定位器
class ServiceLocator:
    _services = {}
    _initialized = False

    @classmethod
    def init(cls):
        # 注册服务(集中配置)
        cls._services = {
            "email": EmailService(),
            "sms": SmsService(),
        }
        cls._initialized = True

    @classmethod
    def get(cls, name):
        if not cls._initialized:
            cls.init()
        service = cls._services.get(name)
        if not service:
            raise ValueError(f"服务未注册: {name}")
        return service

    @classmethod
    def register(cls, name, service):
        cls._services[name] = service

    @classmethod
    def reset(cls):
        cls._services = {}
        cls._initialized = False

# 客户端: 通过定位器获取服务
class OrderService:
    def notify(self, user, message):
        # 从定位器获取服务(客户端不直接创建)
        email = ServiceLocator.get("email")
        return email.send(f"{user}: {message}")

# 使用
order_svc = OrderService()
print(order_svc.notify("张三", "订单已发货"))
# [邮件] 张三: 订单已发货

# 换实现/加服务: 改定位器注册,客户端零改动
ServiceLocator.register("sms", SmsService())
print(ServiceLocator.get("sms").send("促销"))
# [短信] 促销

# 懒加载版本
class LazyServiceLocator:
    _factories = {}
    _cache = {}

    @classmethod
    def register(cls, name, factory):
        cls._factories[name] = factory

    @classmethod
    def get(cls, name):
        if name not in cls._cache:
            cls._cache[name] = cls._factories[name]()
        return cls._cache[name]

LazyServiceLocator.register(
    "email", EmailService)
svc = LazyServiceLocator.get("email")  # 首次创建
svc2 = LazyServiceLocator.get("email") # 复用

服务定位器 vs 依赖注入

# 服务定位器(SL):
# 客户端主动调用: locator.get("svc")
# 依赖隐藏在客户端代码中
# 测试需 reset 定位器

# 依赖注入(DI):
# 客户端声明: __init__(self, svc)
# 依赖显式可见(构造参数)
# 测试直接注入 mock

# 对比:
# DI: 显式、可测试、推荐
# SL: 隐式、集中、特殊场景

# 何时用 SL(慎用):
# - 遗留系统集成
# - 插件体系(运行时注册)
# - 全局框架服务(日志/配置)
# - 无法使用 DI 的框架

# Python 生态:
# 优先: 构造注入 / 模块级对象
# 库: dependency-injector(现代 DI)
# 简单: 模块注册表

Python 简化

# 1. 模块级注册表
# services.py
_services = {}

def register(name, service):
    _services[name] = service

def get(name):
    return _services[name]

# 2. 模块级单例(最简单)
# email_service.py
service = EmailService()
# 直接 import 使用(服务定位的极端简化)

# 3. 依赖注入(推荐)
class OrderService:
    def __init__(self, notifier):
        self.notifier = notifier  # 注入

# 测试:
order_svc = OrderService(FakeNotifier())

# 4. dependency-injector(生产级)
# from dependency_injector import containers, providers
# container = Containers()
# container.email_service()

优缺点总结

# 优点:
# ✅ 集中服务管理
# ✅ 客户端解耦
# ✅ 懒加载
# ✅ 运行时注册(插件)

# 缺点:
# ❌ 隐藏依赖(全局状态)
# ❌ 测试需 reset
# ❌ 服务生命周期管理复杂

# 最佳实践:
# - 优先依赖注入
# - SL 用于插件/框架服务
# - 接口抽象(可替换实现)
# - 明确服务生命周期
# - 测试提供 reset/替换

# 决策:
# 标准业务 → DI
# 插件体系 → SL
# 全局基础设施 → SL(日志/配置)

服务定位器集中"注册与查找"。了解它与依赖注入的权衡,插件、全局服务场景用对工具。

传输对象模式

传输对象模式(Transfer Object / DTO)

传输对象(DTO,Data Transfer Object):在进程/网络间传输数据的对象,只含数据不含业务逻辑。用于批量传输与解耦。

意图

# 问题: 跨层/跨网络传输数据
# - 一次传输多个字段(减少调用)
# - 暴露内部实体(安全/耦合)

# 解决: DTO 承载数据
# 实体 ↔ DTO 转换,接口传输 DTO

# 优点:
# - 减少传输次数(批量)
# - 隔离实体(不暴露内部)
# - 按需裁剪字段
# - 序列化友好

# 缺点:
# - 转换样板代码
# - DTO 与实体同步维护

结构

# 参与者:
# TransferObject: 传输对象(数据载体)
# Entity: 业务实体
# Service: 业务服务(转换 DTO)
# Client: 客户端(接收 DTO)

# 例:
# UserEntity(内部) ↔ UserDTO(传输)
# API 返回 UserDTO(不含密码等)

Python 示例

from dataclasses import dataclass

# 业务实体(内部,含敏感字段)
class User:
    def __init__(self, id, name, email, password):
        self.id = id
        self.name = name
        self.email = email
        self.password = password   # 敏感!

# 传输对象(DTO): 只含需传输字段
@dataclass
class UserDTO:
    id: int
    name: str
    email: str
    # 不含 password!

    def to_dict(self):
        return {
            "id": self.id,
            "name": self.name,
            "email": self.email,
        }

# 业务服务: 实体 ↔ DTO 转换
class UserService:
    def get_user_dto(self, user_id) -> UserDTO:
        entity = self._find_entity(user_id)
        # 实体 → DTO(隐藏敏感字段)
        return UserDTO(
            id=entity.id,
            name=entity.name,
            email=entity.email,
        )

    def _find_entity(self, user_id):
        return User(user_id, "张三",
                    "z@e.com", "secret123")

# 客户端: 只接触 DTO
svc = UserService()
dto = svc.get_user_dto(1)
print(dto.to_dict())
# {'id': 1, 'name': '张三', 'email': 'z@e.com'}
# 密码不会泄露!

批量传输对象

# 批量 DTO(一次传输多个)
@dataclass
class UsersBatchDTO:
    users: list
    total: int
    page: int

    def to_dict(self):
        return {
            "users": [u.to_dict() for u in self.users],
            "total": self.total,
            "page": self.page,
        }

# 服务返回批量 DTO
def list_users_dto(self, page):
    entities = self._find_page(page)
    dtos = [UserDTO(e.id, e.name, e.email)
            for e in entities]
    return UsersBatchDTO(dtos, total=len(dtos), page=page)

# 优点: 一次调用返回所有数据
# 避免 N 次调用(减少网络开销)

现代 Python: Pydantic DTO

# Pydantic 是最流行的 DTO 工具
from pydantic import BaseModel

class UserDTO(BaseModel):
    id: int
    name: str
    email: str

# 自动校验 + 序列化
dto = UserDTO(id=1, name="张三", email="z@e.com")
print(dto.model_dump())
# {'id': 1, 'name': '张三', 'email': 'z@e.com'}

# 从实体转换
user = User(1, "张三", "z@e.com", "secret")
dto = UserDTO.model_validate({
    "id": user.id,
    "name": user.name,
    "email": user.email,
})

# FastAPI 中即接口响应模型
@app.get("/users/{uid}", response_model=UserDTO)
def get_user(uid: int):
    entity = find(uid)
    return UserDTO(id=entity.id,
                   name=entity.name,
                   email=entity.email)
# 响应自动过滤敏感字段

# Pydantic 优势: 校验/转换/文档自动生成

实际应用

# 1. API 请求/响应模型
# 2. 跨服务数据传输
# 3. 批量数据交换(报表/导入)
# 4. 序列化安全(隐藏敏感字段)
# 5. 数据版本控制

# 分层建议:
# Controller ← DTO → Service ← Entity
# 接口层用 DTO,内部用 Entity
# 避免实体直接暴露

# 反模式:
# - DTO 塞满业务逻辑(应纯数据)
# - DTO 与实体完全重复(失去意义)
# - 无必要地复制数据(简单场景直接用 dict)

优缺点总结

# 优点:
# ✅ 减少传输次数
# ✅ 隔离实体(安全)
# ✅ 按需裁剪字段
# ✅ 序列化友好

# 缺点:
# ❌ 转换样板代码
# ❌ 维护成本(同步变更)
# ❌ 数据冗余(复制)

# 最佳实践:
# - DTO 纯数据(不可变 dataclass)
# - 敏感字段不进入 DTO
# - 批量传输减少调用
# - 用 Pydantic/dataclass 简化
# - 明确转换边界

# 决策:
# 跨层/网络传输 → DTO
# 简单内部使用 → dict/dataclass
# API 开发 → Pydantic 模型

传输对象(DTO)是数据交换的"标准信封"。掌握实体-DTO 转换与 Pydantic,接口层安全高效,内部结构不泄露。

设计模式资源

设计模式资源

本节汇总设计模式的学习资源、工具与实践建议,帮助你从"认识模式"走向"用好模式"。

经典书籍

# 必读经典
# 1. 《Design Patterns》(GoF 四人组)
#    设计模式开山之作,23 个模式权威定义

# 2. 《Head First Design Patterns》
#    图解入门首选,Java 示例,易读

# 3. 《Refactoring》(Martin Fowler)
#    重构与模式的结合(识别坏味道→套模式)

# 4. 《Patterns of Enterprise Application Architecture》
#    企业应用模式(分层/数据源/并发)

# Python 专项
# 5. 《Python Design Patterns》(笔者的朋友可参考)
# 6. 《Architecture Patterns with Python》
#    现代 Python 架构(仓储/事件驱动/DDD)

在线资源

# 设计模式参考站
# - refactoring.guru(图解,多语言,强烈推荐)
# - sourcemaking.com(模式+反模式)
# - dofactory.com(速查表)

# Python 示例
# - github 搜索 "python design patterns"
#   常见仓库: faif/python-patterns(经典)

# 视频课程
# - YouTube: Refactoring Guru 频道
# - 国内: B站设计模式教程(Java/Go 为主)

# 学习路径建议:
# 1. refactoring.guru 通读一遍(图解理解)
# 2. 动手写 Python 示例(本教程可参考)
# 3. 在项目里识别与重构应用

模式速查表

# 创建型(5)
# 工厂: 创建对象解耦
# 抽象工厂: 产品族
# 单例: 全局唯一
# 建造者: 分步构建
# 原型: 克隆对象

# 结构型(7)
# 适配器: 接口转换
# 桥接: 抽象与实现分离
# 组合: 树形统一
# 装饰器: 动态加功能
# 外观: 简化接口
# 享元: 共享省内存
# 代理: 控制访问

# 行为型(11)
# 责任链: 请求链式处理
# 命令: 请求封装(撤销)
# 解释器: 文法解释
# 迭代器: 统一遍历
# 中介者: 集中协调
# 备忘录: 状态快照
# 观察者: 发布订阅
# 状态: 状态驱动行为
# 策略: 算法替换
# 模板: 骨架复用
# 访问者: 操作分离

# J2EE(10)
# MVC / 业务代表 / 组合实体 / DAO
# 前端控制器 / 拦截过滤器 / 服务定位器 / DTO

模式选择思维

# 面对设计问题时的思考顺序:
# 1. 能不能不引入模式?(最简单方案优先)
# 2. 变化点是什么?(识别变化维度)
# 3. 选哪个模式?(按变化点匹配)
# 4. 如何最小化引入?(只在需要处用)

# 变化点 → 模式:
# 对象创建变化 → 工厂/原型
# 行为算法变化 → 策略/模板
# 状态行为变化 → 状态
# 通知需求 → 观察者
# 接口不匹配 → 适配器
# 功能动态添加 → 装饰器
# 请求处理链 → 责任链
# 撤销需求 → 命令/备忘录

# 记住: 模式解决"变化"与"重复"
# 没有变化点,就没有模式的价值

实践建议

# 1. 从"识别"开始
#    读代码时标注用到的模式(很多框架都是)

# 2. 先重构后套模式
#    有坏味道(重复/巨大类/条件分支)
#    才考虑模式(Refactoring 配合)

# 3. Python 特色思考
#    部分模式在 Python 有更简写法:
#    单例→模块, 迭代器→生成器
#    策略→函数, 命令→闭包
#    空对象→默认值, 观察者→事件库

# 4. 团队沟通
#    用模式名交流(设计评审更高效)

# 5. 面试准备
#    理解意图与场景,能画结构图
#    会写小示例,能说优缺点

# 6. 持续积累
#    每个项目复盘: 用了什么模式,效果如何

最后的话

# 设计模式是"经验",不是"规则"
# 核心原则(比模式更重要):
# - 简单优先(KISS)
# - 单一职责(SRP)
# - 开闭原则(OCP)
# - 组合优于继承

# 用模式的正确姿势:
# 先有"问题",再找"模式"
# 而不是"会用模式",到处硬套

# 随着经验增长:
# 你会从"记模式"到"自然设计"
# 模式融入直觉,成为内功

# 本教程是起点,实践是最好的老师。
# 多写、多读、多思考,祝你成为
# 一名优秀的软件设计师!