Python 中的责任链模式（Chain of Responsibility Pattern）
责任链模式是一种行为型设计模式，其核心目的是：
将请求的发送者和接收者解耦，让多个对象都有机会处理请求。将这些对象连成一条链，并沿着这条链传递请求，直到有一个对象处理它为止。
形象比喻：就像请假审批——员工先提交给主管，如果主管权限不够，就转给经理；经理不够就转给总监……直到有人能审批或拒绝。
责任链模式的优点
- 解耦发送者和接收者：客户端无需知道谁来处理请求
- 动态组合处理链：运行时可以自由添加、移除、重组处理器
- 单一职责：每个处理器只负责自己能处理的请求
- 灵活性高：容易扩展新处理器
典型应用场景
- 日志系统（DEBUG → INFO → WARNING → ERROR 级别）
- Web 框架中间件（认证 → 授权 → 日志 → 路由 → 响应）
- 事件处理（如 GUI 事件冒泡）
- 审批流程（请假、报销）
- 过滤器链（敏感词过滤、XSS 防御）
- 异常处理（try-except 链）
Python 实现示例：日志处理器链
我们实现一个分级日志系统：不同级别日志交给不同处理器（控制台、文件、邮件）。
fromabcimportABC,abstractmethod# 抽象处理器（Handler）classLogger(ABC):def__init__(self,next_logger=None):self.next_logger=next_logger# 链中的下一个处理器deflog(self,level:int,message:str):ifself.can_handle(level):self.handle(message)# 如果有下一个处理器，继续传递ifself.next_logger:self.next_logger.log(level,message)@abstractmethoddefcan_handle(self,level:int)->bool:pass@abstractmethoddefhandle(self,message:str):pass# 日志级别常量DEBUG=1INFO=2WARNING=3ERROR=4# 具体处理器1：控制台日志（处理所有级别）classConsoleLogger(Logger):defcan_handle(self,level:int)->bool:returnTrue# 所有日志都打印到控制台defhandle(self,message:str):print(f"[CONSOLE]{message}")# 具体处理器2：文件日志（只处理 WARNING 及以上）classFileLogger(Logger):defcan_handle(self,level:int)->bool:returnlevel>=WARNINGdefhandle(self,message:str):print(f"[FILE] 写入日志文件:{message}")# 具体处理器3：邮件通知（只处理 ERROR 级别）classEmailLogger(Logger):defcan_handle(self,level:int)->bool:returnlevel==ERRORdefhandle(self,message:str):print(f"[EMAIL] 发送紧急邮件通知:{message}")# 构建责任链defbuild_chain():# 链顺序：控制台 → 文件 → 邮件email_logger=EmailLogger()file_logger=FileLogger(email_logger)console_logger=ConsoleLogger(file_logger)returnconsole_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 支持高阶函数，可以用列表+循环实现轻量责任链。
fromtypingimportCallable,List# 定义处理器类型Handler=Callable[[str],bool]# 返回 True 表示已处理，不再向下传递defconsole_handler(message:str)->bool:print(f"[CONSOLE]{message}")returnFalse# 继续向下传递deffile_handler(message:str)->bool:if"错误"inmessageor"警告"inmessage:print(f"[FILE] 记录严重日志:{message}")returnFalsedefemail_handler(message:str)->bool:if"致命"inmessageor"失败"inmessage:print(f"[EMAIL] 紧急通知:{message}")returnTrue# 已处理，停止传递returnFalse# 构建链handlers:List[Handler]=[console_handler,file_handler,email_handler]deflog_message(message:str):forhandlerinhandlers:ifhandler(message):break# 某个处理器返回 True，停止链# 使用log_message("程序启动")log_message("用户登录")log_message("内存警告")log_message("数据库连接失败")真实项目常见例子：Web 框架中间件
Flask/Django/FastAPI 的中间件本质就是责任链：
classMiddleware:def__init__(self,next_middleware=None):self.next=next_middlewaredef__call__(self,request):# 前置处理self.process_request(request)# 传递给下一个ifself.next:response=self.next(request)else:response=handle_request(request)# 最终处理# 后置处理returnself.process_response(response)classAuthMiddleware(Middleware):defprocess_request(self,request):ifnotrequest.user.is_authenticated:returnredirect("/login")classLoggingMiddleware(Middleware):defprocess_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 请求中间件链、审批流程系统、敏感词过滤链），或者如何与装饰器模式结合，欢迎继续问！
