Python 责任链模式

Python 中的责任链模式（Chain of Responsibility Pattern）

责任链模式是一种行为型设计模式，其核心目的是：
将请求的发送者和接收者解耦，让多个对象都有机会处理请求。将这些对象连成一条链，并沿着这条链传递请求，直到有一个对象处理它为止。

形象比喻：就像请假审批——员工先提交给主管，如果主管权限不够，就转给经理；经理不够就转给总监……直到有人能审批或拒绝。

责任链模式的优点

• 解耦发送者和接收者：客户端无需知道谁来处理请求
• 动态组合处理链：运行时可以自由添加、移除、重组处理器
• 单一职责：每个处理器只负责自己能处理的请求
• 灵活性高：容易扩展新处理器

典型应用场景

• 日志系统（DEBUG → INFO → WARNING → ERROR 级别）
• Web 框架中间件（认证 → 授权 → 日志 → 路由 → 响应）
• 事件处理（如 GUI 事件冒泡）
• 审批流程（请假、报销）
• 过滤器链（敏感词过滤、XSS 防御）
• 异常处理（try-except 链）

Python 实现示例：日志处理器链

我们实现一个分级日志系统：不同级别日志交给不同处理器（控制台、文件、邮件）。

from abc import ABC, abstractmethod

# 抽象处理器（Handler）
class Logger(ABC):
    def __init__(self, next_logger=None):
        self.next_logger = next_logger  # 链中的下一个处理器

    def log(self, level: int, message: str):
        if self.can_handle(level):
            self.handle(message)
        # 如果有下一个处理器，继续传递
        if self.next_logger:
            self.next_logger.log(level, message)

    @abstractmethod
    def can_handle(self, level: int) -> bool:
        pass

    @abstractmethod
    def handle(self, message: str):
        pass

# 日志级别常量
DEBUG = 1
INFO = 2
WARNING = 3
ERROR = 4

# 具体处理器1：控制台日志（处理所有级别）
class ConsoleLogger(Logger):
    def can_handle(self, level: int) -> bool:
        return True  # 所有日志都打印到控制台

    def handle(self, message: str):
        print(f"[CONSOLE] {message}")

# 具体处理器2：文件日志（只处理 WARNING 及以上）
class FileLogger(Logger):
    def can_handle(self, level: int) -> bool:
        return level >= WARNING

    def handle(self, message: str):
        print(f"[FILE] 写入日志文件: {message}")

# 具体处理器3：邮件通知（只处理 ERROR 级别）
class EmailLogger(Logger):
    def can_handle(self, level: int) -> bool:
        return level == ERROR

    def handle(self, message: str):
        print(f"[EMAIL] 发送紧急邮件通知: {message}")

# 构建责任链
def build_chain():
    # 链顺序：控制台 → 文件 → 邮件
    email_logger = EmailLogger()
    file_logger = FileLogger(email_logger)
    console_logger = ConsoleLogger(file_logger)
    return console_logger

# 客户端使用
if __name__ == "__main__":
    logger_chain = build_chain()

    logger_chain.log(DEBUG, "系统启动正常")          # 只控制台
    logger_chain.log(INFO, "用户登录成功")           # 只控制台
    logger_chain.log(WARNING, "磁盘空间不足")       # 控制台 + 文件
    logger_chain.log(ERROR, "数据库连接失败！")      # 控制台 + 文件 + 邮件

输出：

[CONSOLE] 系统启动正常
[CONSOLE] 用户登录成功
[CONSOLE] 磁盘空间不足
[FILE] 写入日志文件: 磁盘空间不足
[CONSOLE] 数据库连接失败！
[FILE] 写入日志文件: 数据库连接失败！
[EMAIL] 发送紧急邮件通知: 数据库连接失败！

Pythonic 更简洁实现：函数式责任链

Python 支持高阶函数，可以用列表+循环实现轻量责任链。

from typing import Callable, List

# 定义处理器类型
Handler = Callable[[str], bool]  # 返回 True 表示已处理，不再向下传递

def console_handler(message: str) -> bool:
    print(f"[CONSOLE] {message}")
    return False  # 继续向下传递

def file_handler(message: str) -> bool:
    if "错误" in message or "警告" in message:
        print(f"[FILE] 记录严重日志: {message}")
    return False

def email_handler(message: str) -> bool:
    if "致命" in message or "失败" in message:
        print(f"[EMAIL] 紧急通知: {message}")
        return True  # 已处理，停止传递
    return False

# 构建链
handlers: List[Handler] = [console_handler, file_handler, email_handler]

def log_message(message: str):
    for handler in handlers:
        if handler(message):
            break  # 某个处理器返回 True，停止链

# 使用
log_message("程序启动")
log_message("用户登录")
log_message("内存警告")
log_message("数据库连接失败")

真实项目常见例子：Web 框架中间件

Flask/Django/FastAPI 的中间件本质就是责任链：

class Middleware:
    def __init__(self, next_middleware=None):
        self.next = next_middleware

    def __call__(self, request):
        # 前置处理
        self.process_request(request)
        # 传递给下一个
        if self.next:
            response = self.next(request)
        else:
            response = handle_request(request)  # 最终处理
        # 后置处理
        return self.process_response(response)

class AuthMiddleware(Middleware):
    def process_request(self, request):
        if not request.user.is_authenticated:
            return redirect("/login")

class LoggingMiddleware(Middleware):
    def process_request(self, request):
        print(f"请求: {request.path}")

# 构建链：Logging → Auth → 最终处理
app = LoggingMiddleware(AuthMiddleware())

责任链模式结构总结

角色	说明
Handler	抽象处理器（Logger），定义处理接口和链指针
ConcreteHandler	具体处理器（ConsoleLogger 等）
Client	发送请求到链首（logger_chain.log()）

责任链模式 vs 其他模式对比

模式	目的	处理方式	典型场景
责任链	请求沿链传递，直到被处理	动态链、可能不处理	中间件、日志、审批
装饰器	层层增强功能	固定链、全部执行	功能包装
命令	封装请求为对象	支持队列、撤销	操作历史
观察者	广播通知多个对象	一对多广播	事件系统

Python 中的实用建议

• 简单场景：用函数列表实现（最轻量）
• 复杂场景：用类实现，支持状态和动态重组
• 停止传递：处理器返回 True/None/Response 表示是否终止链
• 常见库实现：
• Django 中间件
• Flask 的 @app.before_request
• FastAPI 的 Dependencies 和 Middlewares

注意事项

• 确保链的顺序正确（优先级高的放前面）
• 避免循环链（会导致无限递归）
• 如果没有处理器处理请求，要有默认行为（否则请求丢失）
• 多线程时注意线程安全

责任链模式是构建可插拔、可扩展处理流程的利器，在 Web 框架、日志系统、过滤器中无处不在。

如果你想看更多实战例子（如 HTTP 请求中间件链、审批流程系统、敏感词过滤链），或者如何与装饰器模式结合，欢迎继续问！