第14章 命令模式
学习目标
完成本章学习后，读者将能够：
- 理解命令模式的核心概念与形式化定义
- 掌握将请求封装为对象的多种实现方式
- 设计支持撤销/重做的命令系统
- 实现宏命令与命令队列
- 分析命令模式与函数式编程的关系
- 识别命令模式的适用场景与反模式
14.1 模式定义
14.1.1 正式定义
命令模式（Command Pattern） 将一个请求封装为一个对象，从而使你可用不同的请求对客户进行参数化，对请求排队或记录请求日志，以及支持可撤销的操作。
$$\text{Command}: \text{Request} \xrightarrow{\text{encapsulate}} \text{CommandObject}$$
命令执行模型：
$$\text{Execute}(c) = c.\text{action}() \text{ where } c \in \mathcal{C}$$
撤销/重做模型：
$$\text{Undo}(c) = c^{-1} \text{ where } c^{-1} \circ c = \text{id}$$
$$\text{Redo}(c) = (c^{-1})^{-1} = c$$
命令队列模型：
$$\text{ProcessQueue}(Q) = \forall c \in Q: \text{Execute}(c)$$
14.1.2 历史背景与学术脉络
| 时期 | 发展阶段 | 关键贡献 | 代表人物/文献 |
|---|---|---|---|
| 1987 | 概念萌芽 | Smalltalk中的消息传递机制 | ParcPlace Systems |
| 1994 | GoF正式定义 | 《设计模式》收录为行为型模式 | Gamma, Helm, Johnson, Vlissides |
| 1995 | GUI应用 | 菜单和工具栏命令封装 | Various IDEs |
| 1998 | 事务处理 | 数据库事务命令模式 | Enterprise Systems |
| 2000 | 函数式编程 | 高阶函数与命令模式的关系 | Haskell Community |
| 2005 | Python应用 | 函数作为一等公民简化命令模式 | Python Community |
| 2010 | 响应式编程 | 命令与Observable结合 | RxJS, RxJava |
| 2015 | 微服务 | CQRS命令查询职责分离 | Greg Young, Udi Dahan |
| 2020 | 现代演进 | 异步命令与事件溯源 | Event Sourcing Pattern |
14.1.3 模式动机
问题场景：在软件系统中，请求处理面临以下挑战：
- 耦合问题：请求发送者直接依赖接收者
- 撤销需求：需要支持操作的撤销和重做
- 队列处理：需要将请求排队执行
- 日志记录：需要记录操作历史用于审计或恢复
- 事务管理：需要将多个操作组合为原子事务
解决方案：将请求封装为独立的命令对象：
┌─────────────────────────────────────────────────────────────────────────┐
│ 命令模式架构 │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ Client │─────▶│ Invoker │─────▶│ Command │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
│ │ │ │
│ │ │ execute() │
│ │ ▼ │
│ │ ┌──────────────┐ │
│ │ │ Receiver │ │
│ │ └──────────────┘ │
│ │ │ │
│ ▼ │ │
│ ┌──────────────┐ │ │
│ │ CommandQueue │ │ │
│ │ CommandLog │ │ │
│ │ UndoStack │ │ │
│ └──────────────┘ │ │
│ │
│ 核心思想: │
│ • 请求封装为对象 │
│ • 发送者与接收者解耦 │
│ • 支持撤销、重做、队列、日志 │
│ │
└─────────────────────────────────────────────────────────────────────────┘14.1.4 命令模式分类
| 类型 | 特征 | 示例 |
|---|---|---|
| 简单命令 | 无参数或简单参数，直接执行 | 日志命令、通知命令 |
| 参数化命令 | 携带执行参数 | 移动命令、修改命令 |
| 复合命令 | 组合多个命令 | 宏命令、事务命令 |
| 可撤销命令 | 支持撤销操作 | 编辑命令、转账命令 |
| 异步命令 | 异步执行 | 网络请求命令、文件操作命令 |
14.2 理论基础
14.2.1 UML结构模型
┌─────────────────────────────────────────────────────────────────────────┐
│ 命令模式结构图 │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────┐ ┌─────────────────────────────┐ │
│ │ Invoker │ │ <<interface>> Command │ │
│ ├─────────────────────┤ ├─────────────────────────────┤ │
│ │ - command: Command │─────────│ + execute(): void │ │
│ │ - history: Stack │ uses │ + undo(): void │ │
│ ├─────────────────────┤ │ + can_execute(): bool │ │
│ │ + set_command(c) │ └─────────────────────────────┘ │
│ │ + execute() │ △ │
│ │ + undo() │ │ implements │
│ │ + redo() │ ┌───────────┴───────────┐ │
│ └─────────────────────┘ │ │ │
│ ┌────────┴────────┐ ┌─────────┴─────────┐ │
│ │ ConcreteCommand │ │ CompositeCommand │ │
│ ├─────────────────┤ ├───────────────────┤ │
│ │ - receiver: R │ │ - commands: List │ │
│ │ - state: State │ │ + add(c) │ │
│ │ - params: Any │ │ + remove(c) │ │
│ ├─────────────────┤ └───────────────────┘ │
│ │ + execute() │ │
│ │ + undo() │ │
│ └─────────────────┘ │
│ │ │
│ │ invokes │
│ ▼ │
│ ┌─────────────────┐ │
│ │ Receiver │ │
│ ├─────────────────┤ │
│ │ + action() │ │
│ │ + undo_action() │ │
│ └─────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────┘14.2.2 参与者职责
| 参与者 | 职责 | Python实现要点 |
|---|---|---|
| Command | 声明执行接口 | ABC或Protocol |
| ConcreteCommand | 实现具体命令，绑定接收者 | 持有Receiver引用 |
| Receiver | 执行实际业务逻辑 | 业务类 |
| Invoker | 调用命令执行 | 持有Command引用，管理历史 |
| Client | 创建命令并配置接收者 | 组装Command和Receiver |
14.2.3 命令与函数的关系
┌─────────────────────────────────────────────────────────────────────────┐
│ 命令模式 vs 函数式编程 │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ 面向对象命令模式 函数式编程 │
│ ┌──────────────────┐ ┌──────────────────┐ │
│ │ class Command: │ │ def command(): │ │
│ │ def execute() │ │ ... │ │
│ │ def undo() │ │ │ │
│ └──────────────────┘ └──────────────────┘ │
│ │
│ 命令对象 = 函数 + 状态 函数 = 无状态命令 │
│ │
│ Python中的融合: │
│ • callable对象: __call__方法使对象可调用 │
│ • functools.partial: 部分应用创建命令 │
│ • lambda表达式: 简单命令的内联定义 │
│ • 装饰器: 命令的增强和组合 │
│ │
│ 命令模式优势: │
│ • 支持撤销（需要状态） │
│ • 支持序列化和持久化 │
│ • 支持类型检查和IDE支持 │
│ │
└─────────────────────────────────────────────────────────────────────────┘14.3 Python实现
14.3.1 标准ABC实现
python
from abc import ABC, abstractmethod
from typing import Any, Optional
from dataclasses import dataclass
class Command(ABC):
"""命令抽象基类"""
@abstractmethod
def execute(self) -> Any:
"""执行命令"""
pass
def can_execute(self) -> bool:
"""检查是否可以执行"""
return True
def undo(self) -> None:
"""撤销命令（可选实现）"""
raise NotImplementedError("此命令不支持撤销")
def redo(self) -> None:
"""重做命令（默认调用execute）"""
self.execute()
@property
def description(self) -> str:
"""命令描述"""
return self.__class__.__name__
class Receiver:
"""接收者：执行实际业务逻辑"""
def __init__(self, name: str):
self._name = name
self._state = {}
def action_a(self, data: str) -> str:
"""操作A"""
self._state['last_a'] = data
return f"[{self._name}] 执行操作A: {data}"
def action_b(self, data: int) -> str:
"""操作B"""
self._state['last_b'] = data
return f"[{self._name}] 执行操作B: {data}"
def get_state(self) -> dict:
return dict(self._state)
@dataclass
class SimpleCommand(Command):
"""简单命令：无接收者"""
_payload: str
def execute(self) -> str:
return f"简单命令执行: {self._payload}"
@property
def description(self) -> str:
return f"SimpleCommand({self._payload})"
class ComplexCommand(Command):
"""复杂命令：绑定接收者"""
def __init__(self, receiver: Receiver, action: str, data: Any):
self._receiver = receiver
self._action = action
self._data = data
self._result: Optional[str] = None
def execute(self) -> str:
if self._action == 'a':
self._result = self._receiver.action_a(self._data)
elif self._action == 'b':
self._result = self._receiver.action_b(self._data)
else:
self._result = f"未知操作: {self._action}"
return self._result
@property
def description(self) -> str:
return f"ComplexCommand(action={self._action}, data={self._data})"
class Invoker:
"""调用者：管理命令执行"""
def __init__(self):
self._commands: list[Command] = []
self._history: list[Command] = []
def add_command(self, command: Command) -> None:
self._commands.append(command)
def execute_all(self) -> list:
results = []
for cmd in self._commands:
if cmd.can_execute():
result = cmd.execute()
results.append(result)
self._history.append(cmd)
self._commands.clear()
return results
def get_history(self) -> list:
return [cmd.description for cmd in self._history]
if __name__ == "__main__":
receiver = Receiver("Service")
invoker = Invoker()
invoker.add_command(SimpleCommand("初始化"))
invoker.add_command(ComplexCommand(receiver, 'a', "测试数据"))
invoker.add_command(ComplexCommand(receiver, 'b', 42))
results = invoker.execute_all()
for r in results:
print(r)
print(f"\n执行历史: {invoker.get_history()}")14.3.2 支持撤销/重做的命令
python
from abc import ABC, abstractmethod
from typing import Any, Optional, List
from dataclasses import dataclass, field
from copy import deepcopy
class UndoableCommand(Command):
"""可撤销命令基类"""
@abstractmethod
def execute(self) -> Any:
pass
@abstractmethod
def undo(self) -> None:
pass
def redo(self) -> Any:
return self.execute()
@dataclass
class EditorState:
"""编辑器状态"""
content: str = ""
cursor_position: int = 0
selection_start: int = 0
selection_end: int = 0
class TextEditor:
"""文本编辑器：接收者"""
def __init__(self):
self._state = EditorState()
@property
def content(self) -> str:
return self._state.content
@content.setter
def content(self, value: str) -> None:
self._state.content = value
@property
def cursor_position(self) -> int:
return self._state.cursor_position
@cursor_position.setter
def cursor_position(self, value: int) -> None:
self._state.cursor_position = max(0, min(value, len(self._state.content)))
@property
def state(self) -> EditorState:
return deepcopy(self._state)
@state.setter
def state(self, value: EditorState) -> None:
self._state = deepcopy(value)
def insert(self, text: str) -> None:
pos = self._state.cursor_position
self._state.content = (
self._state.content[:pos] +
text +
self._state.content[pos:]
)
self._state.cursor_position = pos + len(text)
def delete(self, length: int) -> str:
pos = self._state.cursor_position
start = max(0, pos - length)
deleted = self._state.content[start:pos]
self._state.content = (
self._state.content[:start] +
self._state.content[pos:]
)
self._state.cursor_position = start
return deleted
def replace(self, old: str, new: str) -> int:
count = self._state.content.count(old)
self._state.content = self._state.content.replace(old, new)
return count
class InsertCommand(UndoableCommand):
"""插入命令"""
def __init__(self, editor: TextEditor, text: str):
self._editor = editor
self._text = text
self._previous_state: Optional[EditorState] = None
def execute(self) -> str:
self._previous_state = self._editor.state
self._editor.insert(self._text)
return f"插入: '{self._text}'"
def undo(self) -> None:
if self._previous_state:
self._editor.state = self._previous_state
@property
def description(self) -> str:
return f"Insert('{self._text}')"
class DeleteCommand(UndoableCommand):
"""删除命令"""
def __init__(self, editor: TextEditor, length: int):
self._editor = editor
self._length = length
self._previous_state: Optional[EditorState] = None
self._deleted_text: str = ""
def execute(self) -> str:
self._previous_state = self._editor.state
self._deleted_text = self._editor.delete(self._length)
return f"删除: '{self._deleted_text}'"
def undo(self) -> None:
if self._previous_state:
self._editor.state = self._previous_state
@property
def description(self) -> str:
return f"Delete({self._length})"
class ReplaceCommand(UndoableCommand):
"""替换命令"""
def __init__(self, editor: TextEditor, old: str, new: str):
self._editor = editor
self._old = old
self._new = new
self._previous_state: Optional[EditorState] = None
def execute(self) -> str:
self._previous_state = self._editor.state
count = self._editor.replace(self._old, self._new)
return f"替换: '{self._old}' -> '{self._new}' ({count}处)"
def undo(self) -> None:
if self._previous_state:
self._editor.state = self._previous_state
@property
def description(self) -> str:
return f"Replace('{self._old}', '{self._new}')"
class CommandManager:
"""命令管理器：管理撤销/重做栈"""
def __init__(self, max_history: int = 100):
self._undo_stack: List[UndoableCommand] = []
self._redo_stack: List[UndoableCommand] = []
self._max_history = max_history
def execute(self, command: UndoableCommand) -> Any:
"""执行命令并记录历史"""
result = command.execute()
self._undo_stack.append(command)
self._redo_stack.clear()
if len(self._undo_stack) > self._max_history:
self._undo_stack.pop(0)
return result
def undo(self) -> Optional[str]:
"""撤销最近命令"""
if not self._undo_stack:
return None
command = self._undo_stack.pop()
command.undo()
self._redo_stack.append(command)
return f"撤销: {command.description}"
def redo(self) -> Optional[str]:
"""重做最近撤销的命令"""
if not self._redo_stack:
return None
command = self._redo_stack.pop()
command.redo()
self._undo_stack.append(command)
return f"重做: {command.description}"
def can_undo(self) -> bool:
return len(self._undo_stack) > 0
def can_redo(self) -> bool:
return len(self._redo_stack) > 0
def get_undo_history(self) -> List[str]:
return [cmd.description for cmd in reversed(self._undo_stack)]
def get_redo_history(self) -> List[str]:
return [cmd.description for cmd in reversed(self._redo_stack)]
def clear(self) -> None:
self._undo_stack.clear()
self._redo_stack.clear()
if __name__ == "__main__":
editor = TextEditor()
manager = CommandManager()
manager.execute(InsertCommand(editor, "Hello "))
manager.execute(InsertCommand(editor, "World!"))
print(f"内容: {editor.content}")
manager.execute(ReplaceCommand(editor, "World", "Python"))
print(f"替换后: {editor.content}")
print(f"\n撤销历史: {manager.get_undo_history()}")
print(f"\n{manager.undo()}")
print(f"撤销后: {editor.content}")
print(f"\n{manager.redo()}")
print(f"重做后: {editor.content}")14.3.3 宏命令（复合命令）
python
from abc import ABC, abstractmethod
from typing import Any, List, Optional
from dataclasses import dataclass
class Command(ABC):
"""命令接口"""
@abstractmethod
def execute(self) -> Any:
pass
@abstractmethod
def undo(self) -> None:
pass
@property
def description(self) -> str:
return self.__class__.__name__
class MacroCommand(Command):
"""宏命令：组合多个命令"""
def __init__(self, name: str = "Macro"):
self._name = name
self._commands: List[Command] = []
def add(self, command: Command) -> 'MacroCommand':
self._commands.append(command)
return self
def remove(self, command: Command) -> bool:
try:
self._commands.remove(command)
return True
except ValueError:
return False
def clear(self) -> None:
self._commands.clear()
def execute(self) -> List[Any]:
results = []
for cmd in self._commands:
results.append(cmd.execute())
return results
def undo(self) -> None:
for cmd in reversed(self._commands):
cmd.undo()
@property
def description(self) -> str:
cmds = ", ".join(c.description for c in self._commands)
return f"{self._name}({cmds})"
class TransactionCommand(Command):
"""事务命令：原子执行，失败回滚"""
def __init__(self, name: str = "Transaction"):
self._name = name
self._commands: List[Command] = []
self._executed: List[Command] = []
def add(self, command: Command) -> 'TransactionCommand':
self._commands.append(command)
return self
def execute(self) -> dict:
"""执行事务，失败则回滚"""
self._executed.clear()
results = []
try:
for cmd in self._commands:
result = cmd.execute()
results.append(result)
self._executed.append(cmd)
return {"success": True, "results": results}
except Exception as e:
for cmd in reversed(self._executed):
try:
cmd.undo()
except Exception:
pass
self._executed.clear()
return {"success": False, "error": str(e)}
def undo(self) -> None:
for cmd in reversed(self._executed):
cmd.undo()
self._executed.clear()
@property
def description(self) -> str:
return f"{self._name}(commands={len(self._commands)})"
class BankAccount:
"""银行账户"""
def __init__(self, owner: str, balance: float = 0):
self._owner = owner
self._balance = balance
@property
def owner(self) -> str:
return self._owner
@property
def balance(self) -> float:
return self._balance
def deposit(self, amount: float) -> float:
if amount <= 0:
raise ValueError("存款金额必须大于0")
self._balance += amount
return self._balance
def withdraw(self, amount: float) -> float:
if amount <= 0:
raise ValueError("取款金额必须大于0")
if amount > self._balance:
raise ValueError(f"余额不足: 当前余额 {self._balance}")
self._balance -= amount
return self._balance
class TransferCommand(Command):
"""转账命令"""
def __init__(self, from_account: BankAccount, to_account: BankAccount, amount: float):
self._from = from_account
self._to = to_account
self._amount = amount
self._executed = False
def execute(self) -> str:
self._from.withdraw(self._amount)
self._to.deposit(self._amount)
self._executed = True
return f"转账成功: {self._from.owner} -> {self._to.owner}, ¥{self._amount}"
def undo(self) -> None:
if self._executed:
self._to.withdraw(self._amount)
self._from.deposit(self._amount)
self._executed = False
@property
def description(self) -> str:
return f"Transfer({self._from.owner}->{self._to.owner}, ¥{self._amount})"
class DepositCommand(Command):
"""存款命令"""
def __init__(self, account: BankAccount, amount: float):
self._account = account
self._amount = amount
self._executed = False
def execute(self) -> str:
self._account.deposit(self._amount)
self._executed = True
return f"{self._account.owner} 存款 ¥{self._amount}"
def undo(self) -> None:
if self._executed:
self._account.withdraw(self._amount)
self._executed = False
@property
def description(self) -> str:
return f"Deposit({self._account.owner}, ¥{self._amount})"
if __name__ == "__main__":
alice = BankAccount("Alice", 1000)
bob = BankAccount("Bob", 500)
print(f"初始状态: Alice={alice.balance}, Bob={bob.balance}")
macro = MacroCommand("批量操作")
macro.add(DepositCommand(alice, 100))
macro.add(TransferCommand(alice, bob, 200))
print(f"\n执行宏命令: {macro.description}")
results = macro.execute()
for r in results:
print(f" {r}")
print(f"\n执行后: Alice={alice.balance}, Bob={bob.balance}")
print("\n撤销宏命令...")
macro.undo()
print(f"撤销后: Alice={alice.balance}, Bob={bob.balance}")
print("\n=== 事务命令测试 ===")
charlie = BankAccount("Charlie", 100)
transaction = TransactionCommand("批量转账")
transaction.add(TransferCommand(alice, bob, 200))
transaction.add(TransferCommand(bob, charlie, 1000))
result = transaction.execute()
print(f"事务结果: {result}")
print(f"最终状态: Alice={alice.balance}, Bob={bob.balance}, Charlie={charlie.balance}")14.3.4 函数式命令实现
python
from typing import Callable, Any, Optional, List, Tuple
from dataclasses import dataclass
from functools import wraps, partial
@dataclass
class FunctionCommand:
"""函数命令：将函数封装为命令对象"""
execute: Callable[[], Any]
undo: Optional[Callable[[], None]] = None
_description: str = ""
def __call__(self) -> Any:
return self.execute()
@property
def description(self) -> str:
return self._description or self.execute.__name__
def command(
execute: Callable = None,
undo: Callable = None,
description: str = ""
) -> FunctionCommand:
"""命令工厂函数"""
if execute is None:
return partial(command, undo=undo, description=description)
return FunctionCommand(
execute=execute,
undo=undo,
_description=description
)
class CommandBuilder:
"""命令构建器"""
def __init__(self):
self._execute: Optional[Callable] = None
self._undo: Optional[Callable] = None
self._description: str = ""
def on_execute(self, func: Callable) -> 'CommandBuilder':
self._execute = func
return self
def on_undo(self, func: Callable) -> 'CommandBuilder':
self._undo = func
return self
def with_description(self, desc: str) -> 'CommandBuilder':
self._description = desc
return self
def build(self) -> FunctionCommand:
if self._execute is None:
raise ValueError("必须设置execute函数")
return FunctionCommand(
execute=self._execute,
undo=self._undo,
_description=self._description
)
class FunctionalCommandManager:
"""函数式命令管理器"""
def __init__(self):
self._undo_stack: List[FunctionCommand] = []
self._redo_stack: List[FunctionCommand] = []
def execute(self, cmd: FunctionCommand) -> Any:
result = cmd.execute()
self._undo_stack.append(cmd)
self._redo_stack.clear()
return result
def execute_func(
self,
func: Callable,
undo_func: Callable = None,
*args,
**kwargs
) -> Any:
"""直接执行函数并封装为命令"""
cmd = FunctionCommand(
execute=lambda: func(*args, **kwargs),
undo=undo_func
)
return self.execute(cmd)
def undo(self) -> Optional[str]:
if not self._undo_stack:
return None
cmd = self._undo_stack.pop()
if cmd.undo:
cmd.undo()
self._redo_stack.append(cmd)
return f"撤销: {cmd.description}"
def redo(self) -> Optional[str]:
if not self._redo_stack:
return None
cmd = self._redo_stack.pop()
result = cmd.execute()
self._undo_stack.append(cmd)
return f"重做: {cmd.description}"
class Counter:
"""计数器：演示函数式命令"""
def __init__(self, initial: int = 0):
self._value = initial
@property
def value(self) -> int:
return self._value
def increment(self, amount: int = 1) -> int:
self._value += amount
return self._value
def decrement(self, amount: int = 1) -> int:
self._value -= amount
return self._value
def set_value(self, value: int) -> int:
old = self._value
self._value = value
return old
if __name__ == "__main__":
counter = Counter(0)
manager = FunctionalCommandManager()
cmd1 = (
CommandBuilder()
.on_execute(lambda: counter.increment(10))
.on_undo(lambda: counter.decrement(10))
.with_description("增加10")
.build()
)
cmd2 = (
CommandBuilder()
.on_execute(lambda: counter.increment(5))
.on_undo(lambda: counter.decrement(5))
.with_description("增加5")
.build()
)
manager.execute(cmd1)
print(f"执行cmd1后: {counter.value}")
manager.execute(cmd2)
print(f"执行cmd2后: {counter.value}")
manager.undo()
print(f"撤销后: {counter.value}")
manager.redo()
print(f"重做后: {counter.value}")14.4 企业级应用示例
14.4.1 任务调度系统
python
from abc import ABC, abstractmethod
from typing import Any, Optional, List, Dict, Callable
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum, auto
from queue import PriorityQueue
import threading
import time
class TaskStatus(Enum):
"""任务状态"""
PENDING = auto()
RUNNING = auto()
COMPLETED = auto()
FAILED = auto()
CANCELLED = auto()
@dataclass(order=True)
class ScheduledTask:
"""调度任务"""
priority: int
scheduled_time: datetime = field(compare=False)
command: 'TaskCommand' = field(compare=False)
task_id: str = field(compare=False, default="")
status: TaskStatus = field(compare=False, default=TaskStatus.PENDING)
result: Any = field(compare=False, default=None)
error: str = field(compare=False, default="")
created_at: datetime = field(compare=False, default_factory=datetime.now)
def __post_init__(self):
if not self.task_id:
self.task_id = f"task_{int(time.time() * 1000)}"
class TaskCommand(ABC):
"""任务命令基类"""
@abstractmethod
def execute(self) -> Any:
pass
@property
@abstractmethod
def name(self) -> str:
pass
def on_success(self, result: Any) -> None:
pass
def on_failure(self, error: Exception) -> None:
pass
class EmailTaskCommand(TaskCommand):
"""邮件发送任务"""
def __init__(self, to: str, subject: str, body: str):
self._to = to
self._subject = subject
self._body = body
@property
def name(self) -> str:
return "SendEmail"
def execute(self) -> dict:
time.sleep(0.1)
return {
"to": self._to,
"subject": self._subject,
"sent_at": datetime.now().isoformat()
}
def on_success(self, result: Any) -> None:
print(f"[Email] 发送成功: {self._to}")
class ReportTaskCommand(TaskCommand):
"""报表生成任务"""
def __init__(self, report_type: str, parameters: dict):
self._report_type = report_type
self._parameters = parameters
@property
def name(self) -> str:
return f"Generate{self._report_type}Report"
def execute(self) -> dict:
time.sleep(0.2)
return {
"report_type": self._report_type,
"generated_at": datetime.now().isoformat(),
"rows": 100
}
class DataSyncTaskCommand(TaskCommand):
"""数据同步任务"""
def __init__(self, source: str, target: str):
self._source = source
self._target = target
@property
def name(self) -> str:
return "DataSync"
def execute(self) -> dict:
time.sleep(0.3)
return {
"source": self._source,
"target": self._target,
"records_synced": 1000
}
class TaskScheduler:
"""任务调度器"""
def __init__(self, max_workers: int = 3):
self._queue: PriorityQueue = PriorityQueue()
self._tasks: Dict[str, ScheduledTask] = {}
self._workers: List[threading.Thread] = []
self._max_workers = max_workers
self._running = False
self._lock = threading.Lock()
def schedule(
self,
command: TaskCommand,
scheduled_time: datetime = None,
priority: int = 5
) -> str:
"""调度任务"""
scheduled_time = scheduled_time or datetime.now()
task = ScheduledTask(
priority=priority,
scheduled_time=scheduled_time,
command=command
)
with self._lock:
self._tasks[task.task_id] = task
self._queue.put(task)
return task.task_id
def schedule_delayed(
self,
command: TaskCommand,
delay_seconds: float,
priority: int = 5
) -> str:
"""延迟调度"""
scheduled_time = datetime.now() + timedelta(seconds=delay_seconds)
return self.schedule(command, scheduled_time, priority)
def cancel(self, task_id: str) -> bool:
"""取消任务"""
with self._lock:
if task_id in self._tasks:
task = self._tasks[task_id]
if task.status == TaskStatus.PENDING:
task.status = TaskStatus.CANCELLED
return True
return False
def start(self) -> None:
"""启动调度器"""
self._running = True
for i in range(self._max_workers):
worker = threading.Thread(target=self._worker, daemon=True)
worker.start()
self._workers.append(worker)
def stop(self) -> None:
"""停止调度器"""
self._running = False
def _worker(self) -> None:
"""工作线程"""
while self._running:
try:
task = self._queue.get(timeout=1)
if task.status == TaskStatus.CANCELLED:
continue
if task.scheduled_time > datetime.now():
self._queue.put(task)
time.sleep(0.1)
continue
with self._lock:
task.status = TaskStatus.RUNNING
try:
result = task.command.execute()
task.result = result
task.status = TaskStatus.COMPLETED
task.command.on_success(result)
except Exception as e:
task.error = str(e)
task.status = TaskStatus.FAILED
task.command.on_failure(e)
except Exception:
pass
def get_task(self, task_id: str) -> Optional[ScheduledTask]:
return self._tasks.get(task_id)
def get_status(self) -> Dict[str, int]:
status_count = {s: 0 for s in TaskStatus}
with self._lock:
for task in self._tasks.values():
status_count[task.status] += 1
return status_count
if __name__ == "__main__":
scheduler = TaskScheduler(max_workers=2)
scheduler.start()
task1 = scheduler.schedule(
EmailTaskCommand("user@example.com", "欢迎", "欢迎加入"),
priority=1
)
task2 = scheduler.schedule(
ReportTaskCommand("Sales", {"month": "2024-01"}),
priority=3
)
task3 = scheduler.schedule_delayed(
DataSyncTaskCommand("legacy_db", "new_db"),
delay_seconds=0.5,
priority=5
)
time.sleep(1)
print(f"任务状态: {scheduler.get_status()}")
for task_id in [task1, task2, task3]:
task = scheduler.get_task(task_id)
if task:
print(f"{task_id}: {task.status.name} -> {task.result or task.error}")
scheduler.stop()14.4.2 文本编辑器撤销系统
python
from typing import Any, Optional, List, Dict, Tuple
from dataclasses import dataclass, field
from copy import deepcopy
from datetime import datetime
@dataclass
class TextSnapshot:
"""文本快照"""
content: str
cursor_line: int = 0
cursor_column: int = 0
timestamp: datetime = field(default_factory=datetime.now)
class TextBuffer:
"""文本缓冲区"""
def __init__(self, initial_content: str = ""):
self._lines: List[str] = initial_content.split('\n') if initial_content else [""]
self._cursor_line = 0
self._cursor_column = 0
self._selection_start: Optional[Tuple[int, int]] = None
self._selection_end: Optional[Tuple[int, int]] = None
@property
def content(self) -> str:
return '\n'.join(self._lines)
@content.setter
def content(self, value: str) -> None:
self._lines = value.split('\n') if value else [""]
@property
def line_count(self) -> int:
return len(self._lines)
@property
def cursor_position(self) -> Tuple[int, int]:
return (self._cursor_line, self._cursor_column)
@cursor_position.setter
def cursor_position(self, pos: Tuple[int, int]) -> None:
line, col = pos
self._cursor_line = max(0, min(line, len(self._lines) - 1))
self._cursor_column = max(0, min(col, len(self._lines[self._cursor_line])))
def get_snapshot(self) -> TextSnapshot:
return TextSnapshot(
content=self.content,
cursor_line=self._cursor_line,
cursor_column=self._cursor_column
)
def restore_snapshot(self, snapshot: TextSnapshot) -> None:
self.content = snapshot.content
self._cursor_line = snapshot.cursor_line
self._cursor_column = snapshot.cursor_column
def insert_text(self, text: str) -> None:
lines = text.split('\n')
if len(lines) == 1:
current_line = self._lines[self._cursor_line]
self._lines[self._cursor_line] = (
current_line[:self._cursor_column] +
text +
current_line[self._cursor_column:]
)
self._cursor_column += len(text)
else:
current_line = self._lines[self._cursor_line]
before = current_line[:self._cursor_column]
after = current_line[self._cursor_column:]
self._lines[self._cursor_line] = before + lines[0]
for i, line in enumerate(lines[1:-1], 1):
self._lines.insert(self._cursor_line + i, line)
self._lines.insert(self._cursor_line + len(lines) - 1, lines[-1] + after)
self._cursor_line += len(lines) - 1
self._cursor_column = len(lines[-1])
def delete_text(self, length: int) -> str:
deleted = ""
for _ in range(length):
if self._cursor_column > 0:
self._cursor_column -= 1
line = self._lines[self._cursor_line]
deleted = line[self._cursor_column] + deleted
self._lines[self._cursor_line] = (
line[:self._cursor_column] +
line[self._cursor_column + 1:]
)
elif self._cursor_line > 0:
self._cursor_line -= 1
self._cursor_column = len(self._lines[self._cursor_line])
deleted = '\n' + deleted
current = self._lines[self._cursor_line]
next_line = self._lines.pop(self._cursor_line + 1)
self._lines[self._cursor_line] = current + next_line
return deleted
def get_line(self, line_num: int) -> str:
if 0 <= line_num < len(self._lines):
return self._lines[line_num]
return ""
class EditorCommand:
"""编辑器命令基类"""
def __init__(self, buffer: TextBuffer):
self._buffer = buffer
self._snapshot: Optional[TextSnapshot] = None
def execute(self) -> Any:
self._snapshot = self._buffer.get_snapshot()
return self._do_execute()
def undo(self) -> None:
if self._snapshot:
self._buffer.restore_snapshot(self._snapshot)
def _do_execute(self) -> Any:
raise NotImplementedError
@property
def description(self) -> str:
return self.__class__.__name__
class TypeCommand(EditorCommand):
"""输入命令"""
def __init__(self, buffer: TextBuffer, text: str):
super().__init__(buffer)
self._text = text
def _do_execute(self) -> str:
self._buffer.insert_text(self._text)
return f"输入: '{self._text}'"
@property
def description(self) -> str:
return f"Type('{self._text[:10]}...')" if len(self._text) > 10 else f"Type('{self._text}')"
class BackspaceCommand(EditorCommand):
"""退格命令"""
def __init__(self, buffer: TextBuffer, count: int = 1):
super().__init__(buffer)
self._count = count
self._deleted = ""
def _do_execute(self) -> str:
self._deleted = self._buffer.delete_text(self._count)
return f"删除: '{self._deleted}'"
@property
def description(self) -> str:
return f"Backspace({self._count})"
class ReplaceAllCommand(EditorCommand):
"""全部替换命令"""
def __init__(self, buffer: TextBuffer, old: str, new: str):
super().__init__(buffer)
self._old = old
self._new = new
self._count = 0
def _do_execute(self) -> str:
content = self._buffer.content
self._count = content.count(self._old)
self._buffer.content = content.replace(self._old, self._new)
return f"替换: '{self._old}' -> '{self._new}' ({self._count}处)"
@property
def description(self) -> str:
return f"ReplaceAll('{self._old}', '{self._new}')"
class UndoManager:
"""撤销管理器"""
def __init__(self, max_history: int = 100):
self._undo_stack: List[EditorCommand] = []
self._redo_stack: List[EditorCommand] = []
self._max_history = max_history
self._transaction_commands: List[EditorCommand] = []
self._in_transaction = False
def execute(self, command: EditorCommand) -> Any:
result = command.execute()
if self._in_transaction:
self._transaction_commands.append(command)
else:
self._undo_stack.append(command)
self._redo_stack.clear()
if len(self._undo_stack) > self._max_history:
self._undo_stack.pop(0)
return result
def begin_transaction(self) -> None:
self._in_transaction = True
self._transaction_commands.clear()
def commit_transaction(self) -> None:
if self._transaction_commands:
macro = MacroEditorCommand(self._transaction_commands)
self._undo_stack.append(macro)
self._redo_stack.clear()
self._in_transaction = False
self._transaction_commands.clear()
def rollback_transaction(self) -> None:
for cmd in reversed(self._transaction_commands):
cmd.undo()
self._in_transaction = False
self._transaction_commands.clear()
def undo(self) -> Optional[str]:
if not self._undo_stack:
return None
command = self._undo_stack.pop()
command.undo()
self._redo_stack.append(command)
return f"撤销: {command.description}"
def redo(self) -> Optional[str]:
if not self._redo_stack:
return None
command = self._redo_stack.pop()
command.execute()
self._undo_stack.append(command)
return f"重做: {command.description}"
def can_undo(self) -> bool:
return len(self._undo_stack) > 0
def can_redo(self) -> bool:
return len(self._redo_stack) > 0
def get_history(self) -> List[str]:
return [cmd.description for cmd in self._undo_stack]
class MacroEditorCommand(EditorCommand):
"""宏命令"""
def __init__(self, commands: List[EditorCommand]):
self._commands = commands
self._snapshots: List[TextSnapshot] = []
def execute(self) -> List[Any]:
self._snapshots.clear()
results = []
for cmd in self._commands:
self._snapshots.append(cmd._buffer.get_snapshot())
results.append(cmd._do_execute())
return results
def undo(self) -> None:
for i, cmd in enumerate(reversed(self._commands)):
if self._snapshots:
snapshot = self._snapshots.pop()
cmd._buffer.restore_snapshot(snapshot)
@property
def description(self) -> str:
return f"Macro({len(self._commands)} commands)"
class TextEditor:
"""文本编辑器"""
def __init__(self):
self._buffer = TextBuffer()
self._undo_manager = UndoManager()
@property
def content(self) -> str:
return self._buffer.content
@property
def cursor_position(self) -> Tuple[int, int]:
return self._buffer.cursor_position
def type_text(self, text: str) -> str:
return self._undo_manager.execute(TypeCommand(self._buffer, text))
def backspace(self, count: int = 1) -> str:
return self._undo_manager.execute(BackspaceCommand(self._buffer, count))
def replace_all(self, old: str, new: str) -> str:
return self._undo_manager.execute(ReplaceAllCommand(self._buffer, old, new))
def undo(self) -> Optional[str]:
return self._undo_manager.undo()
def redo(self) -> Optional[str]:
return self._undo_manager.redo()
def begin_transaction(self) -> None:
self._undo_manager.begin_transaction()
def commit_transaction(self) -> None:
self._undo_manager.commit_transaction()
def rollback_transaction(self) -> None:
self._undo_manager.rollback_transaction()
def get_history(self) -> List[str]:
return self._undo_manager.get_history()
if __name__ == "__main__":
editor = TextEditor()
editor.type_text("Hello World")
print(f"内容: {editor.content}")
editor.type_text("\nThis is a test.")
print(f"内容:\n{editor.content}")
print(f"\n{editor.undo()}")
print(f"撤销后: {editor.content}")
print(f"\n{editor.redo()}")
print(f"重做后:\n{editor.content}")
print("\n=== 事务测试 ===")
editor.begin_transaction()
editor.type_text("Line 1\n")
editor.type_text("Line 2\n")
editor.type_text("Line 3")
editor.commit_transaction()
print(f"事务后:\n{editor.content}")
print(f"\n{editor.undo()}")
print(f"撤销事务后: {editor.content}")
print(f"\n历史记录: {editor.get_history()}")14.4.3 CQRS命令处理器
python
from abc import ABC, abstractmethod
from typing import Any, Optional, List, Dict, TypeVar, Generic, Type
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum, auto
import uuid
class CommandResult:
"""命令结果"""
def __init__(self, success: bool, data: Any = None, error: str = ""):
self.success = success
self.data = data
self.error = error
self.timestamp = datetime.now()
@classmethod
def ok(cls, data: Any = None) -> 'CommandResult':
return cls(success=True, data=data)
@classmethod
def fail(cls, error: str) -> 'CommandResult':
return cls(success=False, error=error)
TCommand = TypeVar('TCommand')
class CommandHandler(ABC, Generic[TCommand]):
"""命令处理器基类"""
@abstractmethod
def handle(self, command: TCommand) -> CommandResult:
pass
@property
@abstractmethod
def command_type(self) -> Type:
pass
@dataclass
class Command:
"""命令基类"""
command_id: str = field(default_factory=lambda: str(uuid.uuid4()))
timestamp: datetime = field(default_factory=datetime.now)
user_id: str = "system"
@dataclass
class CreateUserCommand(Command):
"""创建用户命令"""
username: str = ""
email: str = ""
role: str = "user"
@dataclass
class UpdateUserCommand(Command):
"""更新用户命令"""
user_id: str = ""
username: Optional[str] = None
email: Optional[str] = None
role: Optional[str] = None
@dataclass
class DeleteUserCommand(Command):
"""删除用户命令"""
user_id: str = ""
@dataclass
class User:
"""用户实体"""
id: str
username: str
email: str
role: str
created_at: datetime = field(default_factory=datetime.now)
updated_at: datetime = field(default_factory=datetime.now)
class UserRepository:
"""用户仓储"""
def __init__(self):
self._users: Dict[str, User] = {}
def save(self, user: User) -> User:
self._users[user.id] = user
return user
def find_by_id(self, user_id: str) -> Optional[User]:
return self._users.get(user_id)
def find_by_username(self, username: str) -> Optional[User]:
for user in self._users.values():
if user.username == username:
return user
return None
def delete(self, user_id: str) -> bool:
if user_id in self._users:
del self._users[user_id]
return True
return False
def list_all(self) -> List[User]:
return list(self._users.values())
class CreateUserHandler(CommandHandler[CreateUserCommand]):
"""创建用户处理器"""
def __init__(self, repository: UserRepository):
self._repository = repository
@property
def command_type(self) -> Type:
return CreateUserCommand
def handle(self, command: CreateUserCommand) -> CommandResult:
if self._repository.find_by_username(command.username):
return CommandResult.fail(f"用户名已存在: {command.username}")
user = User(
id=str(uuid.uuid4()),
username=command.username,
email=command.email,
role=command.role
)
saved = self._repository.save(user)
return CommandResult.ok({"user_id": saved.id, "username": saved.username})
class UpdateUserHandler(CommandHandler[UpdateUserCommand]):
"""更新用户处理器"""
def __init__(self, repository: UserRepository):
self._repository = repository
@property
def command_type(self) -> Type:
return UpdateUserCommand
def handle(self, command: UpdateUserCommand) -> CommandResult:
user = self._repository.find_by_id(command.user_id)
if not user:
return CommandResult.fail(f"用户不存在: {command.user_id}")
if command.username is not None:
user.username = command.username
if command.email is not None:
user.email = command.email
if command.role is not None:
user.role = command.role
user.updated_at = datetime.now()
self._repository.save(user)
return CommandResult.ok({"user_id": user.id, "updated": True})
class DeleteUserHandler(CommandHandler[DeleteUserCommand]):
"""删除用户处理器"""
def __init__(self, repository: UserRepository):
self._repository = repository
@property
def command_type(self) -> Type:
return DeleteUserCommand
def handle(self, command: DeleteUserCommand) -> CommandResult:
if self._repository.delete(command.user_id):
return CommandResult.ok({"deleted": True, "user_id": command.user_id})
return CommandResult.fail(f"用户不存在: {command.user_id}")
class CommandBus:
"""命令总线"""
def __init__(self):
self._handlers: Dict[Type, CommandHandler] = {}
self._middleware: List[callable] = []
self._command_log: List[Dict] = []
def register(self, handler: CommandHandler) -> None:
self._handlers[handler.command_type] = handler
def add_middleware(self, middleware: callable) -> None:
self._middleware.append(middleware)
def dispatch(self, command: Command) -> CommandResult:
command_type = type(command)
if command_type not in self._handlers:
return CommandResult.fail(f"未找到命令处理器: {command_type.__name__}")
for middleware in self._middleware:
middleware(command)
handler = self._handlers[command_type]
result = handler.handle(command)
self._command_log.append({
'command_id': command.command_id,
'command_type': command_type.__name__,
'success': result.success,
'timestamp': datetime.now().isoformat()
})
return result
def get_log(self, limit: int = 100) -> List[Dict]:
return self._command_log[-limit:]
def logging_middleware(command: Command) -> None:
print(f"[Command] {type(command).__name__}: {command.command_id}")
def validation_middleware(command: Command) -> None:
if hasattr(command, 'username') and not command.username:
raise ValueError("用户名不能为空")
if __name__ == "__main__":
repository = UserRepository()
bus = CommandBus()
bus.register(CreateUserHandler(repository))
bus.register(UpdateUserHandler(repository))
bus.register(DeleteUserHandler(repository))
bus.add_middleware(logging_middleware)
create_cmd = CreateUserCommand(
username="alice",
email="alice@example.com",
role="admin"
)
result = bus.dispatch(create_cmd)
print(f"创建用户: {result.data}")
user_id = result.data['user_id']
update_cmd = UpdateUserCommand(
user_id=user_id,
email="alice_new@example.com"
)
result = bus.dispatch(update_cmd)
print(f"更新用户: {result.data}")
print(f"\n用户列表:")
for user in repository.list_all():
print(f" - {user.username} ({user.email})")
print(f"\n命令日志:")
for log in bus.get_log():
print(f" - {log['command_type']}: {log['success']}")14.5 模式变体与扩展
14.5.1 异步命令
python
from abc import ABC, abstractmethod
from typing import Any, Optional, List, Awaitable
from dataclasses import dataclass
import asyncio
class AsyncCommand(ABC):
"""异步命令基类"""
@abstractmethod
async def execute(self) -> Any:
pass
async def undo(self) -> None:
raise NotImplementedError("此命令不支持撤销")
@property
def description(self) -> str:
return self.__class__.__name__
@dataclass
class AsyncCommandResult:
"""异步命令结果"""
success: bool
data: Any = None
error: str = ""
class AsyncFileWriteCommand(AsyncCommand):
"""异步文件写入命令"""
def __init__(self, filepath: str, content: str):
self._filepath = filepath
self._content = content
self._previous_content: Optional[str] = None
async def execute(self) -> AsyncCommandResult:
try:
try:
async with asyncio.open_file(self._filepath, 'r') as f:
self._previous_content = await f.read()
except FileNotFoundError:
self._previous_content = None
async with asyncio.open_file(self._filepath, 'w') as f:
await f.write(self._content)
return AsyncCommandResult(success=True, data=self._filepath)
except Exception as e:
return AsyncCommandResult(success=False, error=str(e))
async def undo(self) -> None:
if self._previous_content is not None:
async with asyncio.open_file(self._filepath, 'w') as f:
await f.write(self._previous_content)
@property
def description(self) -> str:
return f"WriteFile({self._filepath})"
class AsyncHttpCommand(AsyncCommand):
"""异步HTTP请求命令"""
def __init__(self, url: str, method: str = "GET", data: dict = None):
self._url = url
self._method = method
self._data = data
async def execute(self) -> AsyncCommandResult:
await asyncio.sleep(0.1)
return AsyncCommandResult(
success=True,
data={"url": self._url, "status": 200, "response": "OK"}
)
@property
def description(self) -> str:
return f"HttpRequest({self._method} {self._url})"
class AsyncCommandQueue:
"""异步命令队列"""
def __init__(self):
self._queue: asyncio.Queue = asyncio.Queue()
self._results: List[AsyncCommandResult] = []
async def enqueue(self, command: AsyncCommand) -> None:
await self._queue.put(command)
async def process_all(self) -> List[AsyncCommandResult]:
results = []
while not self._queue.empty():
command = await self._queue.get()
result = await command.execute()
results.append(result)
return results
async def process_concurrent(self, max_concurrent: int = 5) -> List[AsyncCommandResult]:
commands = []
while not self._queue.empty():
commands.append(await self._queue.get())
semaphore = asyncio.Semaphore(max_concurrent)
async def run_with_limit(cmd: AsyncCommand) -> AsyncCommandResult:
async with semaphore:
return await cmd.execute()
results = await asyncio.gather(*[run_with_limit(cmd) for cmd in commands])
return list(results)
async def main():
queue = AsyncCommandQueue()
for i in range(5):
await queue.enqueue(AsyncHttpCommand(f"https://api.example.com/{i}"))
print("=== 顺序执行 ===")
results = await queue.process_all()
for r in results:
print(f" {r.data}")
for i in range(10):
await queue.enqueue(AsyncHttpCommand(f"https://api.example.com/batch/{i}"))
print("\n=== 并发执行 ===")
results = await queue.process_concurrent(max_concurrent=3)
print(f" 完成 {len(results)} 个请求")
if __name__ == "__main__":
asyncio.run(main())14.5.2 命令序列化
python
from abc import ABC, abstractmethod
from typing import Any, Dict, Type, Optional
from dataclasses import dataclass, asdict
import json
class SerializableCommand(ABC):
"""可序列化命令"""
@abstractmethod
def execute(self) -> Any:
pass
@abstractmethod
def to_dict(self) -> Dict[str, Any]:
pass
@classmethod
@abstractmethod
def from_dict(cls, data: Dict[str, Any]) -> 'SerializableCommand':
pass
@property
@abstractmethod
def command_type(self) -> str:
pass
@dataclass
class TransferCommand(SerializableCommand):
"""转账命令"""
from_account: str
to_account: str
amount: float
reference: str = ""
def execute(self) -> str:
return f"转账: {self.from_account} -> {self.to_account}, ¥{self.amount}"
def to_dict(self) -> Dict[str, Any]:
return {
'command_type': self.command_type,
'from_account': self.from_account,
'to_account': self.to_account,
'amount': self.amount,
'reference': self.reference
}
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'TransferCommand':
return cls(
from_account=data['from_account'],
to_account=data['to_account'],
amount=data['amount'],
reference=data.get('reference', '')
)
@property
def command_type(self) -> str:
return 'transfer'
class CommandSerializer:
"""命令序列化器"""
_registry: Dict[str, Type[SerializableCommand]] = {}
@classmethod
def register(cls, command_class: Type[SerializableCommand]) -> None:
instance = command_class.__new__(command_class)
cls._registry[instance.command_type] = command_class
@classmethod
def serialize(cls, command: SerializableCommand) -> str:
return json.dumps(command.to_dict())
@classmethod
def deserialize(cls, data: str) -> SerializableCommand:
obj = json.loads(data)
command_type = obj.get('command_type')
if command_type not in cls._registry:
raise ValueError(f"未知命令类型: {command_type}")
command_class = cls._registry[command_type]
return command_class.from_dict(obj)
class CommandLogger:
"""命令日志"""
def __init__(self, filepath: str):
self._filepath = filepath
def log(self, command: SerializableCommand) -> None:
serialized = CommandSerializer.serialize(command)
with open(self._filepath, 'a') as f:
f.write(serialized + '\n')
def replay(self) -> list:
commands = []
with open(self._filepath, 'r') as f:
for line in f:
if line.strip():
cmd = CommandSerializer.deserialize(line.strip())
commands.append(cmd)
return commands
if __name__ == "__main__":
CommandSerializer.register(TransferCommand)
cmd1 = TransferCommand("ACC001", "ACC002", 1000.0, "工资")
cmd2 = TransferCommand("ACC002", "ACC003", 500.0, "还款")
serialized = CommandSerializer.serialize(cmd1)
print(f"序列化: {serialized}")
deserialized = CommandSerializer.deserialize(serialized)
print(f"反序列化: {deserialized.execute()}")
logger = CommandLogger("commands.log")
logger.log(cmd1)
logger.log(cmd2)
print("\n重放命令:")
for cmd in logger.replay():
print(f" {cmd.execute()}")14.6 反模式与最佳实践
14.6.1 常见反模式
反模式1：命令膨胀
python
from abc import ABC, abstractmethod
from typing import Any
class BadCommand(ABC):
"""错误示例：命令包含过多业务逻辑"""
@abstractmethod
def execute(self) -> Any:
pass
class DoEverythingCommand(BadCommand):
"""过度复杂的命令"""
def __init__(self, db, cache, email_service, logger):
self._db = db
self._cache = cache
self._email = email_service
self._logger = logger
def execute(self) -> Any:
pass
class GoodCommand(ABC):
"""正确示例：命令只封装调用"""
@abstractmethod
def execute(self) -> Any:
pass
class TransferFundsCommand(GoodCommand):
"""简洁的转账命令"""
def __init__(self, service, from_id, to_id, amount):
self._service = service
self._from_id = from_id
self._to_id = to_id
self._amount = amount
def execute(self) -> Any:
return self._service.transfer(self._from_id, self._to_id, self._amount)反模式2：忽略撤销
python
from abc import ABC, abstractmethod
class CommandWithoutUndo(ABC):
"""错误示例：不支持撤销"""
@abstractmethod
def execute(self):
pass
class DeleteCommand(CommandWithoutUndo):
"""无法撤销的删除命令"""
def __init__(self, data, item_id):
self._data = data
self._item_id = item_id
def execute(self):
del self._data[self._item_id]
class UndoableCommand(ABC):
"""正确示例：支持撤销"""
@abstractmethod
def execute(self):
pass
@abstractmethod
def undo(self):
pass
class SafeDeleteCommand(UndoableCommand):
"""可撤销的删除命令"""
def __init__(self, data, item_id):
self._data = data
self._item_id = item_id
self._backup = None
def execute(self):
self._backup = self._data.get(self._item_id)
del self._data[self._item_id]
def undo(self):
if self._backup is not None:
self._data[self._item_id] = self._backup14.6.2 最佳实践清单
| 实践 | 说明 | 代码示例 |
|---|---|---|
| 单一职责 | 命令只封装单一操作 | class InsertCommand |
| 支持撤销 | 保存执行前状态 | self._previous_state = state |
| 命名清晰 | 命令名表达意图 | CreateUserCommand |
| 不可变参数 | 命令参数不应被修改 | @dataclass(frozen=True) |
| 序列化支持 | 支持持久化和恢复 | to_dict() / from_dict() |
| 错误处理 | 优雅处理执行失败 | try-except 包装 |
| 日志记录 | 记录命令执行历史 | CommandLogger |
| 线程安全 | 并发环境下保护状态 | threading.Lock |
14.7 决策指南
14.7.1 是否使用命令模式？
┌─────────────────────────────────────────────────────────────┐
│ 命令模式决策树 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 问题：是否需要撤销/重做功能？ │
│ │ │
│ ├── 是 ──→ ✓ 使用命令模式 │
│ │ │
│ └── 否 │
│ │ │
│ ▼ │
│ 问题：是否需要将请求排队或延迟执行？ │
│ │ │
│ ├── 是 ──→ ✓ 使用命令模式 │
│ │ │
│ └── 否 │
│ │ │
│ ▼ │
│ 问题：是否需要记录操作日志或审计？ │
│ │ │
│ ├── 是 ──→ ✓ 使用命令模式 │
│ │ │
│ └── 否 │
│ │ │
│ ▼ │
│ 问题：是否需要支持事务操作？ │
│ │ │
│ ├── 是 ──→ ✓ 使用命令模式 │
│ │ │
│ └── 否 ──→ 考虑直接调用 │
│ │
└─────────────────────────────────────────────────────────────┘14.7.2 实现技术选择
| 场景 | 推荐实现 | 理由 |
|---|---|---|
| 简单操作 | 函数/lambda | 简洁、无需类 |
| 需要撤销 | 类命令 | 可保存状态 |
| 复杂业务 | CQRS | 分离命令与查询 |
| 异步操作 | async命令 | 非阻塞执行 |
| 需要持久化 | 可序列化命令 | 支持存储和恢复 |
14.7.3 与其他模式的关系
| 模式 | 关系 | 协作方式 |
|---|---|---|
| 策略 | 算法封装 | 命令封装请求，策略封装算法 |
| 原型 | 复制命令 | 命令可被克隆 |
| 备忘录 | 状态保存 | 命令用备忘录保存撤销状态 |
| 责任链 | 请求传递 | 命令可在链中传递 |
| 观察者 | 状态通知 | 命令执行后通知观察者 |
14.8 快速参考卡片
┌─────────────────────────────────────────────────────────────────────────┐
│ 命令模式速查表 │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ 定义: 将请求封装为对象，实现调用者与接收者解耦 │
│ │
│ 核心公式: │
│ Command: Request → CommandObject │
│ Undo(c) = c⁻¹ where c⁻¹ ∘ c = id │
│ │
│ ───────────────────────────────────────────────────────────────────── │
│ │
│ 参与者: │
│ • Command → 声明执行接口 │
│ • ConcreteCommand → 实现具体命令，绑定接收者 │
│ • Receiver → 执行实际业务逻辑 │
│ • Invoker → 调用命令执行 │
│ • Client → 创建命令并配置接收者 │
│ │
│ ───────────────────────────────────────────────────────────────────── │
│ │
│ 命令类型: │
│ 简单命令: 无参数或简单参数 │
│ 参数化命令: 携带执行参数 │
│ 复合命令: 组合多个命令（宏命令） │
│ 可撤销命令: 支持撤销操作 │
│ 异步命令: 异步执行 │
│ │
│ ───────────────────────────────────────────────────────────────────── │
│ │
│ Python实现要点: │
│ class Command: │
│ def execute(self): ... │
│ def undo(self): ... │
│ │
│ class CommandManager: │
│ def execute(self, cmd): │
│ result = cmd.execute() │
│ self._undo_stack.append(cmd) │
│ return result │
│ │
│ def undo(self): │
│ cmd = self._undo_stack.pop() │
│ cmd.undo() │
│ self._redo_stack.append(cmd) │
│ │
│ ───────────────────────────────────────────────────────────────────── │
│ │
│ 适用场景: │
│ ✓ 需要撤销/重做功能 │
│ ✓ 需要将请求排队或延迟执行 │
│ ✓ 需要记录操作日志 │
│ ✓ 需要支持事务操作 │
│ ✓ 需要解耦调用者和接收者 │
│ │
│ 不适用场景: │
│ ✗ 简单直接的方法调用 │
│ ✗ 不需要撤销功能 │
│ ✗ 操作数量极少 │
│ │
│ ───────────────────────────────────────────────────────────────────── │
│ │
│ 经典案例: │
│ • GUI编辑器撤销/重做 │
│ • 任务调度系统 │
│ • 数据库事务 │
│ • 宏录制与回放 │
│ • CQRS命令处理 │
│ │
└─────────────────────────────────────────────────────────────────────────┘14.9 思考与实践
14.9.1 思考题
概念辨析：命令模式与策略模式都封装行为，它们的核心区别是什么？什么情况下两者可以结合使用？
撤销实现：对于涉及外部系统（如数据库、网络服务）的命令，如何实现可靠的撤销？考虑补偿事务的概念。
性能考量：当命令历史栈很大时，如何优化内存使用？考虑快照策略和增量保存。
并发安全：在多线程环境中，如何保证命令执行的原子性和撤销栈的一致性？
CQRS架构：命令查询职责分离（CQRS）如何与命令模式结合？这种架构有什么优势和挑战？
14.9.2 实践练习
练习1：实现图形编辑器命令系统
设计一个图形编辑器，支持：
- 创建、移动、缩放、删除图形
- 多级撤销/重做
- 命令宏（批量操作）
- 命令历史面板
练习2：实现分布式命令执行器
设计一个分布式命令执行框架：
- 命令序列化与传输
- 远程命令执行
- 失败重试机制
- 执行结果回调
练习3：实现事务命令管理器
设计一个支持事务的命令管理器：
- 开始/提交/回滚事务
- 嵌套事务支持
- 事务隔离级别
- 死锁检测
14.10 小结
命令模式是一种行为型设计模式，通过将请求封装为对象，实现了调用者与接收者的解耦，同时提供了撤销、重做、队列、日志等强大功能。本章深入探讨了：
- 理论基础：命令模式的形式化定义、撤销/重做模型、命令与函数的关系
- 实现技术：ABC、函数式命令、宏命令、事务命令等多种Python实现方式
- 企业应用：任务调度系统、文本编辑器撤销系统、CQRS命令处理器等实际案例
- 模式变体：异步命令、命令序列化等扩展形式
- 最佳实践：避免命令膨胀、忽略撤销等反模式，掌握命令设计原则
命令模式的核心价值在于：通过封装请求为对象，实现请求的参数化、队列化、日志化，以及最重要的撤销/重做功能。
参考资料
- Gamma, E., et al. Design Patterns: Elements of Reusable Object-Oriented Software. Addison-Wesley, 1994.
- Fowler, M. Patterns of Enterprise Application Architecture. Addison-Wesley, 2002.
- Young, G. CQRS Documents. https://cqrs.files.wordpress.com/2010/11/cqrs_documents.pdf
- Python Documentation. functools — Higher-order functions. https://docs.python.org/3/library/functools.html
- Microsoft. Command Pattern. https://docs.microsoft.com/en-us/azure/architecture/patterns/command