行为型模式

行为型模式关注对象之间的通信和职责分配。本文将介绍四种重要的行为型模式：策略、命令、状态和责任链模式。

策略模式

策略模式定义一系列算法，将每个算法封装起来，并使它们可以相互替换。

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

class SortStrategy(ABC):
    """排序策略接口"""
    
    @abstractmethod
    def sort(self, data: List[int]) -> List[int]:
        pass

class BubbleSortStrategy(SortStrategy):
    """冒泡排序策略"""
    
    def sort(self, data: List[int]) -> List[int]:
        arr = data.copy()
        n = len(arr)
        for i in range(n):
            for j in range(0, n-i-1):
                if arr[j] > arr[j+1]:
                    arr[j], arr[j+1] = arr[j+1], arr[j]
        return arr

class QuickSortStrategy(SortStrategy):
    """快速排序策略"""
    
    def sort(self, data: List[int]) -> List[int]:
        if len(data) <= 1:
            return data
        pivot = data[len(data) // 2]
        left = [x for x in data if x < pivot]
        middle = [x for x in data if x == pivot]
        right = [x for x in data if x > pivot]
        return self.sort(left) + middle + self.sort(right)

class InsertionSortStrategy(SortStrategy):
    """插入排序策略"""
    
    def sort(self, data: List[int]) -> List[int]:
        arr = data.copy()
        for i in range(1, len(arr)):
            key = arr[i]
            j = i - 1
            while j >= 0 and key < arr[j]:
                arr[j + 1] = arr[j]
                j -= 1
            arr[j + 1] = key
        return arr

class Sorter:
    """排序器，使用策略模式"""
    
    def __init__(self, strategy: SortStrategy):
        self._strategy = strategy
    
    def set_strategy(self, strategy: SortStrategy):
        self._strategy = strategy
    
    def sort(self, data: List[int]) -> List[int]:
        return self._strategy.sort(data)

# 使用
data = [64, 34, 25, 12, 22, 11, 90]

sorter = Sorter(BubbleSortStrategy())
print(f"冒泡排序: {sorter.sort(data)}")

sorter.set_strategy(QuickSortStrategy())
print(f"快速排序: {sorter.sort(data)}")

# 使用函数作为策略
def python_builtin_sort(data: List[int]) -> List[int]:
    return sorted(data)

# 策略也可以是函数
sorter.set_strategy(python_builtin_sort)
print(f"内置排序: {sorter.sort(data)}")

命令模式

命令模式将请求封装成对象，支持参数化、排队、日志记录和事务支持。

from abc import ABC, abstractmethod
from typing import List, Optional
from datetime import datetime

class Command(ABC):
    """命令接口"""
    
    @abstractmethod
    def execute(self) -> str:
        pass
    
    @abstractmethod
    def undo(self) -> str:
        pass
    
    @abstractmethod
    def description(self) -> str:
        pass

class TextEditor:
    """文本编辑器"""
    
    def __init__(self):
        self.content = ""
    
    def insert(self, text: str, position: int):
        self.content = self.content[:position] + text + self.content[position:]
    
    def delete(self, position: int, length: int) -> str:
        deleted = self.content[position:position+length]
        self.content = self.content[:position] + self.content[position+length:]
        return deleted

class InsertCommand(Command):
    """插入命令"""
    
    def __init__(self, editor: TextEditor, text: str, position: int):
        self.editor = editor
        self.text = text
        self.position = position
    
    def execute(self) -> str:
        self.editor.insert(self.text, self.position)
        return f"插入 '{self.text}' 到位置 {self.position}"
    
    def undo(self) -> str:
        self.editor.delete(self.position, len(self.text))
        return f"撤销插入 '{self.text}'"
    
    def description(self) -> str:
        return f"InsertCommand('{self.text}', {self.position})"

class DeleteCommand(Command):
    """删除命令"""
    
    def __init__(self, editor: TextEditor, position: int, length: int):
        self.editor = editor
        self.position = position
        self.length = length
        self.deleted_text = ""
    
    def execute(self) -> str:
        self.deleted_text = self.editor.delete(self.position, self.length)
        return f"删除 '{self.deleted_text}'"
    
    def undo(self) -> str:
        self.editor.insert(self.deleted_text, self.position)
        return f"恢复 '{self.deleted_text}'"
    
    def description(self) -> str:
        return f"DeleteCommand({self.position}, {self.length})"

class CommandHistory:
    """命令历史记录"""
    
    def __init__(self):
        self._history: List[Command] = []
        self._redo_stack: List[Command] = []
    
    def execute(self, command: Command) -> str:
        result = command.execute()
        self._history.append(command)
        self._redo_stack.clear()
        return result
    
    def undo(self) -> Optional[str]:
        if not self._history:
            return None
        command = self._history.pop()
        result = command.undo()
        self._redo_stack.append(command)
        return result
    
    def redo(self) -> Optional[str]:
        if not self._redo_stack:
            return None
        command = self._redo_stack.pop()
        result = command.execute()
        self._history.append(command)
        return result

# 使用
editor = TextEditor()
history = CommandHistory()

print(history.execute(InsertCommand(editor, "Hello", 0)))
print(f"内容: '{editor.content}'")  # 'Hello'

print(history.execute(InsertCommand(editor, " World", 5)))
print(f"内容: '{editor.content}'")  # 'Hello World'

print(history.execute(DeleteCommand(editor, 5, 6)))
print(f"内容: '{editor.content}'")  # 'Hello'

print(history.undo())  # 撤销删除
print(f"内容: '{editor.content}'")  # 'Hello World'

print(history.redo())  # 重做删除
print(f"内容: '{editor.content}'")  # 'Hello'

状态模式

状态模式允许对象在内部状态改变时改变其行为。

from abc import ABC, abstractmethod
from typing import Optional

class OrderState(ABC):
    """订单状态接口"""
    
    @abstractmethod
    def next(self, order: 'Order') -> Optional['OrderState']:
        pass
    
    @abstractmethod
    def cancel(self, order: 'Order') -> Optional['OrderState']:
        pass
    
    @abstractmethod
    def status(self) -> str:
        pass

class PendingState(OrderState):
    """待处理状态"""
    
    def next(self, order: 'Order') -> Optional['OrderState']:
        print("订单确认，开始处理")
        return ProcessingState()
    
    def cancel(self, order: 'Order') -> Optional['OrderState']:
        print("订单已取消")
        return CancelledState()
    
    def status(self) -> str:
        return "待处理"

class ProcessingState(OrderState):
    """处理中状态"""
    
    def next(self, order: 'Order') -> Optional['OrderState']:
        print("订单处理完成，准备发货")
        return ShippedState()
    
    def cancel(self, order: 'Order') -> Optional['OrderState']:
        print("订单处理中无法取消，将进行退款处理")
        return RefundedState()
    
    def status(self) -> str:
        return "处理中"

class ShippedState(OrderState):
    """已发货状态"""
    
    def next(self, order: 'Order') -> Optional['OrderState']:
        print("订单已送达")
        return DeliveredState()
    
    def cancel(self, order: 'Order') -> Optional['OrderState']:
        print("已发货无法取消，请拒收或申请退货")
        return self
    
    def status(self) -> str:
        return "已发货"

class DeliveredState(OrderState):
    """已送达状态"""
    
    def next(self, order: 'Order') -> Optional['OrderState']:
        print("订单已完成")
        return CompletedState()
    
    def cancel(self, order: 'Order') -> Optional['OrderState']:
        print("已送达，请申请退货")
        return self
    
    def status(self) -> str:
        return "已送达"

class CompletedState(OrderState):
    """已完成状态"""
    
    def next(self, order: 'Order') -> Optional['OrderState']:
        print("订单已完成，无法继续")
        return self
    
    def cancel(self, order: 'Order') -> Optional['OrderState']:
        print("订单已完成，无法取消")
        return self
    
    def status(self) -> str:
        return "已完成"

class CancelledState(OrderState):
    """已取消状态"""
    
    def next(self, order: 'Order') -> Optional['OrderState']:
        print("订单已取消，无法继续")
        return self
    
    def cancel(self, order: 'Order') -> Optional['OrderState']:
        print("订单已取消")
        return self
    
    def status(self) -> str:
        return "已取消"

class RefundedState(OrderState):
    """已退款状态"""
    
    def next(self, order: 'Order') -> Optional['OrderState']:
        print("订单已退款，无法继续")
        return self
    
    def cancel(self, order: 'Order') -> Optional['OrderState']:
        print("订单已退款")
        return self
    
    def status(self) -> str:
        return "已退款"

class Order:
    """订单类"""
    
    def __init__(self, order_id: str):
        self.order_id = order_id
        self._state: OrderState = PendingState()
    
    def next_state(self):
        new_state = self._state.next(self)
        if new_state:
            self._state = new_state
    
    def cancel(self):
        new_state = self._state.cancel(self)
        if new_state:
            self._state = new_state
    
    def get_status(self) -> str:
        return self._state.status()

# 使用
order = Order("ORD001")
print(f"状态: {order.get_status()}")  # 待处理

order.next_state()  # 订单确认，开始处理
print(f"状态: {order.get_status()}")  # 处理中

order.next_state()  # 订单处理完成，准备发货
print(f"状态: {order.get_status()}")  # 已发货

责任链模式

责任链模式为请求创建一个处理者链，每个处理者决定处理请求或传递给下一个处理者。

from abc import ABC, abstractmethod
from typing import Optional

class Handler(ABC):
    """处理器接口"""
    
    def __init__(self):
        self._next: Optional[Handler] = None
    
    def set_next(self, handler: 'Handler') -> 'Handler':
        self._next = handler
        return handler
    
    @abstractmethod
    def handle(self, request: dict) -> Optional[str]:
        pass

class AuthenticationHandler(Handler):
    """认证处理器"""
    
    def handle(self, request: dict) -> Optional[str]:
        if not request.get("token"):
            return "认证失败: 缺少令牌"
        
        if request["token"] != "valid_token":
            return "认证失败: 无效令牌"
        
        print("认证通过")
        if self._next:
            return self._next.handle(request)
        return None

class AuthorizationHandler(Handler):
    """授权处理器"""
    
    def handle(self, request: dict) -> Optional[str]:
        role = request.get("role")
        if role not in ["admin", "user"]:
            return f"授权失败: 未知角色 {role}"
        
        required_role = request.get("required_role", "user")
        if role != "admin" and required_role == "admin":
            return "授权失败: 需要管理员权限"
        
        print("授权通过")
        if self._next:
            return self._next.handle(request)
        return None

class RateLimitHandler(Handler):
    """限流处理器"""
    
    def __init__(self, max_requests: int = 100):
        super().__init__()
        self.request_count = 0
        self.max_requests = max_requests
    
    def handle(self, request: dict) -> Optional[str]:
        self.request_count += 1
        if self.request_count > self.max_requests:
            return "限流: 请求过于频繁"
        
        print(f"限流检查通过 ({self.request_count}/{self.max_requests})")
        if self._next:
            return self._next.handle(request)
        return None

# 构建处理链
auth = AuthenticationHandler()
authz = AuthorizationHandler()
rate_limit = RateLimitHandler(max_requests=3)

auth.set_next(authz).set_next(rate_limit)

# 测试
request1 = {"token": "valid_token", "role": "admin"}
print(auth.handle(request1))

print("\n---")
request2 = {"token": "invalid_token", "role": "user"}
print(auth.handle(request2))

print("\n---")
request3 = {"token": "valid_token", "role": "guest"}
print(auth.handle(request3))

模式对比

模式	核心思想	应用场景
策略	算法可互换	排序、定价、验证规则
命令	撤销/重做	编辑器、事务、任务队列
状态	行为随状态变	订单流程、游戏状态、工作流
责任链	传递请求	中间件、审批流程、事件处理

行为型模式的关键是管理对象间的协作：策略封装算法，命令封装动作，状态封装行为变化，责任链封装处理流程。