Python 责任链模式

责任链模式是一种行为设计模式，它允许你将请求沿着处理链进行传递。收到请求后，每个处理者都可以对请求进行处理，或者将其传递给链上的下一个处理者。

生活中的类比

想象一下公司里的请假审批流程：

  • 1-3 天 → 组长审批
  • 4-7 天 → 部门经理审批
  • 8 天以上 → 总经理审批

当你提交请假申请时，申请会按照这个链条依次传递，直到找到有权限处理的人。这就是责任链模式的现实体现。

模式的核心思想

责任链模式的核心在于解耦请求的发送者和接收者，让多个对象都有机会处理请求，从而避免请求发送者与接收者之间的紧密耦合。

责任链模式的结构

让我们通过 UML 类图来理解责任链模式的组成结构：

核心组件说明

Handler（抽象处理者）

  • 定义处理请求的接口
  • 包含设置后继者的方法
  • 可以实现后继链的默认行为

ConcreteHandler（具体处理者）

  • 实现具体的处理逻辑
  • 如果可以处理请求，就进行处理
  • 否则将请求转发给后继者

基础实现：请假审批系统

让我们用 Python 实现一个简单的请假审批系统来理解责任链模式。

步骤 1：定义抽象处理者

实例

from abc import ABC, abstractmethod
from typing import Optional

class Approver(ABC):
    """审批者抽象类"""
   
    def __init__(self, name: str):
        self.name = name
        self.successor: Optional['Approver'] = None
   
    def set_successor(self, successor: 'Approver'):
        """设置后继审批者"""
        self.successor = successor
   
    @abstractmethod
    def process_request(self, leave_request: 'LeaveRequest'):
        """处理请假请求的抽象方法"""
        pass

步骤 2：定义请求对象

实例

from dataclasses import dataclass

@dataclass
class LeaveRequest:
    """请假请求数据类"""
    employee_name: str    # 员工姓名
    leave_days: int       # 请假天数
    reason: str           # 请假原因
   
    def __str__(self):
        return f"{self.employee_name} 申请请假 {self.leave_days} 天，原因：{self.reason}"

步骤 3：实现具体处理者

实例

class TeamLeader(Approver):
    """组长 - 处理 1-3 天的请假"""
   
    def process_request(self, leave_request: LeaveRequest):
        if leave_request.leave_days <= 3:
            print(f"组长 {self.name} 批准了 {leave_request}")
        elif self.successor is not None:
            # 超出权限，传递给下一级
            print(f"组长 {self.name} 无权限，转交给上级处理")
            self.successor.process_request(leave_request)
        else:
            print("无法处理该请求：没有合适的审批者")

class DepartmentManager(Approver):
    """部门经理 - 处理 4-7 天的请假"""
   
    def process_request(self, leave_request: LeaveRequest):
        if 4 <= leave_request.leave_days <= 7:
            print(f"部门经理 {self.name} 批准了 {leave_request}")
        elif self.successor is not None:
            # 超出权限，传递给下一级
            print(f"部门经理 {self.name} 无权限，转交给上级处理")
            self.successor.process_request(leave_request)
        else:
            print("无法处理该请求：没有合适的审批者")

class GeneralManager(Approver):
    """总经理 - 处理 8 天以上的请假"""
   
    def process_request(self, leave_request: LeaveRequest):
        if leave_request.leave_days >= 8:
            print(f"总经理 {self.name} 批准了 {leave_request}")
        else:
            print("无法处理该请求：请假天数不符合要求")

步骤 4：构建责任链并测试

实例

def main():
    # 创建审批者
    team_leader = TeamLeader("张三")
    dept_manager = DepartmentManager("李四")
    general_manager = GeneralManager("王五")
   
    # 构建责任链：组长 → 部门经理 → 总经理
    team_leader.set_successor(dept_manager)
    dept_manager.set_successor(general_manager)
   
    # 创建测试用例
    test_cases = [
        LeaveRequest("小明", 2, "感冒发烧"),
        LeaveRequest("小红", 5, "回家探亲"),
        LeaveRequest("小刚", 10, "结婚休假"),
        LeaveRequest("小李", 15, "出国旅游")
    ]
   
    print("=== 请假审批测试 ===\n")
   
    # 处理所有请假请求
    for request in test_cases:
        print(f"处理请求: {request.employee_name} 请假 {request.leave_days} 天")
        team_leader.process_request(request)
        print("-" * 50)

if __name__ == "__main__":
    main()

运行结果

=== 请假审批测试 ===

处理请求: 小明 请假 2 天
组长 张三 批准了 小明 申请请假 2 天，原因：感冒发烧
--------------------------------------------------
处理请求: 小红 请假 5 天
组长 张三 无权限，转交给上级处理
部门经理 李四 批准了 小红 申请请假 5 天，原因：回家探亲
--------------------------------------------------
处理请求: 小刚 请假 10 天
组长 张三 无权限，转交给上级处理
部门经理 李四 无权限，转交给上级处理
总经理 王五 批准了 小刚 申请请假 10 天，原因：结婚休假
--------------------------------------------------
处理请求: 小李 请假 15 天
组长 张三 无权限，转交给上级处理
部门经理 李四 无权限，转交给上级处理
总经理 王五 批准了 小李 申请请假 15 天，原因：出国旅游
--------------------------------------------------

高级应用：Web 请求过滤器

责任链模式在 Web 开发中非常常见，特别是在中间件和过滤器链中。让我们实现一个简单的 HTTP 请求过滤器链。

步骤 1：定义请求和响应对象

实例

from dataclasses import dataclass
from typing import Dict, Any

@dataclass
class HttpRequest:
    """HTTP 请求对象"""
    method: str
    path: str
    headers: Dict[str, str]
    body: Any = None
    user: Dict[str, Any] = None

@dataclass  
class HttpResponse:
    """HTTP 响应对象"""
    status_code: int
    headers: Dict[str, str]
    body: Any = None

步骤 2：实现过滤器链

实例

class Filter(ABC):
    """过滤器抽象类"""
   
    def __init__(self):
        self.next_filter: Optional['Filter'] = None
   
    def set_next(self, next_filter: 'Filter'):
        """设置下一个过滤器"""
        self.next_filter = next_filter
   
    def do_filter(self, request: HttpRequest, response: HttpResponse) -> bool:
        """
        执行过滤操作
        返回 True 表示继续执行，False 表示中断
        """
        if self.handle(request, response):
            if self.next_filter is not None:
                return self.next_filter.do_filter(request, response)
            return True
        return False
   
    @abstractmethod
    def handle(self, request: HttpRequest, response: HttpResponse) -> bool:
        """具体的过滤处理逻辑"""
        pass

class AuthenticationFilter(Filter):
    """身份认证过滤器"""
   
    def handle(self, request: HttpRequest, response: HttpResponse) -> bool:
        print("执行身份认证...")
       
        # 模拟认证逻辑
        token = request.headers.get('Authorization')
        if token == "Bearer valid-token":
            request.user = {"id": 1, "name": "张三", "role": "user"}
            print("✓ 身份认证成功")
            return True
        else:
            response.status_code = 401
            response.body = {"error": "未授权的访问"}
            print("✗ 身份认证失败")
            return False

class LoggingFilter(Filter):
    """日志记录过滤器"""
   
    def handle(self, request: HttpRequest, response: HttpResponse) -> bool:
        print(f"记录请求日志: {request.method} {request.path}")
        return True  # 总是继续执行

class PermissionFilter(Filter):
    """权限检查过滤器"""
   
    def handle(self, request: HttpRequest, response: HttpResponse) -> bool:
        print("检查用户权限...")
       
        if request.user and request.user.get('role') == 'admin':
            print("✓ 权限检查通过")
            return True
        else:
            response.status_code = 403
            response.body = {"error": "权限不足"}
            print("✗ 权限检查失败")
            return False

步骤 3：使用过滤器链处理请求

实例

def process_http_request(request: HttpRequest) -> HttpResponse:
    """处理 HTTP 请求"""
   
    # 创建默认响应
    response = HttpResponse(
        status_code=200,
        headers={"Content-Type": "application/json"},
        body={"message": "请求成功"}
    )
   
    # 创建过滤器链
    auth_filter = AuthenticationFilter()
    logging_filter = LoggingFilter()
    permission_filter = PermissionFilter()
   
    # 构建过滤器链：日志 → 认证 → 权限
    logging_filter.set_next(auth_filter)
    auth_filter.set_next(permission_filter)
   
    # 执行过滤器链
    print("开始处理 HTTP 请求...")
    logging_filter.do_filter(request, response)
    print("请求处理完成")
   
    return response

# 测试不同的请求场景
def test_filter_chain():
    print("=== HTTP 过滤器链测试 ===\n")
   
    # 测试用例 1：有效的管理员请求
    print("测试用例 1：有效的管理员请求")
    request1 = HttpRequest(
        method="GET",
        path="/admin/data",
        headers={"Authorization": "Bearer valid-token"}
    )
    response1 = process_http_request(request1)
    print(f"响应状态码: {response1.status_code}")
    print(f"响应内容: {response1.body}\n")
   
    # 测试用例 2：未授权的请求
    print("测试用例 2：未授权的请求")
    request2 = HttpRequest(
        method="GET",
        path="/admin/data",
        headers={"Authorization": "Bearer invalid-token"}
    )
    response2 = process_http_request(request2)
    print(f"响应状态码: {response2.status_code}")
    print(f"响应内容: {response2.body}\n")

if __name__ == "__main__":
    test_filter_chain()

责任链模式的优缺点

优点

1. 降低耦合度

请求发送者不需要知道哪个对象会处理它的请求，接收者也不需要知道请求的全貌。

2. 增强灵活性

可以动态地添加、删除或重新排列处理者，而不影响客户端代码。

3. 简化对象职责

每个处理者只需关注自己职责范围内的请求，符合单一职责原则。

4. 便于扩展

新增处理者非常容易，只需要实现处理接口并添加到链中即可。

缺点

1. 请求可能未被处理

如果责任链配置不当，请求可能到达链的末端而没有被任何处理者处理。

2. 性能考虑

较长的责任链可能会影响性能，特别是在处理大量请求时。

3. 调试困难

请求的处理路径可能不明确，调试时难以跟踪请求的完整处理流程。

实际应用场景

1. 事件处理系统

图形界面中的事件冒泡机制就是责任链模式的典型应用。

实例

class Event:
    def __init__(self, name, target):
        self.name = name
        self.target = target
        self.handled = False

class Widget:
    def __init__(self, parent=None):
        self.parent = parent
   
    def handle_event(self, event):
        # 先自己处理，如果处理不了就传递给父组件
        if self.on_event(event):
            event.handled = True
        elif self.parent:
            self.parent.handle_event(event)
   
    def on_event(self, event):
        # 子类重写这个方法来实现具体的事件处理
        return False

2. 日志系统

不同级别的日志消息由不同的处理器处理。

实例

import logging

# 创建日志记录器
logger = logging.getLogger('my_app')
logger.setLevel(logging.DEBUG)

# 创建处理器链
console_handler = logging.StreamHandler()
file_handler = logging.FileHandler('app.log')

# 设置处理器级别
console_handler.setLevel(logging.WARNING)
file_handler.setLevel(logging.DEBUG)

# 添加到记录器
logger.addHandler(console_handler)
logger.addHandler(file_handler)

# 使用：DEBUG 和 INFO 级别只会记录到文件，WARNING 及以上会同时输出到控制台
logger.debug('调试信息')      # 只记录到文件
logger.warning('警告信息')    # 同时输出到控制台和文件

最佳实践和注意事项

1. 设置默认处理行为

确保责任链的末端有一个默认处理者，避免请求丢失。

实例

class DefaultHandler(Approver):
    """默认处理者 - 处理所有未被处理的请求"""
   
    def process_request(self, leave_request: LeaveRequest):
        print(f"默认处理者：无法处理 {leave_request}，请检查请求参数")

2. 控制链的长度

避免创建过长的责任链，这会降低性能并增加调试难度。

3. 明确的处理规则

确保每个处理者都有清晰的处理边界，避免职责重叠。

4. 考虑使用组合模式

对于复杂的责任链，可以考虑使用组合模式来管理处理者之间的关系。

总结

责任链模式是一个强大而灵活的设计模式，它通过将请求的发送者和接收者解耦，提供了一种优雅的方式来处理需要多个对象协作的场景。

关键要点

  1. 核心思想：让多个对象都有机会处理请求，避免请求发送者与接收者的耦合
  2. 适用场景：多个对象可以处理同一请求，但具体由哪个对象处理需要在运行时确定
  3. 实现要点：定义处理接口、构建处理链、传递请求
  4. Python 特色：利用 Python 的动态特性，可以更灵活地实现责任链模式