Chain-of-responsibility 职责链模式

设计模式 2020-12-30 6601 字 253 浏览 点赞

起步

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

如果没有职责链

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

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

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.

就两点:

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

可读性更好的职责链模式

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

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 创作,采用 知识共享署名 3.0,可自由转载、引用,但需署名作者且注明文章出处。

还不快抢沙发

添加新评论