命令模式

命令模式（Command Pattern）将一个请求封装为一个对象，从而使你可用不同的请求对客户进行参数化；对请求排队或记录请求日志，以及支持可撤销的操作。

问题定义

场景1：智能家居控制

# ❌ 问题：紧密耦合
class SmartHome:
"""智能家居控制器"""
def turn_on_light(self):
print("开灯")
def turn_off_light(self):
print("关灯")
def open_curtains(self):
print("开窗帘")
def close_curtains(self):
print("关窗帘")
def turn_on_ac(self):
print("开空调")
def turn_off_ac(self):
print("关空调")
# 问题：每个设备都需要单独的方法，无法参数化
# 想要批量执行、撤销操作很难实现

场景2：文本编辑器

# ❌ 问题：无法撤销
class TextEditor:
"""文本编辑器"""
def __init__(self):
self.text = ""
def type(self, chars):
self.text += chars
print(f"输入: {chars}")
def delete(self, count):
self.text = self.text[:-count]
print(f"删除{count}个字符")
# 问题：无法撤销操作
# 想要实现撤销需要复杂的逻辑

解决方案

命令模式将请求封装成对象，从而可以使用不同的请求对客户进行参数化。

标准实现

命令接口

from abc import ABC, abstractmethod
class Command(ABC):
"""命令接口"""
@abstractmethod
def execute(self):
pass
@abstractmethod
def undo(self):
pass

接收者

class Light:
"""电灯 - 接收者"""
def on(self):
print("💡 开灯")
def off(self):
print("🌙 关灯")
class AC:
"""空调 - 接收者"""
def on(self):
print("❄️ 开空调")
def off(self):
print("🔥 关空调")

具体命令

class LightOnCommand(Command):
"""开灯命令"""
def __init__(self, light):
self._light = light
def execute(self):
self._light.on()
def undo(self):
self._light.off()
class LightOffCommand(Command):
"""关灯命令"""
def __init__(self, light):
self._light = light
def execute(self):
self._light.off()
def undo(self):
self._light.on()
class ACOnCommand(Command):
"""开空调命令"""
def __init__(self, ac):
self._ac = ac
def execute(self):
self._ac.on()
def undo(self):
self._ac.off()
class ACOffCommand(Command):
"""关空调命令"""
def __init__(self, ac):
self._ac = ac
def execute(self):
self._ac.off()
def undo(self):
self._ac.on()

调用者

class RemoteControl:
"""遥控器 - 调用者"""
def __init__(self):
self._command = None
self._undo_command = None
def set_command(self, command: Command):
"""设置命令"""
self._command = command
def press_button(self):
"""按下按钮"""
if self._command:
self._undo_command = self._command
self._command.execute()
def press_undo(self):
"""按下撤销按钮"""
if self._undo_command:
self._undo_command.undo()

客户端使用

# 创建接收者
living_room_light = Light()
bedroom_ac = AC()
# 创建命令
light_on = LightOnCommand(living_room_light)
light_off = LightOffCommand(living_room_light)
ac_on = ACOnCommand(bedroom_ac)
ac_off = ACOffCommand(bedroom_ac)
# 创建调用者
remote = RemoteControl()
# 执行命令
print("开灯:")
remote.set_command(light_on)
remote.press_button()
print("\n开空调:")
remote.set_command(ac_on)
remote.press_button()
print("\n撤销操作:")
remote.press_undo()
print("\n关灯:")
remote.set_command(light_off)
remote.press_button()

实战应用

应用1：智能家居系统

from abc import ABC, abstractmethod
class Command(ABC):
@abstractmethod
def execute(self):
pass
@abstractmethod
def undo(self):
pass
class SmartHomeController:
"""智能家居控制器 - 接收者"""
def turn_on_light(self, room):
print(f"💡 打开{room}的灯")
def turn_off_light(self, room):
print(f"🌙 关闭{room}的灯")
def open_curtains(self, room):
print(f"🪟 打开{room}的窗帘")
def close_curtains(self, room):
print(f"🪟 关闭{room}的窗帘")
def set_temperature(self, temp):
print(f"🌡️ 设置空调温度: {temp}°C")
# 具体命令
class LightOnCommand(Command):
def __init__(self, controller, room):
self._controller = controller
self._room = room
def execute(self):
self._controller.turn_on_light(self._room)
def undo(self):
self._controller.turn_off_light(self._room)
class CurtainsOpenCommand(Command):
def __init__(self, controller, room):
self._controller = controller
self._room = room
def execute(self):
self._controller.open_curtains(self._room)
def undo(self):
self._controller.close_curtains(self._room)
class SetTemperatureCommand(Command):
def __init__(self, controller, temp):
self._controller = controller
self._temp = temp
self._prev_temp = 25  # 默认温度
def execute(self):
self._prev_temp = 24  # 假设当前是24度
self._controller.set_temperature(self._temp)
def undo(self):
self._controller.set_temperature(self._prev_temp)
# 宏命令 - 执行多个命令
class MacroCommand(Command):
"""宏命令"""
def __init__(self, commands):
self._commands = commands
def execute(self):
for command in self._commands:
command.execute()
def undo(self):
# 反向撤销
for command in reversed(self._commands):
command.undo()
# 调用者
class VoiceControl:
"""语音控制器 - 调用者"""
def __init__(self):
self._commands = {}
self._history = []
def add_command(self, name, command):
"""添加语音指令"""
self._commands[name] = command
def execute_voice_command(self, command_name):
"""执行语音指令"""
if command_name in self._commands:
command = self._commands[command_name]
self._history.append(command)
print(f"🎤 语音指令: '{command_name}'")
command.execute()
return True
return False
def undo_last(self):
"""撤销上一个指令"""
if self._history:
command = self._history.pop()
print("🔄 撤销操作")
command.undo()
# 使用
controller = SmartHomeController()
# 创建命令
light_on = LightOnCommand(controller, "客厅")
curtains_open = CurtainsOpenCommand(controller, "客厅")
set_temp = SetTemperatureCommand(controller, 26)
# 创建回家场景宏命令
come_home = MacroCommand([light_on, curtains_open, set_temp])
# 创建语音控制器
voice = VoiceControl()
voice.add_command("回家", come_home)
voice.add_command("开灯", light_on)
voice.add_command("开窗帘", curtains_open)
# 执行语音指令
voice.execute_voice_command("回家")
voice.execute_voice_command("开灯")
print("\n撤销操作:")
voice.undo_last()

应用2：文本编辑器

from abc import ABC, abstractmethod
class Command(ABC):
@abstractmethod
def execute(self):
pass
@abstractmethod
def undo(self):
pass
class TextEditor:
"""文本编辑器 - 接收者"""
def __init__(self):
self.text = ""
self.cursor = 0
def insert(self, chars, position):
"""插入文本"""
self.text = self.text[:position] + chars + self.text[position:]
self.cursor = position + len(chars)
print(f"插入文本: '{chars}'")
def delete(self, start, end):
"""删除文本"""
deleted = self.text[start:end]
self.text = self.text[:start] + self.text[end:]
print(f"删除文本: '{deleted}'")
def copy(self, start, end):
"""复制文本"""
return self.text[start:end]
def paste(self, text, position):
"""粘贴文本"""
self.insert(text, position)
def get_text(self):
"""获取文本"""
return self.text
# 具体命令
class InsertCommand(Command):
def __init__(self, editor, text, position=None):
self._editor = editor
self._text = text
self._position = position if position is not None else editor.cursor
def execute(self):
self._editor.insert(self._text, self._position)
def undo(self):
start = self._position
end = self._position + len(self._text)
self._editor.delete(start, end)
class DeleteCommand(Command):
def __init__(self, editor, start, end):
self._editor = editor
self._start = start
self._end = end
self._deleted_text = ""
def execute(self):
self._deleted_text = self._editor.copy(self._start, self._end)
self._editor.delete(self._start, self._end)
def undo(self):
self._editor.paste(self._deleted_text, self._start)
class CutCommand(Command):
def __init__(self, editor, start, end):
self._editor = editor
self._start = start
self._end = end
self._cut_text = ""
def execute(self):
self._cut_text = self._editor.copy(self._start, self._end)
self._editor.delete(self._start, self._end)
print(f"剪切文本: '{self._cut_text}'")
def undo(self):
self._editor.paste(self._cut_text, self._start)
# 调用者
class EditorApp:
"""编辑器应用 - 调用者"""
def __init__(self, editor):
self._editor = editor
self._history = []
self._current_position = 0
def insert(self, text):
command = InsertCommand(self._editor, text)
self._execute(command)
def delete(self, count):
start = self._editor.cursor - count
end = self._editor.cursor
command = DeleteCommand(self._editor, start, end)
self._execute(command)
def cut(self, start, end):
command = CutCommand(self._editor, start, end)
self._execute(command)
def undo(self):
if self._current_position > 0:
self._current_position -= 1
command = self._history[self._current_position]
command.undo()
print("撤销操作")
else:
print("没有可撤销的操作")
def redo(self):
if self._current_position < len(self._history):
command = self._history[self._current_position]
command.execute()
self._current_position += 1
print("重做操作")
def _execute(self, command):
# 如果在历史中间执行新命令，删除后续历史
if self._current_position < len(self._history):
self._history = self._history[:self._current_position]
command.execute()
self._history.append(command)
self._current_position += 1
def show_text(self):
print(f"\n当前文本: '{self._editor.get_text()}'")
# 使用
editor = TextEditor()
app = EditorApp(editor)
app.insert("Hello ")
app.insert("World!")
app.show_text()
# 当前文本: 'Hello World!'
app.delete(1)
app.show_text()
# 当前文本: 'Hello World'
app.undo()
app.show_text()
# 当前文本: 'Hello World!'
app.redo()
app.show_text()
# 当前文本: 'Hello World'

应用3：队列和延迟执行

from abc import ABC, abstractmethod
from queue import Queue
from threading import Thread
import time
class Command(ABC):
@abstractmethod
def execute(self):
pass
class Task:
"""任务 - 接收者"""
def __init__(self, name):
self.name = name
def run(self):
print(f"执行任务: {self.name}")
class TaskCommand(Command):
"""任务命令"""
def __init__(self, task):
self._task = task
def execute(self):
self._task.run()
class CommandQueue:
"""命令队列 - 调用者"""
def __init__(self):
self._queue = Queue()
self._running = False
self._worker_thread = None
def add_command(self, command):
"""添加命令到队列"""
self._queue.put(command)
print(f"📥 命令已加入队列，队列长度: {self._queue.qsize()}")
def start(self):
"""启动工作线程"""
if not self._running:
self._running = True
self._worker_thread = Thread(target=self._process_commands)
self._worker_thread.start()
print("🚀 命令队列已启动")
def stop(self):
"""停止工作线程"""
self._running = False
if self._worker_thread:
self._worker_thread.join()
print("🛑 命令队列已停止")
def _process_commands(self):
"""处理队列中的命令"""
while self._running:
try:
command = self._queue.get(timeout=1)
print("📤 从队列获取命令")
command.execute()
self._queue.task_done()
time.sleep(1)  # 模拟执行时间
except:
pass
class DelayedCommandExecutor:
"""延迟命令执行器"""
def __init__(self):
self._scheduled_commands = []
def schedule(self, command, delay_seconds):
"""延迟执行命令"""
import threading
def delayed_execute():
time.sleep(delay_seconds)
print(f"⏰ 延迟{delay_seconds}秒后执行:")
command.execute()
thread = Thread(target=delayed_execute)
thread.start()
print(f"⏳ 命令已安排在{delay_seconds}秒后执行")
# 使用
# 创建命令队列
queue = CommandQueue()
queue.start()
# 添加多个命令到队列
queue.add_command(TaskCommand(Task("任务1")))
queue.add_command(TaskCommand(Task("任务2")))
queue.add_command(TaskCommand(Task("任务3")))
queue.add_command(TaskCommand(Task("任务4")))
# 等待队列处理
time.sleep(6)
queue.stop()
# 延迟执行
print("\n延迟执行示例:")
executor = DelayedCommandExecutor()
executor.schedule(TaskCommand(Task("延迟任务1")), 2)
executor.schedule(TaskCommand(Task("延迟任务2")), 4)
time.sleep(6)

优缺点

✅ 优点

❌ 缺点

适用场景

本章要点

• ✅ 命令模式将请求封装成对象
• ✅ 解耦发送者和接收者
• ✅ 支持撤销和重做
• ✅ 支持宏命令组合
• ✅ 适用于智能家居、编辑器、队列等
• ✅ 可以延迟执行和排队

下一步：迭代器模式 🚀