Python 设计模式——责任链模式（Chain of Responsibility Pattern）详解

本文面向零基础新手，从概念到“链”的构建与传递，从简单示例到多级审批，全方位讲解责任链模式。

---

一、什么是责任链模式？

1.1 通俗理解

责任链模式是一种行为型设计模式，核心思想是：

把多个“处理者”排成一条链；请求从链头进入，依次经过每个处理者；每个处理者要么自己处理并结束，要么把请求传给下一个处理者，直到某个人处理了或链结束。这样发送请求的一方不需要知道具体由谁处理，只需把请求丢给链头即可。

可以这样类比：

• 现实中的责任链：请假 1 天找组长批，3 天找经理批，7 天找总监批。你只提交一张请假单，先到组长；组长能批就批，不能批就转给经理；经理能批就批，不能批再转总监。你不需要知道“该找谁”，只要把单子交给第一个人，请求会沿着链一路传下去，直到有人处理。
• 代码里：一个请求（如“审批请假”“校验表单”“记录日志”）可能被多种处理者处理；把这些处理者串成链，请求从链头进入，谁符合条件谁处理，不符合就传给下一个。发送方只依赖“链头”，处理者之间通过“下一个”连接，解耦发送方与具体处理者。

所以，责任链模式解决的是：多个对象都有机会处理请求，但不想让发送方写死“该调谁”，而是把处理者串成链，让请求沿链传递，由链上的某一位（或几位）处理。

1.2 为什么需要责任链？

场景一：请求的接收者不唯一、顺序不固定

例如：一个事件要依次经过“日志 → 权限 → 业务”；或一个审批要“组长 → 经理 → 总监”。若发送方写死调用顺序和具体类，以后加一个环节或改顺序就要改发送方。用责任链：发送方只调“链头”，链上每个节点决定“我处理还是传给下一个”，扩展链即可，不用改发送方。

场景二：发送方不想知道具体处理者

发送方只关心“把请求发出去”，不关心最终谁处理、有多少环节。责任链把“谁在处理”隐藏在链里，发送方只依赖抽象处理者（链头）。

场景三：动态调整处理顺序或处理者

链可以在运行时组装：有时是 A→B→C，有时是 A→C→B，或动态增删节点。发送方代码不变，只换链的组成。

1.3 适用场景（什么时候用？）

场景	说明
多级审批、工作流	请假、报销、工单，按级别或规则依次传递。
事件/请求的多层处理	日志、权限、限流、业务逻辑，按链传递。
多种校验/过滤	表单校验：非空 → 格式 → 长度 → 业务规则，一环不过就返回，过了就下一环。
发送方与处理者解耦	发送方不关心具体谁处理、有多少环节。

简单记忆：多个对象都可能处理同一类请求，且希望按“链”传递、由链上某节点处理时，用责任链。

---

二、责任链模式的结构（两个角色）

角色	说明
处理者（Handler）	抽象类/接口，定义“处理请求”的方法，并持有一个“下一个处理者”（successor）；在处理方法里，若自己不能/不愿处理，就调用下一个处理者的处理方法。
具体处理者（ConcreteHandler）	实现“处理请求”的逻辑：若轮到自己能处理就处理并结束（可选：不再传递）；否则调用 next.handle(request) 把请求传给下一个。

关系可以理解为：

• 客户把请求交给链头（第一个具体处理者）。
• 链上每个处理者：能处理就处理并结束；不能处理就 self._next.handle(request)，直到某节点处理了或链尾（next 为 None）结束。

---

三、示例一：请假审批（最经典）

需求：请假 1 天组长批，3 天内经理批，7 天内总监批，超过 7 天不批。请求带“天数”，沿链传递，谁符合谁批。

3.1 请求对象（可选，便于扩展）

class LeaveRequest:
    def __init__(self, name: str, days: int):
        self.name = name
        self.days = days

3.2 抽象处理者 + 具体处理者

from abc import ABC, abstractmethod

class Handler(ABC):
    """处理者：持有下一个处理者，定义处理接口"""
    def __init__(self):
        self._next = None

    def set_next(self, handler: "Handler") -> "Handler":
        self._next = handler
        return handler  # 方便链式 set_next

    def handle(self, request: LeaveRequest) -> str:
        """子类可重写：能处理就处理，否则传下去"""
        if self._next:
            return self._next.handle(request)
        return "无人处理"

3.3 具体处理者：组长、经理、总监

class TeamLeadHandler(Handler):
    def handle(self, request: LeaveRequest) -> str:
        if request.days <= 1:
            return f"组长批准 {request.name} 请假 {request.days} 天"
        if self._next:
            return self._next.handle(request)
        return "无人处理"

class ManagerHandler(Handler):
    def handle(self, request: LeaveRequest) -> str:
        if request.days <= 3:
            return f"经理批准 {request.name} 请假 {request.days} 天"
        if self._next:
            return self._next.handle(request)
        return "无人处理"

class DirectorHandler(Handler):
    def handle(self, request: LeaveRequest) -> str:
        if request.days <= 7:
            return f"总监批准 {request.name} 请假 {request.days} 天"
        return "请假天数超过 7 天，不予批准"

3.4 组装链并发送请求

# 组装链：组长 -> 经理 -> 总监
team_lead = TeamLeadHandler()
manager = ManagerHandler()
director = DirectorHandler()
team_lead.set_next(manager).set_next(director)

# 请求只交给链头
req1 = LeaveRequest("张三", 1)
req2 = LeaveRequest("李四", 3)
req3 = LeaveRequest("王五", 5)
req4 = LeaveRequest("赵六", 10)

print(team_lead.handle(req1))  # 组长批准 张三 请假 1 天
print(team_lead.handle(req2))  # 经理批准 李四 请假 3 天
print(team_lead.handle(req3))  # 总监批准 王五 请假 5 天
print(team_lead.handle(req4))  # 请假天数超过 7 天，不予批准

要点：客户只调 team_lead.handle(request)，不关心后面是经理还是总监；谁符合条件谁处理，不符合就自动传到下一个。

---

四、示例二：日志级别链（按级别选择处理者）

需求：日志有 DEBUG、INFO、ERROR；每个处理者只处理自己级别及以上的日志（如 ERROR 处理者只处理 ERROR）。请求带级别和消息，沿链传递，直到某处理者“认领”。

class LogLevel:
    DEBUG, INFO, ERROR = 0, 1, 2

class LogRequest:
    def __init__(self, level: int, message: str):
        self.level = level
        self.message = message

class LogHandler(ABC):
    def __init__(self, level: int):
        self.level = level
        self._next = None

    def set_next(self, h: "LogHandler"):
        self._next = h
        return h

    def handle(self, req: LogRequest) -> bool:
        if req.level >= self.level:
            self._write(req)
            return True
        if self._next:
            return self._next.handle(req)
        return False

    def _write(self, req: LogRequest):
        pass

class DebugHandler(LogHandler):
    def __init__(self):
        super().__init__(LogLevel.DEBUG)

    def _write(self, req: LogRequest):
        print(f"[DEBUG] {req.message}")

class InfoHandler(LogHandler):
    def __init__(self):
        super().__init__(LogLevel.INFO)

    def _write(self, req: LogRequest):
        print(f"[INFO] {req.message}")

class ErrorHandler(LogHandler):
    def __init__(self):
        super().__init__(LogLevel.ERROR)

    def _write(self, req: LogRequest):
        print(f"[ERROR] {req.message}")

# 链：DEBUG -> INFO -> ERROR（通常只用一个，这里演示链）
chain = DebugHandler()
chain.set_next(InfoHandler()).set_next(ErrorHandler())
chain.handle(LogRequest(LogLevel.INFO, "用户登录"))  # [INFO] 用户登录
chain.handle(LogRequest(LogLevel.ERROR, "数据库异常"))  # [ERROR] 数据库异常

---

五、示例三：表单校验链（一环不过就失败）

需求：用户名校验：非空 → 长度 3～10 → 仅字母数字。任一环不通过就返回错误，通过则传下一环。

class ValidationRequest:
    def __init__(self, value: str):
        self.value = value
        self.error = None

class Validator(ABC):
    def __init__(self):
        self._next = None

    def set_next(self, v: "Validator"):
        self._next = v
        return v

    def validate(self, req: ValidationRequest) -> bool:
        if not self._check(req):
            return False
        if self._next:
            return self._next.validate(req)
        return True

    def _check(self, req: ValidationRequest) -> bool:
        return True

class NotEmptyValidator(Validator):
    def _check(self, req: ValidationRequest) -> bool:
        if not req.value or not req.value.strip():
            req.error = "不能为空"
            return False
        return True

class LengthValidator(Validator):
    def __init__(self, min_len: int, max_len: int):
        super().__init__()
        self.min_len = min_len
        self.max_len = max_len

    def _check(self, req: ValidationRequest) -> bool:
        n = len(req.value)
        if n < self.min_len or n > self.max_len:
            req.error = f"长度需在 {self.min_len}～{self.max_len} 之间"
            return False
        return True

class AlnumValidator(Validator):
    def _check(self, req: ValidationRequest) -> bool:
        if not req.value.isalnum():
            req.error = "只能包含字母和数字"
            return False
        return True

# 组装链
v = NotEmptyValidator()
v.set_next(LengthValidator(3, 10)).set_next(AlnumValidator())

req = ValidationRequest("ab")
print(v.validate(req), req.error)  # False 长度需在 3～10 之间
req2 = ValidationRequest("abc123")
print(v.validate(req2), req2.error)  # True None

要点：请求对象带 error，某环节不通过时设置 error 并返回 False，链停止；全部通过才返回 True。

---

六、责任链的常见变体

6.1 链尾默认处理

若希望“没人处理时”有一个统一结果，可以在链尾放一个“默认处理者”，不设置 _next，在 handle 里返回“无法处理”或默认逻辑。

6.2 请求只被一个节点处理 vs 被多个节点处理

• 只一个：某节点处理完后直接 return，不再调用 _next.handle()（如审批：谁批了谁返回）。
• 多个：某节点处理完后仍调用 _next.handle()，让后面节点继续（如日志链：DEBUG、INFO、ERROR 都打一遍）。按业务选择。

6.3 链式 set_next 便于组装

handler_a.set_next(handler_b).set_next(handler_c) 让 set_next 返回传入的 handler，便于一行串起整条链。

---

七、责任链 vs 其他模式

模式	目的	与责任链的区别
责任链	请求沿链传递，由某节点处理	强调“链”“传递”“多节点之一处理”。
装饰器	层层包装，每层都处理并转发	通常每层都参与，不“选一个”；责任链常“选一个处理就停”。
状态	根据状态切换行为	是对象内部状态变化；责任链是多个对象组成链。

简单记忆：责任链 = 一条链 + 请求传递 + 某节点处理（或传下去）。

---

八、常见问题与注意点

8.1 保证链上有节点能处理

若所有节点都不处理且没有默认节点，请求会“掉空”。可以在链尾放一个“兜底处理者”，或保证业务上至少有一个节点会处理。

8.2 避免成环

链应是单向的，不要让某个节点的 _next 指回前面，否则会死循环。

8.3 谁组装链

可以在应用启动时、配置里、或工厂里组装；发送方只拿“链头”，不关心链的构成。

---

九、小结

• 责任链模式：把多个处理者排成一条链，请求从链头进入，依次传递；每个处理者能处理就处理并结束，不能处理就传给下一个，从而让发送方与具体处理者解耦，且便于增删、调序节点。
• 两个角色：处理者（抽象，含 set_next、handle）、具体处理者（实现处理逻辑，决定处理还是传递）。
• 典型用途：多级审批、工作流、日志级别、表单多步校验、事件管道等。
• 实现要点：每个处理者持有一个“下一个”；handle 里能处理就 return，不能就 self._next.handle(request)。

建议先写“请假审批”链，再写“表单校验”链，体会“请求沿链传递、谁符合谁处理”，这样对责任链模式会掌握得比较扎实。