起步

职责链模式属于行为型。它包含一系列处理对象，每个对象只处理它能够处理的请求——也就是每个处理对象的职责。而之所以叫职责链，是因为多个有职责的对象像链表似的连在一起。

如果没有职责链

现在我们要做一个内容审查的功能：对用户发布的内容进行关键字检查，要求不准有色情词汇，不准有反动言论，不准说脏话；如果存在上述“不准”，就禁止内容发布。

如果这个世界上没有所谓的职责链模式，代码也许会呈现如下：

def has_sexy_words(contents: str) -> bool:
    """
        是否含有色情词汇
    """
    keyword = ... # 假装色情词汇
    return keyword in contents

def has_reactionary_words(contents: str) -> bool:
    """
        是否含有反动言论
    """
    keyword = ... # 假装反动词汇
    return keyword in contents

def has_insulted_words(contents: str) -> bool:
    """
        是否有脏话
    """
    keyword = ... # 假装脏话
    return keyword in contents

def is_content_valid(contents: str) -> bool:
    """
        检查内容是否有效
    """
    if has_sexy_words(contents):
        return False
    
    if has_reactionary_words(contents):
        return False
    
    if has_insulted_words(contents):
        return False
    
    return True

class Twitter(object):
    def make_contents(self, user) -> str:
        contents = user.make_contents()
        return contents
    
    def can_post_contents(self) -> bool:
        """
            是否允许发布内容
        """
        contents = self.make_contents()
        return is_content_valid(contents)

这样就存在两个问题，如果词汇检查（has_xxx_words()）是很复杂的操作，单靠函数是 cover 不住的；其次是，假如将来要增删检查函数，肯定要修改 is\_content\_valid()，破坏了开闭原则。

职责链模式

如同策略模式，我们应该把每个检查封装成一个处理对象。

但与策略模式不同的是，策略模式一次只从众多策略对象中选出一个来使用，而职责链模式需要把处理对象串起来，让需要被处理的数据从链表头，一路走到链表尾（如果可以的话）。代码如下：

from abc import abstractmethod

class BaseChecker(object):
    def __init__(self) -> None:
        self.next = None
    
    def add_checker(self, checker):
        # 避免出现循环链表，先要清除 next 指向的对象
        checker.next = None

        # 将 checker 追加到职责链的末尾
        cur = self
        while cur.next:
            cur = cur.next
        cur.next = checker
    
    def is_valid(self, contents: str) -> bool:
        # 如果存在 不良 词汇, 立即返回无效(False)
        if self.has_xxx_words(contents):
            return False
        # 如果职责链上还有其他 checker, 交给下一个 checker 检查
        if self.next is not None:
            self.next.is_valid(contents)
        # 如果职责链上没有 checker 了, 说明 contents 有效
        return True

    @abstractmethod  
    def has_xxx_words(self, contents: str) -> bool:
        pass

class SexyChecker(BaseChecker):
    def has_xxx_words(self, contents: str) -> bool:
        keyword = ... # 假装色情词汇
        return keyword in contents

class ReactionaryChecker(BaseChecker):
    def has_xxx_words(self, contents: str) -> bool:
        keyword = ... # 假装反动词汇
        return keyword in contents

class InsultedChecker(BaseChecker):
    def has_xxx_words(self, contents: str) -> bool:
        keyword = ... # 假装脏话
        return keyword in contents

class Twitter(object):
    def make_contents(self, user) -> str:
        contents = user.make_contents()
        return contents
    
    def can_post_contents(self) -> bool:
        """
            是否允许发布内容
        """
        # 添加 checker
        checker = SexyChecker()
        checker.add_checker( ReactionaryChecker() )
        checker.add_checker( InsultedChecker() )
        
        contents = self.make_contents()
        return checker.is_valid(contents)

当调用 checker.is_valid() ，依次触发的 checker 为：SexyChecker -> ReactionaryChecker -> InsultedChecker。这就是一个职责链，只要其中一个处理对象发现内容不合规（视作能够处理此次请求），就立即终止检查并返回检查结果。

此外，checker.add_checker() 用来在职责链上添加处理对象。这是职责链模式的组成部分，也是它，使得职责链中的处理对象可以按需搭配，提升了代码的扩展性。

就在这里总结一下什么是职责链模式，我们看看维基百科怎么说：

In object-oriented design, the chain-of-responsibility pattern is a design pattern consisting of a source of command objects and a series of processing objects. Each processing object contains logic that defines the types of command objects that it can handle; the rest are passed to the next processing object in the chain. A mechanism also exists for adding new processing objects to the end of this chain.

就两点：

每个处理对象只处理它能够处理的请求，如果处理不了，将请求在链中传递下去；
可以将处理对象追加到职责链中。

可读性更好的职责链模式

老实说，上面的职责链模式代码可读性不怎么好，尤其对不熟悉链表的人不友好。另一种常见的、可读性更高的实现方式是：将处理对象和链解耦，用列表存储处理对象。

from typing import List
from abc import abstractmethod

class BaseChecker(object):
    """
        处理对象的基类
    """
    @abstractmethod  
    def has_xxx_words(self, contents: str) -> bool:
        pass

class SexyChecker(BaseChecker):
    def has_xxx_words(self, contents: str) -> bool:
        keyword = ... # 假装色情词汇
        return keyword in contents

class ReactionaryChecker(BaseChecker):
    def has_xxx_words(self, contents: str) -> bool:
        keyword = ... # 假装反动词汇
        return keyword in contents

class InsultedChecker(BaseChecker):
    def has_xxx_words(self, contents: str) -> bool:
        keyword = ... # 假装脏话
        return keyword in contents

class CheckerChain(object):
    """
        链
    """
    def __init__(self) -> None:
        self.checkers: List[BaseChecker] = []
    
    def add_checker(self, checker):
        self.checkers.append(checker)
    
    def is_valid(self, contents: str) -> bool:
        if not self.checkers:
            raise ValueError("empty chain")
        
        for checker in self.add_checker:
            if checker.has_xxx_words(contents):
                return False
        return True


class Twitter(object):
    def make_contents(self, user) -> str:
        contents = user.make_contents()
        return contents
    
    def can_post_contents(self) -> bool:
        """
            是否允许发布内容
        """
        # usage
        chain = CheckerChain()
        chain.add_checker( SexyChecker() )
        chain.add_checker( ReactionaryChecker() )
        chain.add_checker( InsultedChecker() )
        
        contents = self.make_contents()
        return chain.is_valid(contents)

总结

如果你同时熟悉装饰器模式，你会发现它跟职责链模式极为相似。它们的区别在于：装饰器模式中，所有的装饰器会依次处理传进来的请求；职责链模式下，严格要求整条链中只有一个处理器会处请求，处理之后马上返回。

但是，设计模式的初衷是让代码高内聚低耦合、扩展性高、可读性好，而不是冰冷的枷锁要将 coder 们束缚其中。所以在开源世界里，你会发现有的大牛在实现职责链模式时，允许职责链上一次有多个处理器处理请求。即，处理器能处理并处理请求后，不会立即退出，而是将请求向下传递，让其他能够处理这个请求的处理器处理（绕口;-)）。

有关心情

Guan

有关心情

Chain-of-responsibility 职责链模式

起步

如果没有职责链

职责链模式

可读性更好的职责链模式

总结

参考