04-行为型模式¶

学习时间: 约5-6小时 难度级别: ⭐⭐⭐⭐ 中高级 前置知识: SOLID设计原则、创建型模式、结构型模式 学习目标: 掌握11种行为型模式，理解对象间职责分配与通信机制

🎯 学习目标¶

掌握策略、观察者、命令三大高频行为型模式
理解模板方法与策略模式的异同
掌握Python迭代器协议与迭代器模式的关系
理解责任链模式在中间件系统中的应用
能根据场景选择合适的行为型模式

目录¶

1. 行为型模式概述
2. 策略模式（Strategy）
3. 观察者模式（Observer）
4. 命令模式（Command）
5. 模板方法模式（Template Method）
6. 迭代器模式（Iterator）
7. 状态模式（State）
8. 责任链模式（Chain of Responsibility）
9. 中介者模式（Mediator）
10. 备忘录模式（Memento）
11. 访问者模式（Visitor）
12. 解释器模式（Interpreter）
13. 行为型模式对比
14. 练习与自我检查

1. 行为型模式概述¶

行为型模式关注对象之间的职责分配和通信方式，让复杂的控制流变得清晰。

模式	一句话	使用频率
策略	算法可互换	⭐⭐⭐⭐⭐
观察者	事件通知	⭐⭐⭐⭐⭐
命令	操作对象化	⭐⭐⭐⭐
模板方法	骨架+钩子	⭐⭐⭐⭐
迭代器	统一遍历	⭐⭐⭐⭐
状态	状态驱动行为	⭐⭐⭐⭐
责任链	链式处理	⭐⭐⭐⭐
中介者	集中协调	⭐⭐⭐
备忘录	状态快照	⭐⭐⭐
访问者	分离操作与结构	⭐⭐
解释器	语法解析	⭐⭐

2. 策略模式（Strategy）¶

意图¶

定义一系列算法，将每个算法封装起来，使它们可以互换。

适用场景¶

多种方式处理同一问题（排序、计价、验证等）
消除大量 if/elif/else 分支
算法需要在运行时动态切换

Python实现¶

Python

from abc import ABC, abstractmethod
from typing import List

# ====== 策略接口 ======
class PricingStrategy(ABC):
    @abstractmethod
    def calculate(self, price: float) -> float:
        pass

# ====== 具体策略 ======
class NormalPricing(PricingStrategy):
    def calculate(self, price):
        return price

class VIPPricing(PricingStrategy):
    def calculate(self, price):
        return price * 0.8  # 8折

class PromotionPricing(PricingStrategy):
    def __init__(self, discount_rate):
        self.discount_rate = discount_rate

    def calculate(self, price):
        return price * self.discount_rate

class SecondHalfPrice(PricingStrategy):
    """第二件半价"""
    def calculate(self, price):
        return price * 0.75  # 两件均价

# ====== 上下文 ======
class ShoppingCart:
    def __init__(self, strategy: PricingStrategy = None):
        self._strategy = strategy or NormalPricing()
        self._items: List[tuple] = []

    def set_strategy(self, strategy: PricingStrategy):
        """运行时切换策略"""
        self._strategy = strategy

    def add_item(self, name, price):
        self._items.append((name, price))

    def total(self):
        return sum(self._strategy.calculate(p) for _, p in self._items)

# ====== 使用 ======
cart = ShoppingCart()
cart.add_item("Book", 100)
cart.add_item("Pen", 50)

print(f"Normal: ¥{cart.total()}")        # 150
cart.set_strategy(VIPPricing())
print(f"VIP: ¥{cart.total()}")            # 120
cart.set_strategy(PromotionPricing(0.6))
print(f"Promotion: ¥{cart.total()}")      # 90

Python函数式简化¶

Python中可以直接用函数代替策略类：

Python

from typing import Callable

def normal(price): return price
def vip(price): return price * 0.8
def promotion(price, rate=0.6): return price * rate

class ShoppingCartFP:
    def __init__(self, strategy: Callable = normal):
        self.strategy = strategy
        self.items = []

    def add(self, name, price):
        self.items.append((name, price))

    def total(self):
        return sum(self.strategy(p) for _, p in self.items)

cart = ShoppingCartFP(strategy=vip)
cart.add("Book", 100)
print(cart.total())  # 80.0

Java实现¶

Java

public interface PricingStrategy {
    double calculate(double price);
}

public class VIPPricing implements PricingStrategy {  // extends继承；implements实现接口
    public double calculate(double price) { return price * 0.8; }
}

public class ShoppingCart {
    private PricingStrategy strategy;

    public void setStrategy(PricingStrategy strategy) {
        this.strategy = strategy;
    }

    public double getTotal() {
        return items.stream()
            .mapToDouble(item -> strategy.calculate(item.price))
            .sum();
    }
}

3. 观察者模式（Observer）¶

意图¶

定义对象间的一对多依赖关系，当一个对象状态改变时，所有依赖者自动通知。

适用场景¶

事件系统（GUI事件、DOM事件）
发布-订阅（消息队列、RSS）
数据绑定（MVVM）

Python实现¶

Python

from abc import ABC, abstractmethod
from typing import Dict, List, Callable

# ====== 方式一：经典OOP实现 ======
class EventManager:
    """事件管理器（发布-订阅）"""
    def __init__(self):
        self._listeners: Dict[str, List[Callable]] = {}

    def subscribe(self, event_type: str, listener: Callable):
        if event_type not in self._listeners:
            self._listeners[event_type] = []
        self._listeners[event_type].append(listener)

    def unsubscribe(self, event_type: str, listener: Callable):
        if event_type in self._listeners:
            self._listeners[event_type].remove(listener)

    def notify(self, event_type: str, data=None):
        for listener in self._listeners.get(event_type, []):
            listener(data)

# ====== 具体发布者 ======
class UserService:
    def __init__(self):
        self.events = EventManager()

    def register(self, username, email):
        user = {"username": username, "email": email}
        print(f"User registered: {username}")
        # 注册成功后发布事件
        self.events.notify("user_registered", user)

    def login(self, username):
        print(f"User logged in: {username}")
        self.events.notify("user_logged_in", {"username": username})

# ====== 具体订阅者 ======
class EmailService:
    def send_welcome(self, user):
        print(f"  📧 Sending welcome email to {user['email']}")

class AnalyticsService:
    def track_registration(self, user):
        print(f"  📊 Tracking registration: {user['username']}")

class NotificationService:
    def on_login(self, data):
        print(f"  🔔 Login notification: {data['username']}")

# ====== 组装 ======
user_service = UserService()

email = EmailService()
analytics = AnalyticsService()
notification = NotificationService()

# 订阅事件
user_service.events.subscribe("user_registered", email.send_welcome)
user_service.events.subscribe("user_registered", analytics.track_registration)
user_service.events.subscribe("user_logged_in", notification.on_login)

# 触发事件
user_service.register("alice", "alice@example.com")
# User registered: alice
#   📧 Sending welcome email to alice@example.com
#   📊 Tracking registration: alice

user_service.login("alice")
# User logged in: alice
#   🔔 Login notification: alice

方式二：Python信号机制¶

Python

# 使用 blinker 库实现信号（类似Django的signals）
from collections import defaultdict  # defaultdict带默认值的字典，避免KeyError

class Signal:
    """简化版信号实现"""
    def __init__(self, name=""):
        self.name = name
        self._receivers = []

    def connect(self, func):
        """装饰器方式注册"""
        self._receivers.append(func)
        return func

    def send(self, sender=None, **kwargs):  # *args接收任意位置参数；**kwargs接收任意关键字参数
        results = []
        for receiver in self._receivers:
            results.append(receiver(sender, **kwargs))
        return results

# 定义信号
user_registered = Signal("user_registered")
order_created = Signal("order_created")

# 注册处理器
@user_registered.connect
def handle_new_user(sender, **kwargs):
    print(f"New user: {kwargs.get('username')}")

@user_registered.connect
def send_email(sender, **kwargs):
    print(f"Sending email to: {kwargs.get('email')}")

# 触发
user_registered.send(None, username="bob", email="bob@test.com")

注意事项¶

⚠️ 内存泄漏风险：观察者模式最常见的隐患是忘记取消订阅。如果订阅者对象已经不再使用但未调用 unsubscribe()，发布者仍然持有其引用，导致订阅者无法被垃圾回收，造成内存泄漏。

最佳实践： - 在订阅者销毁/关闭时，务必调用 unsubscribe() 清理所有订阅 - Python 可使用 weakref 弱引用避免强引用导致的泄漏 - Java 中考虑使用 WeakReference 或在 close()/dispose() 方法中集中清理 - 长生命周期的发布者 + 短生命周期的订阅者是最高危组合

Java实现¶

Java

import java.util.*;

public interface EventListener {  // interface定义类型契约
    void update(String eventType, Object data);
}

public class EventManager {
    private Map<String, List<EventListener>> listeners = new HashMap<>();

    public void subscribe(String type, EventListener listener) {
        listeners.computeIfAbsent(type, k -> new ArrayList<>()).add(listener);
    }

    public void notify(String type, Object data) {
        for (EventListener l : listeners.getOrDefault(type, List.of())) {
            l.update(type, data);
        }
    }
}

4. 命令模式（Command）¶

意图¶

将请求封装为一个对象，从而支持撤销、排队、日志等操作。

Python实现¶

Python

from abc import ABC, abstractmethod
from typing import List

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

    @abstractmethod
    def undo(self): pass

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

    def insert(self, text, position=-1):
        if position == -1:
            self.content += text
        else:
            self.content = self.content[:position] + text + self.content[position:]

    def delete(self, start, length):
        deleted = self.content[start:start + length]
        self.content = self.content[:start] + self.content[start + length:]
        return deleted

    def __str__(self):
        return f"'{self.content}'"

# ====== 具体命令 ======
class InsertCommand(Command):
    def __init__(self, editor: TextEditor, text: str, position: int = -1):
        self.editor = editor
        self.text = text
        self.position = position

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

    def undo(self):
        if self.position == -1:
            pos = len(self.editor.content) - len(self.text)
        else:
            pos = self.position
        self.editor.delete(pos, len(self.text))

class DeleteCommand(Command):
    def __init__(self, editor: TextEditor, start: int, length: int):
        self.editor = editor
        self.start = start
        self.length = length
        self._deleted = ""

    def execute(self):
        self._deleted = self.editor.delete(self.start, self.length)

    def undo(self):
        self.editor.insert(self._deleted, self.start)

# ====== 调用者（支持撤销/重做） ======
class CommandManager:
    def __init__(self):
        self._history: List[Command] = []
        self._redo_stack: List[Command] = []

    def execute(self, command: Command):
        command.execute()
        self._history.append(command)
        self._redo_stack.clear()

    def undo(self):
        if not self._history:
            print("Nothing to undo!")
            return
        command = self._history.pop()
        command.undo()
        self._redo_stack.append(command)

    def redo(self):
        if not self._redo_stack:
            print("Nothing to redo!")
            return
        command = self._redo_stack.pop()
        command.execute()
        self._history.append(command)

# ====== 使用 ======
editor = TextEditor()
manager = CommandManager()

manager.execute(InsertCommand(editor, "Hello"))
print(editor)  # 'Hello'

manager.execute(InsertCommand(editor, " World"))
print(editor)  # 'Hello World'

manager.undo()
print(editor)  # 'Hello'

manager.redo()
print(editor)  # 'Hello World'

manager.execute(DeleteCommand(editor, 5, 6))
print(editor)  # 'Hello'

实际应用¶

文本编辑器的撤销/重做
GUI框架的按钮/菜单操作绑定
数据库事务
任务队列和调度系统

5. 模板方法模式（Template Method）¶

意图¶

定义算法的骨架，将某些步骤延迟到子类实现。

Python实现¶

Python

from abc import ABC, abstractmethod

class DataProcessor(ABC):
    """模板方法定义处理流程骨架"""

    def process(self, source):
        """模板方法 — 定义算法骨架，不可覆盖"""
        data = self.read_data(source)
        parsed = self.parse_data(data)
        if self.should_validate():  # 钩子方法
            self.validate(parsed)
        result = self.transform(parsed)
        self.save(result)
        return result

    @abstractmethod
    def read_data(self, source):
        """抽象方法 — 子类必须实现"""
        pass

    @abstractmethod
    def parse_data(self, raw_data):
        pass

    def should_validate(self):
        """钩子方法 — 子类可选择覆盖"""
        return True

    def validate(self, data):
        print(f"Validating {len(data)} records")

    @abstractmethod
    def transform(self, data):
        pass

    def save(self, result):
        print(f"Saved {len(result)} records")

class CSVProcessor(DataProcessor):
    def read_data(self, source):
        print(f"Reading CSV: {source}")
        return "name,age\nAlice,25\nBob,30"

    def parse_data(self, raw_data):
        lines = raw_data.strip().split("\n")
        headers = lines[0].split(",")
        return [dict(zip(headers, line.split(","))) for line in lines[1:]]  # zip并行遍历多个可迭代对象

    def transform(self, data):
        return [{**d, "age": int(d["age"])} for d in data]

class JSONProcessor(DataProcessor):
    def read_data(self, source):
        print(f"Reading JSON: {source}")
        return '[{"name":"Alice","age":25}]'

    def parse_data(self, raw_data):
        import json
        return json.loads(raw_data)  # json.loads将JSON字符串转为Python对象

    def should_validate(self):
        return False  # JSON自带格式校验，跳过

    def transform(self, data):
        return data

# 使用
csv_processor = CSVProcessor()
csv_processor.process("data.csv")

json_processor = JSONProcessor()
json_processor.process("data.json")

策略 vs 模板方法¶

	策略模式	模板方法
粒度	替换整个算法	替换算法的某些步骤
关系	组合（has-a）	继承（is-a）
灵活性	运行时切换	编译时确定

6. 迭代器模式（Iterator）¶

意图¶

提供一种方法顺序访问集合中的元素，而不暴露集合的内部结构。

Python迭代器协议¶

Python内置了迭代器模式支持，通过 __iter__() 和 __next__() 实现：

Python

class FibonacciIterator:
    """斐波那契数列迭代器"""
    def __init__(self, max_count):
        self.max_count = max_count
        self.count = 0
        self.a, self.b = 0, 1

    def __iter__(self):
        return self

    def __next__(self):
        if self.count >= self.max_count:
            raise StopIteration
        self.count += 1
        result = self.a
        self.a, self.b = self.b, self.a + self.b
        return result

# 使用
for num in FibonacciIterator(10):
    print(num, end=" ")
# 0 1 1 2 3 5 8 13 21 34

# ====== 自定义可迭代集合 ======
class TreeNode:
    def __init__(self, value, children=None):
        self.value = value
        self.children = children or []

class TreeIterator:
    """树的深度优先迭代器"""
    def __init__(self, root: TreeNode):
        self._stack = [root]

    def __iter__(self):
        return self

    def __next__(self):
        if not self._stack:
            raise StopIteration
        node = self._stack.pop()
        # 逆序压栈，保证左子节点先出
        self._stack.extend(reversed(node.children))
        return node.value

class Tree:
    def __init__(self, root: TreeNode):
        self.root = root

    def __iter__(self):
        return TreeIterator(self.root)

# 构建树并遍历
tree = Tree(TreeNode("root", [
    TreeNode("A", [TreeNode("A1"), TreeNode("A2")]),
    TreeNode("B", [TreeNode("B1")]),
]))

print(list(tree))  # ['root', 'A', 'A1', 'A2', 'B', 'B1']

生成器：更Pythonic的迭代器¶

Python

def fibonacci(max_count):
    """生成器函数 — 等价于FibonacciIterator但更简洁"""
    a, b = 0, 1
    for _ in range(max_count):
        yield a  # yield生成器：惰性产出值，节省内存
        a, b = b, a + b

# 树的深度优先遍历（生成器版）
def dfs(node):
    yield node.value
    for child in node.children:
        yield from dfs(child)  # yield from 递归委托

print(list(fibonacci(8)))  # [0, 1, 1, 2, 3, 5, 8, 13]

7. 状态模式（State）¶

意图¶

允许对象在内部状态改变时改变其行为，对象看起来好像修改了其类。

Python实现¶

Python

from abc import ABC, abstractmethod

# ====== 状态接口 ======
class OrderState(ABC):
    @abstractmethod
    def pay(self, order): pass

    @abstractmethod
    def ship(self, order): pass

    @abstractmethod
    def deliver(self, order): pass

    @abstractmethod
    def cancel(self, order): pass

# ====== 具体状态 ======
class PendingState(OrderState):
    def pay(self, order):
        print("✅ Payment received!")
        order.state = PaidState()

    def ship(self, order):
        print("❌ Cannot ship: not paid yet")

    def deliver(self, order):
        print("❌ Cannot deliver: not paid yet")

    def cancel(self, order):
        print("✅ Order cancelled")
        order.state = CancelledState()

class PaidState(OrderState):
    def pay(self, order):
        print("❌ Already paid")

    def ship(self, order):
        print("✅ Order shipped!")
        order.state = ShippedState()

    def deliver(self, order):
        print("❌ Cannot deliver: not shipped yet")

    def cancel(self, order):
        print("✅ Refunding and cancelling")
        order.state = CancelledState()

class ShippedState(OrderState):
    def pay(self, order): print("❌ Already paid")
    def ship(self, order): print("❌ Already shipped")

    def deliver(self, order):
        print("✅ Order delivered!")
        order.state = DeliveredState()

    def cancel(self, order):
        print("❌ Cannot cancel: already shipped")

class DeliveredState(OrderState):
    def pay(self, order): print("❌ Already completed")
    def ship(self, order): print("❌ Already completed")
    def deliver(self, order): print("❌ Already delivered")
    def cancel(self, order): print("❌ Cannot cancel: already delivered")

class CancelledState(OrderState):
    def pay(self, order): print("❌ Order cancelled")
    def ship(self, order): print("❌ Order cancelled")
    def deliver(self, order): print("❌ Order cancelled")
    def cancel(self, order): print("❌ Already cancelled")

# ====== 上下文 ======
class Order:
    def __init__(self, order_id):
        self.order_id = order_id
        self.state: OrderState = PendingState()

    def pay(self): self.state.pay(self)
    def ship(self): self.state.ship(self)
    def deliver(self): self.state.deliver(self)
    def cancel(self): self.state.cancel(self)

    @property
    def status(self):
        return self.state.__class__.__name__.replace("State", "")

# 使用
order = Order("ORD-001")
print(f"Status: {order.status}")  # Pending
order.ship()     # ❌ Cannot ship: not paid yet
order.pay()      # ✅ Payment received!
print(f"Status: {order.status}")  # Paid
order.ship()     # ✅ Order shipped!
order.deliver()  # ✅ Order delivered!
order.cancel()   # ❌ Cannot cancel: already delivered

状态 vs 策略¶

	状态模式	策略模式
状态转换	状态之间互相知道，自动转换	策略互不知道
切换方式	内部自动	外部手动
意图	管理状态机	替换算法

8. 责任链模式（Chain of Responsibility）¶

意图¶

将请求沿着处理者链传递，每个处理者可以处理请求或传递给下一个。

Python实现¶

Python

from abc import ABC, abstractmethod
from typing import Optional

# ====== 中间件/处理器基类 ======
class Middleware(ABC):
    def __init__(self):
        self._next: Optional[Middleware] = None

    def set_next(self, handler: 'Middleware') -> 'Middleware':
        self._next = handler
        return handler  # 支持链式设置

    def handle(self, request: dict) -> dict:
        if self._next:
            return self._next.handle(request)
        return request

# ====== 具体中间件 ======
class AuthMiddleware(Middleware):
    def handle(self, request):
        token = request.get("token")
        if not token:
            return {"error": "401 Unauthorized"}
        if token != "valid_token":
            return {"error": "403 Forbidden"}
        print("  ✅ Auth passed")
        request["user"] = "alice"
        return super().handle(request)  # super()调用父类方法

class RateLimitMiddleware(Middleware):
    def __init__(self, max_requests=100):
        super().__init__()
        self.max_requests = max_requests
        self._count = 0

    def handle(self, request):
        self._count += 1
        if self._count > self.max_requests:
            return {"error": "429 Too Many Requests"}
        print(f"  ✅ Rate limit OK ({self._count}/{self.max_requests})")
        return super().handle(request)

class LoggingMiddleware(Middleware):
    def handle(self, request):
        print(f"  📝 Request: {request.get('path', '/')}")
        response = super().handle(request)
        print(f"  📝 Response: {response}")
        return response

class ValidationMiddleware(Middleware):
    def handle(self, request):
        if not request.get("path"):
            return {"error": "400 Bad Request: missing path"}
        print("  ✅ Validation passed")
        return super().handle(request)

# ====== 构建中间件链 ======
logging = LoggingMiddleware()
rate_limit = RateLimitMiddleware(max_requests=10)
auth = AuthMiddleware()
validation = ValidationMiddleware()

# 设置链: Logging → RateLimit → Auth → Validation
logging.set_next(rate_limit).set_next(auth).set_next(validation)

# 处理请求
print("--- Request 1 (valid) ---")
result = logging.handle({"path": "/api/data", "token": "valid_token"})

print("\n--- Request 2 (no token) ---")
result = logging.handle({"path": "/api/data"})

实际应用¶

Web框架中间件（Django Middleware、Express.js Middleware）
Java Servlet Filter链
日志框架的Handler链
GUI事件冒泡机制

9. 中介者模式（Mediator）¶

意图¶

用一个中介对象来封装一系列对象之间的交互，使各对象不需要直接引用彼此。

Python实现¶

Python

from abc import ABC, abstractmethod

class ChatRoom:
    """中介者：聊天室"""
    def __init__(self, name):
        self.name = name
        self._users = {}

    def join(self, user):
        self._users[user.name] = user
        user.room = self
        self._broadcast(f"[{user.name} joined the room]", sender=None)

    def send(self, message, sender, target=None):
        if target:
            # 私聊
            if target in self._users:
                self._users[target].receive(message, sender.name, private=True)
        else:
            # 群发
            self._broadcast(message, sender)

    def _broadcast(self, message, sender):
        for name, user in self._users.items():
            if sender is None or name != sender.name:
                user.receive(message, sender.name if sender else "System")

class User:
    def __init__(self, name):
        self.name = name
        self.room: ChatRoom = None

    def send(self, message, target=None):
        if self.room:
            self.room.send(message, self, target)

    def receive(self, message, sender, private=False):
        tag = "[DM]" if private else ""
        print(f"  {self.name} received {tag} from {sender}: {message}")

# 使用
room = ChatRoom("dev-chat")
alice = User("Alice")
bob = User("Bob")
charlie = User("Charlie")

room.join(alice)
room.join(bob)
room.join(charlie)

alice.send("Hello everyone!")   # Bob和Charlie收到
bob.send("Hi Alice!", "Alice")  # 仅Alice收到（私聊）

10. 备忘录模式（Memento）¶

意图¶

在不破坏封装性的前提下，捕获和恢复对象的内部状态。

Python实现¶

Python

import copy
from datetime import datetime

class EditorMemento:
    """备忘录：保存编辑器状态快照"""
    def __init__(self, content, cursor_pos, selection):
        self._content = content
        self._cursor_pos = cursor_pos
        self._selection = selection
        self._timestamp = datetime.now()

    @property  # @property将方法变为属性访问
    def timestamp(self):
        return self._timestamp

    @property
    def description(self):
        preview = self._content[:30] + "..." if len(self._content) > 30 else self._content  # 切片操作：[start:end:step]提取子序列
        return f"[{self._timestamp.strftime('%H:%M:%S')}] '{preview}'"

class Editor:
    """发起者：文本编辑器"""
    def __init__(self):
        self.content = ""
        self.cursor_pos = 0
        self.selection = ""

    def type_text(self, text):
        self.content += text
        self.cursor_pos = len(self.content)

    def save(self) -> EditorMemento:
        return EditorMemento(
            copy.copy(self.content),
            self.cursor_pos,
            self.selection
        )

    def restore(self, memento: EditorMemento):
        self.content = memento._content
        self.cursor_pos = memento._cursor_pos
        self.selection = memento._selection

class History:
    """管理者：保存备忘录历史"""
    def __init__(self, editor: Editor):
        self._editor = editor
        self._snapshots = []

    def backup(self):
        self._snapshots.append(self._editor.save())

    def undo(self):
        if not self._snapshots:
            return
        memento = self._snapshots.pop()
        self._editor.restore(memento)

    def show_history(self):
        for i, s in enumerate(self._snapshots):  # enumerate同时获取索引和值
            print(f"  {i}: {s.description}")

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

editor.type_text("Hello")
history.backup()

editor.type_text(" World")
history.backup()

editor.type_text("!!!")
print(f"Current: '{editor.content}'")  # Hello World!!!

history.undo()
print(f"After undo: '{editor.content}'")  # Hello World

history.undo()
print(f"After undo: '{editor.content}'")  # Hello

11. 访问者模式（Visitor）¶

意图¶

在不修改类结构的前提下，定义新的操作。将算法与数据结构分离。

Python实现¶

Python

from abc import ABC, abstractmethod

# ====== 元素 ======
class Shape(ABC):
    @abstractmethod
    def accept(self, visitor): pass

class Circle(Shape):
    def __init__(self, radius):
        self.radius = radius

    def accept(self, visitor):
        return visitor.visit_circle(self)

class Rectangle(Shape):
    def __init__(self, width, height):
        self.width = width
        self.height = height

    def accept(self, visitor):
        return visitor.visit_rectangle(self)

# ====== 访问者 ======
class ShapeVisitor(ABC):
    @abstractmethod
    def visit_circle(self, circle): pass
    @abstractmethod
    def visit_rectangle(self, rect): pass

class AreaCalculator(ShapeVisitor):
    """计算面积的访问者"""
    def visit_circle(self, c):
        import math
        return math.pi * c.radius ** 2

    def visit_rectangle(self, r):
        return r.width * r.height

class DrawVisitor(ShapeVisitor):
    """绘制的访问者"""
    def visit_circle(self, c):
        return f"Drawing circle with r={c.radius}"

    def visit_rectangle(self, r):
        return f"Drawing rectangle {r.width}x{r.height}"

# 新增操作不需要修改Shape类
shapes = [Circle(5), Rectangle(3, 4), Circle(10)]

area_calc = AreaCalculator()
total = sum(s.accept(area_calc) for s in shapes)
print(f"Total area: {total:.2f}")

drawer = DrawVisitor()
for s in shapes:
    print(s.accept(drawer))

适用场景：数据结构稳定但操作频繁变化时使用。编译器中常用于对AST的多种遍历处理。

12. 解释器模式（Interpreter）¶

意图¶

给定一个语言，定义其文法的一种表示，并用解释器来解释该语言中的句子。

Python简化实现¶

Python

from abc import ABC, abstractmethod  # ABC抽象基类；abstractmethod强制子类实现

class Expression(ABC):
    @abstractmethod
    def interpret(self, context: dict) -> bool:
        pass

class VariableExpression(Expression):
    def __init__(self, name):
        self.name = name

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

class AndExpression(Expression):
    def __init__(self, left, right):
        self.left = left
        self.right = right

    def interpret(self, context):
        return self.left.interpret(context) and self.right.interpret(context)

class OrExpression(Expression):
    def __init__(self, left, right):
        self.left = left
        self.right = right

    def interpret(self, context):
        return self.left.interpret(context) or self.right.interpret(context)

class NotExpression(Expression):
    def __init__(self, expr):
        self.expr = expr

    def interpret(self, context):
        return not self.expr.interpret(context)

# 构建规则: (is_vip AND has_balance) OR is_admin
rule = OrExpression(
    AndExpression(VariableExpression("is_vip"), VariableExpression("has_balance")),
    VariableExpression("is_admin")
)

print(rule.interpret({"is_vip": True, "has_balance": True, "is_admin": False}))   # True
print(rule.interpret({"is_vip": True, "has_balance": False, "is_admin": False}))  # False
print(rule.interpret({"is_vip": False, "has_balance": False, "is_admin": True}))  # True

实际应用：正则表达式引擎、SQL解析器、规则引擎。因复杂度高，实际项目中较少直接手写。

13. 行为型模式对比¶

高频模式选择指南¶

Text Only

需要替换算法？           → 策略模式
需要事件通知？           → 观察者模式
需要撤销/重做？          → 命令模式
需要定义算法骨架？       → 模板方法
需要遍历集合？           → 迭代器模式
对象行为随状态改变？     → 状态模式
请求需要链式处理？       → 责任链模式
多对象复杂交互？         → 中介者模式
需要保存/恢复状态？      → 备忘录模式
数据结构稳定但操作常变？ → 访问者模式

容易混淆的模式¶

对比	区别
策略 vs 状态	策略外部切换算法；状态内部自动转换
命令 vs 策略	命令封装操作（可撤销）；策略封装算法
观察者 vs 中介者	观察者是分布式通知；中介者是集中式协调
模板方法 vs 策略	模板方法用继承覆盖步骤；策略用组合替换整体

14. 练习与自我检查¶

✏️ 练习题¶

策略模式：实现一个日志系统，支持多种输出策略（控制台、文件、远程服务器），可在运行时切换。
观察者模式：实现一个股票价格监控系统，当价格超过阈值时通知多个订阅者（邮件、短信、APP推送）。
命令 + 备忘录：扩展文本编辑器的命令模式实现，添加宏录制功能——记录一系列命令并一次性回放。
状态模式：实现一个自动售货机状态机，包含状态：待投币、已投币、出货中、缺货。画出状态转换图并用代码实现。
责任链：为一个Web API实现中间件链：CORS → RateLimit → Auth → Validation → Handler → ErrorHandler。测试不同请求场景下的链式处理行为。
迭代器：用生成器实现一个二叉树的中序遍历、前序遍历和层序遍历迭代器。

面试要点¶

Q1: 策略模式和状态模式有什么区别？ A: 策略模式让客户端主动选择算法，策略之间互不感知；状态模式中状态对象知道其他状态，状态转换由内部自动完成。策略是"做什么"的选择，状态是"处于什么阶段"的管理。

Q2: 观察者模式在实际项目中怎么用？ A: 最常见的是事件系统和发布-订阅。如：用户注册后发布事件，邮件服务、积分服务、数据统计服务各自订阅处理。Django的Signal、Spring的ApplicationEvent都是观察者模式的实现。

Q3: 什么场景下用命令模式？ A: 需要将操作对象化的场景：(1)撤销/重做 (2)操作排队 (3)宏录制 (4)事务回滚 (5)远程操作。核心是把"做什么"封装成对象，与"谁来做"、"何时做"解耦。

Q4: 责任链模式在哪些框架中用到？ A: Django Middleware、Express.js的中间件、Java Servlet Filter、Netty的ChannelPipeline。请求沿链传递，每个节点决定处理或传给下一个。

自我检查清单¶

能区分策略、状态、命令三者的意图
能实现一个完整的观察者/事件系统
理解Python的迭代器协议和生成器
能用责任链实现中间件系统
能解释模板方法和策略的选择依据
至少完成3道练习题

下一章: 05-设计模式实战 — 在真实项目中综合运用设计模式