Skip to content

设计模式

1 单例模式(Singleton Pattern)

同步可参考个人另一博文:《Python:用于有效对象管理的单例模式》

单例模式确保一个类只有一个实例,并提供一个全局访问点。它适用于需要唯一对象的场景,如:数据库连接、配置文件管理等。

Python 中实现单例模式的方式有多种,以下是几种常见的实现方法:

1.1 使用 __new__ 方法

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
class Singleton:
    _instance = None

    def __new__(cls, *args, **kwargs):
        if not cls._instance:
            cls._instance = super().__new__(cls)
        return cls._instance


# 示例
s1 = Singleton()
s2 = Singleton()
print(s1 is s2)  # 输出: True

或者:

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
class Singleton:
    def __new__(cls, *args, **kwargs):
        if not hasattr(cls, '_instance'):
            cls._instance = super().__new__(cls)
        return cls._instance


class MyClass(Singleton):
    pass


my1 = MyClass()
my2 = MyClass()
print(my1 is my2)  # 输出: True

1.2 使用装饰器

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
def singleton(cls):
    instances = {}

    def get_instance(*args, **kwargs):
        if cls not in instances:
            instances[cls] = cls(*args, **kwargs)
        return instances[cls]

    return get_instance


@singleton
class Singleton:
    pass


# 示例
s1 = Singleton()
s2 = Singleton()
print(s1 is s2)  # 输出: True

1.3 共享属性

共享属性单例的原理基于 Borg 模式,该模式通过共享状态(即共享属性)来实现单例效果。具体来说,每个实例对象的属性字典(dict )都指向同一个共享状态字典。这意味着,无论创建多少实例,这些实例的属性都共享相同的状态。

工作原理:

  • 类属性共享状态:类中定义一个共享的字典来保存状态;
  • 重写 __new__ 方法:在 __new__ 方法中,将每个新实例的 __dict__ 指向共享状态字典;
  • 实例共享属性:所有实例共享相同的属性和值,修改一个实例的属性会影响所有实例;

实现了状态共享,不同于传统单例模式,该模式允许多个实例存在,但它们共享状态。如,示例:

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
class Borg:
    _shared_state = {}  # 类属性,共享所有实例的状态

    def __new__(cls, *args, **kwargs):
        obj = super().__new__(cls)
        obj.__dict__ = cls._shared_state  # 使所有实例的 __dict__ 指向相同的字典
        return obj


class MyClass(Borg):
    def __init__(self, name):
        self.name = name


a = MyClass("Alice")
b = MyClass("Bob")

print(a.name)  # 输出: Bob
print(b.name)  # 输出: Bob
print(a is b)  # 输出: False

在这个例子中,a 和 b 虽然是不同的实例,但由于它们共享同一个 __dict__,因此属性 name 的修改在所有实例中都会反映出来。

1.4 使用元类(MetaClass)

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
class SingletonMeta(type):
    _instances = {}

    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            cls._instances[cls] = super().__call__(*args, **kwargs)
        return cls._instances[cls]


class Singleton(metaclass=SingletonMeta):
    pass


# 示例
s1 = Singleton()
s2 = Singleton()
print(s1 is s2)  # 输出: True

1.5 使用模块(import方法)

模块在 Python 中天然就是单例的,因为模块在第一次导入时会被缓存,后续的导入都是直接从缓存中获取。

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
# singleton_module.py
class Singleton:
    pass


singleton = Singleton()

# 示例
from singleton_module import singleton

s1 = singleton
s2 = singleton
print(s1 is s2)  # 输出: True

1.6 线程安全单例

线程安全的单例模式用于在多线程环境下确保单例对象的唯一性。如果多个线程同时尝试创建单例实例,可能会导致多个实例被创建,这违反了单例模式的设计初衷。通过确保线程安全,可以避免数据不一致、资源浪费等问题,确保系统的稳定性和正确性。

线程安全的单例可以通过以下方法实现:

  • 使用锁(Lock):在创建实例时加锁,确保只有一个线程能创建实例;
  • 双重检查锁(Double-Checked Locking):先检查实例是否存在,如果不存在则加锁创建;
  • 模块级别单例:Python模块本身是线程安全的,导入时只会执行一次初始化;

使用锁适用于需要严格控制访问的场景,双重检查锁更高效,而模块级别单例则简单直接。

线程安全:示例

  • 使用锁(Lock)
Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
import threading


class Singleton:
    _instance = None
    _lock = threading.Lock()

    def __new__(cls, *args, **kwargs):
        with cls._lock:
            if not cls._instance:
                # cls._instance = super(Singleton, cls).__new__(cls)
                cls._instance = super().__new__(cls)
        return cls._instance


singleton1 = Singleton()
singleton2 = Singleton()

print(singleton1 is singleton2)  # 输出: True

说明: 使用 threading.Lock() 来确保在实例化时只有一个线程可以进入创建实例的代码块。

【备注】: super(Singleton, cls).__new__(cls) 可以简化为 super().__new__(cls) 在 Python 3 中,super() 不需要显式传递类和实例对象。它会自动解析为当前类的父类并返回该父类的 __new__ 方法。因此,两者效果相同,但 super().__new__(cls) 更简洁。

  • 双重检查锁(Double-Checked Locking)
Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
import threading


class Singleton:
    _instance = None
    _lock = threading.Lock()

    def __new__(cls, *args, **kwargs):
        if not cls._instance:
            with cls._lock:
                if not cls._instance:
                    cls._instance = super().__new__(cls)
        return cls._instance


singleton1 = Singleton()
singleton2 = Singleton()

print(singleton1 is singleton2)  # 输出: True

说明: 双重检查锁先检查 _instance 是否存在,避免了不必要的加锁开销,提升性能。

  • 模块级别单例
Python
1
2
3
4
5
6
# singleton.py
class Singleton:
    pass


singleton = Singleton()
Python
1
2
3
4
5
6
7
# main.py
from singleton import singleton

singleton1 = singleton
singleton2 = singleton

print(singleton1 is singleton2)  # 输出: True

说明: 在 Python 中,模块本身是线程安全的,导入时只会执行一次初始化,后续导入不会再次初始化。

2 工厂模式(Factory Pattern)

同步可参考个人另一博文:《Python: 开始使用工厂模式设计》

工厂模式是一种创建型设计模式,用于定义一个接口或抽象类来创建对象,但将实例化推迟到子类中。其核心思想是通过工厂方法来处理对象的创建,而不是在代码中直接调用构造函数。这使得代码更加灵活和可扩展,尤其是在需要通过不同条件创建不同类型对象时。 以下是常见工厂实现方式:

  • 简单工厂模式(Simple Factory)
  • 概念: 由一个工厂类根据传入的参数决定创建哪种具体产品;
  • 优点: 客户端与产品的创建分离,客户端不需要知道产品创建的逻辑,只需要消费该产品即可;
  • 缺点: 工厂类集成了所有产品的创建逻辑,当工厂类出现问题,所有产品都会出现问题;还有当新增加产品都会修改工厂类,违背开闭原则 ;
  • 应用场景: 创建单一产品时;

  • 工厂方法模式(Factory Method)

  • 概念: 定义一个创建对象的接口,由子类决定实例化哪个类;
  • 优点: 遵循开放/封闭原则,更容易扩展;
  • 缺点: 需要为每种产品都创建一个具体工厂类,类的数量会增加;
  • 应用场景: 当有多个产品类型,需要根据具体子类来确定产品实例时;

  • 抽象工厂模式(Abstract Factory)

  • 概念: 提供一个接口,创建一系列相关或相互依赖的对象,而无需指定具体类。
  • 优点: 可以创建一系列相关的产品,确保产品之间的一致性。
  • 缺点: 复杂度较高,增加新产品族时需要扩展抽象工厂。
  • 应用场景: 需要创建多个相关产品对象时,如跨平台开发。

  • 总结

  • 简单工厂: 适合单一产品创建。
  • 工厂方法: 适合多个产品类型创建,扩展性更好。
  • 抽象工厂: 适合创建多个相关产品族,确保一致性。

2.1 简单工厂模式

简单工厂模式通常包含一个工厂类,该类具有一个创建方法,根据传入的参数决定实例化哪个具体类。

python_features2.1.png

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
from abc import ABC, abstractmethod


class Phone(ABC):
    @abstractmethod
    def make(self):
        pass


class Huawei(Phone):
    def make(self):
        return "Manufacture of Huawei mobile phones!"


class Iphone(Phone):
    def make(self):
        return "Manufacture of iPhone mobile phones!"


class PhoneFactory:
    @staticmethod
    def create_phones(phone_type):
        # Python 3.10+:match-case
        # match phone_type:
        #     case "huawei":
        #         return Huawei()
        #     case "iphone":
        #         return Iphone()
        #     case _:
        #         print("No Phone Type.")
        if phone_type == "huawei":
            return Huawei()
        elif phone_type == "iphone":
            return Iphone()
        else:
            print("No Phone Type.")


# 使用简单工厂创建对象
huawei = PhoneFactory.create_phones("huawei")
print(huawei.make())  # 输出: Manufacture of Huawei mobile phones!

iphone = PhoneFactory.create_phones("iphone")
print(iphone.make())  # 输出: Manufacture of iPhone mobile phones!

说明: PhoneFactory 类提供了一个静态方法 create_phones,根据传入的 phone_type 参数,返回相应的 Huawei 或 Iphone 实例。客户端只需要调用工厂方法,无需关心具体的 Huawei 或 Iphone 类。

实际生产用例:日志处理系统中的简单工厂模式

在实际生产中,简单工厂模式可以用于创建不同类型的日志处理器(如控制台日志、文件日志、数据库日志等),以便根据配置或运行时条件灵活地选择日志记录方式。

示例代码:

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
class Logger:
    def log(self, message):
        pass


class ConsoleLogger(Logger):
    def log(self, message):
        print(f"ConsoleLogger: {message}")


class FileLogger(Logger):
    def log(self, message):
        with open("logfile.txt", "a") as file:
            file.write(f"FileLogger: {message}\n")


class DatabaseLogger(Logger):
    def log(self, message):
        # 模拟数据库写入
        print(f"DatabaseLogger: {message} (Written to database)")


class LoggerFactory:
    @staticmethod
    def create_logger(logger_type):
        if logger_type == "console":
            return ConsoleLogger()
        elif logger_type == "file":
            return FileLogger()
        elif logger_type == "database":
            return DatabaseLogger()
        else:
            raise ValueError("Unknown logger type")


# 使用简单工厂根据配置创建日志处理器
logger_type = "file"  # 可以从配置文件或运行时动态确定
logger = LoggerFactory.create_logger(logger_type)
logger.log("This is a test log message.")

2.2 工厂方法模式

定义一个用于创建对象的工厂接口,但让子类决定实例化哪个类,工厂方法模式使得类的实例化推迟到子类。

python_features2.2.png

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
from abc import ABC, abstractmethod


class Phone(ABC):
    @abstractmethod
    def make(self):
        pass


class Huawei(Phone):
    def make(self):
        return "Manufacture of Huawei mobile phones!"


class Iphone(Phone):
    def make(self):
        return "Manufacture of iPhone mobile phones!"


class PhoneFactory(ABC):
    @abstractmethod
    def create_phones(self):
        pass


class HuaweiFactory(PhoneFactory):
    def create_phones(self):
        return Huawei()


class IphoneFactory(PhoneFactory):
    def create_phones(self):
        return Iphone()


factory = HuaweiFactory()
phone = factory.create_phones()
print(phone.make())  # 输出: Manufacture of Huawei mobile phones!

说明:PhoneFactory 工厂接口,由子类决定实例化对象,如:HuaweiFactory 实例化 Huawei、IphoneFactory 实例化 Iphone。

实际生产用例:数据库连接器
在一个大型系统中,需要根据不同的数据库类型(如 MySQL、PostgreSQL、SQLite)创建对应的数据库连接器。工厂方法模式可以使代码灵活地支持多种数据库,而无需在主代码中硬编码数据库类型。

示例代码:

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
from abc import ABC, abstractmethod


class DatabaseConnector(ABC):
    @abstractmethod
    def connect(self):
        pass


class MySQLConnector(DatabaseConnector):
    def connect(self):
        print("Connecting to MySQL")


class PostgreSQLConnector(DatabaseConnector):
    def connect(self):
        print("Connecting to PostgreSQL")


class SQLiteConnector(DatabaseConnector):
    def connect(self):
        print("Connecting to SQLite")


class DatabaseFactory(ABC):
    @abstractmethod
    def create_connector(self):
        pass


class MySQLFactory(DatabaseFactory):
    def create_connector(self):
        return MySQLConnector()


class PostgreSQLFactory(DatabaseFactory):
    def create_connector(self):
        return PostgreSQLConnector()


class SQLiteFactory(DatabaseFactory):
    def create_connector(self):
        return SQLiteConnector()


# 使用工厂方法模式根据配置创建数据库连接器
config = "PostgreSQL"  # 可以从配置文件或运行时动态确定
if config == "MySQL":
    factory = MySQLFactory()
elif config == "PostgreSQL":
    factory = PostgreSQLFactory()
elif config == "SQLite":
    factory = SQLiteFactory()

connector = factory.create_connector()
connector.connect()  # 输出:Connecting to PostgreSQL
  • 代码解耦:工厂方法模式将对象的创建与使用分离,使得代码更为灵活,增加了系统的可扩展性。添加新的类型时,只需添加新的工厂类,不需要修改现有代码;
  • 符合开闭原则:代码对扩展开放,对修改封闭。新对象的创建逻辑可以通过扩展新的工厂类来实现,而不影响现有代码的稳定性;
  • 提高代码可读性:使用工厂方法模式后,创建对象的代码更简洁清晰,更容易维护;

2.3 抽象工厂模式

提供一个创建一系列相关或相互依赖对象的接口,而无需指定具体类,由其子类工厂实现创建相关对象。

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
from abc import ABC, abstractmethod


# 抽象产品
class Button(ABC):
    # 按钮
    @abstractmethod
    def click(self):
        pass


class Checkbox(ABC):
    # 复选框
    @abstractmethod
    def toggle(self):
        pass


# 具体产品
class WindowsButton(Button):
    def click(self):
        return "Windows Button Clicked"


class MacOSButton(Button):
    def click(self):
        return "MacOS Button Clicked"


class WindowsCheckbox(Checkbox):
    def toggle(self):
        return "Windows Checkbox Toggled"


class MacOSCheckbox(Checkbox):
    def toggle(self):
        return "MacOS Checkbox Toggled"


# 抽象工厂
class GUIFactory(ABC):
    @abstractmethod
    def create_button(self):
        pass

    @abstractmethod
    def create_checkbox(self):
        pass


# 具体工厂
class WindowsFactory(GUIFactory):
    def create_button(self):
        return WindowsButton()

    def create_checkbox(self):
        return WindowsCheckbox()


class MacOSFactory(GUIFactory):
    def create_button(self):
        return MacOSButton()

    def create_checkbox(self):
        return MacOSCheckbox()


# 客户端代码
def create_ui(factory: GUIFactory):
    button = factory.create_button()
    checkbox = factory.create_checkbox()
    print(button.click())
    print(checkbox.toggle())


# 实际使用
factory = WindowsFactory()
create_ui(factory)
# 输出:
# Windows Button Clicked
# Windows Checkbox Toggled

factory = MacOSFactory()
create_ui(factory)
# 输出:
# MacOS Button Clicked
# MacOS Checkbox Toggled

说明:GUIFactory 抽象工厂接口定义了创建一组相关或相互依赖对象的接口,不指定具体类,由具体子类工厂实现创建相关的对象(WindowsFactory、MacOSFactory)。

使用场景

  • 跨平台应用:如,可以根据不同平台创建不同的界面组件、封装不同共有云 API 接口(华为、阿里等);
  • 产品族的创建:如果需要创建一组相关的对象,而不是单一对象,可以使用抽象工厂;

工厂方法专注于创建单个产品,抽象工厂则是用于创建一系列相关产品。两者都遵循依赖倒置原则和开放/封闭原则,但抽象工厂模式更为复杂,适用于需要生成一组关联对象的场景。

2.4 单例模式与工厂模式结合

确保工厂类只存在一个实例:

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
class SingletonFactory:
    _instance = None

    def __new__(cls, *args, **kwargs):
        if not cls._instance:
            cls._instance = super().__new__(cls, *args, **kwargs)
        return cls._instance


class DogFactory(SingletonFactory):
    def create_animal(self):
        return Dog()


factory1 = DogFactory()
factory2 = DogFactory()
print(factory1 is factory2)  # 输出: True

3 观察者模式(Observer Pattern)

观察者模式是一种行为设计模式,它定义了一种一对多的依赖关系。当一个对象的状态发生变化时,所有依赖于它的对象都会收到通知并自动更新。

组成部分:

  • Subject(主题):维护一组观察者,并提供方法来添加、删除和通知观察者;
  • Observer(观察者):定义一个接口,用于接收主题更新的通知;
  • ConcreteSubject(具体主题):实现主题接口,并在状态变化时通知所有观察者;
  • ConcreteObserver(具体观察者):实现观察者接口,并在接收到通知后更新自身状态;
Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
from abc import ABC, abstractmethod


class Subject:
    # Subject 类:代表观察者模式中的“主题”或“被观察对象”。它维护一个 _observers 列表,用于存储所有已注册的观察者
    # 在设计上,也可以考虑抽象一层,例如: 抽象主题 Subject(): 定义 __init__(), attach(), detach(), notify() ,具体主题ConcreteSubject(Subject),示例比较简单没有抽象
    def __init__(self):
        self._observers = []

    def attach(self, observer):
        # attach(self, observer):将一个观察者添加到 _observers 列表中
        self._observers.append(observer)

    def detach(self, observer):
        # detach(self, observer):从 _observers 列表中移除指定的观察者
        self._observers.remove(observer)

    def notify(self, message):
        # notify(self, message):遍历 _observers 列表,调用每个观察者的 update 方法,将 message 传递给他们
        for observer in self._observers:
            observer.update(message)


class Observer(ABC):
    # Observer 类:定义了观察者的接口。它包含一个 update(self, message) 方法,所有具体的观察者都必须实现这个方法以处理来自主题的通知
    @abstractmethod
    def update(self, message):
        pass


class ConcreteObserver(Observer):
    # ConcreteObserver 类:继承自 Observer,是一个具体的观察者实现
    def __init__(self, name):
        # __init__(self, name):初始化观察者,并给它分配一个 name 属性
        self.name = name

    def update(self, message):
        # update(self, message):实现了 Observer 的 update 方法,当收到来自主题的通知时,打印出收到的消息
        print(f"{self.name} received message: {message}")


# 使用示例
subject = Subject()  # 创建一个主题

# 创建2个观察者
observer1 = ConcreteObserver("Observer 1")
observer2 = ConcreteObserver("Observer 2")

# 将观察者注册到对应主题
subject.attach(observer1)
subject.attach(observer2)

# 通知观察者,传递消息
subject.notify("New event occurred!")

# 输出:
# Observer 1 received message: New event occurred!
# Observer 2 received message: New event occurred!

【扩展知识点】:

在 Python 中,当你定义一个抽象接口(即使用 abc.ABC 和 abc.abstractmethod 时),抽象方法通常需要定义 self 参数,特别是在实例方法中。self 是指向实例本身的引用,允许访问实例属性和方法。抽象方法和普通实例方法一样,需要传递 self 作为第一个参数。

Python
1
2
3
4
5
6
from abc import ABC, abstractmethod

class MyAbstractClass(ABC):
    @abstractmethod
    def my_method(self, param):
        pass

本例中,my_method 是一个抽象方法,self 需要作为第一个参数。

4 责任链模式(Chain of Responsibility Pattern)

4.1 介绍

责任链模式(Chain of Responsibility Pattern)是一种行为型设计模式,它允许多个对象有机会处理请求,避免了请求发送者与接收者之间的紧耦合;通过将多个处理对象串联成一条链,请求会沿着这条链传递,直到有对象处理它或者到达链末尾。可以将责任链模式理解为一个“接力赛跑”,每个对象都有机会处理请求,如果不能处理,就将请求传递给下一个对象。

模式结构:
责任链模式主要包含以下角色:

  • Handler(抽象处理者):抽象基类,定义处理请求的方法,通常提供将请求传递给下一个处理者的接口;
  • ConcreteHandler(具体处理者):继承 Handler,实现抽象处理者的处理逻辑,如果不能处理,则将请求传递给下一个处理者;
  • Client(客户端):创建具体处理者对象并设置它们的职责链;

原理:
责任链模式的核心在于链式调用,每个处理者持有对下一个处理者的引用,当一个处理者无法处理请求时,将请求转发给下一个处理者。这种方式使得请求的发送者无需知道请求将由哪个对象处理,增加了系统的灵活性和可扩展性。

优势:

  • 降低耦合:请求的发送者与处理者之间不需要显式地知道对方,增加了系统的灵活性;
  • 增强灵活性:可以动态地添加或移除处理者,轻松改变处理顺序;
  • 遵循单一职责原则:每个处理者只关注自身的处理逻辑,职责清晰;

缺点:

  • 可能无法保证请求被处理:如果链中没有处理者能够处理请求,可能导致请求未被处理;
  • 调试困难:链中的处理者可能较多,跟踪请求的处理路径可能较为复杂;
  • 性能问题:如果责任链过长,可能导致处理请求的时间增加;

常见应用场景:

  • 订单审批流程:在企业中,订单的审批通常需要经过多个层级,例如经理、主管、总监等,根据订单金额的不同,订单需要不同层级的审批,责任链模式可以动态地构建审批流程;
  • 表单验证:在用户提交表单时,可能需要进行多个验证步骤,如字段格式验证、逻辑验证、权限验证等。每个验证步骤可以作为责任链中的一个处理者,逐步验证输入数据的合法性;
  • 日志处理系统:日志系统可能需要根据日志级别(如 DEBUG、INFO、WARN、ERROR)将日志信息发送到不同的输出目标(如控制台、文件、远程服务器)。责任链模式可以动态地配置不同级别日志的处理逻辑;
  • 数据处理流水线:在数据处理过程中,可能需要经过多个处理步骤,如数据清洗、转换、验证、存储等。每个处理步骤可以作为责任链中的一个处理者,依次处理数据;

4.2 代码示例

  • 示例1:订单审批流程
    需求描述: 订单金额不同,需要不同级别的审批人员进行审批:
  • 金额 < 1000:由经理审批
  • 1000 ≤ 金额 < 5000:由主管审批
  • 金额 ≥ 5000:由总监审批
Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
from __future__ import annotations

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional, Any


@dataclass
class Order:
    amount: float  # 金额
    description: str  # 描述

# 处理程序接口
class Handler(ABC):
    @abstractmethod
    def set_next(self, handler: Handler) -> Handler:
        # 声明一个用于构建处理程序链的方法
        pass

    @abstractmethod
    def handle(self, data) -> Optional[str]:
        # 声明一个用于执行处理方法
        pass

# 基础类,定义默认处理程序行为
class AbstractHandler(Handler):
    _next_handler: Handler = None

    def set_next(self, handler: Handler) -> Handler:
        self._next_handler = handler
        return handler

    @abstractmethod
    def handle(self, data: Any) -> str:
        if self._next_handler:
            return self._next_handler.handle(data)
        return ""


# 具体处理者:经理
class ManagerHandler(AbstractHandler):
    def handle(self, data: Order) -> str:
        if data.amount < 1000:
            return f"Manager approved order: {data.description} for ${data.amount}"
        else:
            return super().handle(data)


# 具体处理者:主管
class SupervisorHandler(AbstractHandler):
    def handle(self, data: Order) -> str:
        if 1000 <= data.amount < 5000:
            return f"Supervisor approved order: {data.description} for ${data.amount}"
        else:
            return super().handle(data)


# 具体处理者:总监
class DirectorHandler(AbstractHandler):
    def handle(self, data: Order) -> str:
        if data.amount >= 5000:
            return f"Director approved order: {data.description} for ${data.amount}"
        else:
            return super().handle(data)


if __name__ == '__main__':
    # 创建订单
    orders = [
        Order(500, "Office Supplies"),
        Order(2000, "New Computers"),
        Order(7000, "Office Renovation")
    ]

    # 处理订单
    manager = ManagerHandler()
    supervisor = SupervisorHandler()
    director = DirectorHandler()
    manager.set_next(supervisor).set_next(director)
    for order in orders:
        print(manager.handle(order))

# 输出:
# Manager approved order: Office Supplies for $500
# Supervisor approved order: New Computers for $2000
# Director approved order: Office Renovation for $7000

代码解释: * Order 类:使用 @dataclass 装饰器定义订单对象,包含 amount 和 description 两个属性; * AbstractHandler 基础类:实现了默认责任链模式行为,包含:默认处理行为及 _next_handler 对下一个处理者(审批者)的引用; * 具体处理者(ManagerHandler, SupervisorHandler, DirectorHandler):实现 handle 方法,根据订单金额决定是否处理或转发给下一个处理者; * 客户端代码: * 构建责任链:manager -> supervisor -> director; * 创建多个订单并通过责任链进行审批;

  • 示例2:表单验证
    需求描述:
    用户提交表单时,需要经过多个验证步骤:
  • 非空验证;
  • 格式验证(如邮箱格式);
  • 逻辑验证(如密码匹配);
Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
from abc import ABC, abstractmethod

# 请求对象
@dataclass
class FormData:
    username: str
    email: str
    password: str
    confirm_password: str

# 抽象处理者
class Validator(ABC):
    def __init__(self, successor=None):
        self._successor = successor

    @abstractmethod
    def validate(self, data: FormData):
        pass

# 具体处理者:非空验证
class NotEmptyValidator(Validator):
    def validate(self, data: FormData):
        if not all([data.username, data.email, data.password, data.confirm_password]):
            raise ValueError("All fields must be filled out.")
        if self._successor:
            self._successor.validate(data)

# 具体处理者:格式验证
import re

class EmailFormatValidator(Validator):
    def validate(self, data: FormData):
        email_regex = r"[^@]+@[^@]+\.[^@]+"
        if not re.match(email_regex, data.email):
            raise ValueError("Invalid email format.")
        if self._successor:
            self._successor.validate(data)

# 具体处理者:逻辑验证
class PasswordMatchValidator(Validator):
    def validate(self, data: FormData):
        if data.password != data.confirm_password:
            raise ValueError("Passwords do not match.")
        if self._successor:
            self._successor.validate(data)

# 客户端代码
if __name__ == "__main__":
    # 构建验证链
    # password_validator = PasswordMatchValidator()
    # email_validator = EmailFormatValidator(password_validator)
    # not_empty_validator = NotEmptyValidator(email_validator)
    not_empty_validator = NotEmptyValidator(EmailFormatValidator(PasswordMatchValidator()))
    # 创建表单数据
    form_data = [
        FormData("john_doe", "john@example.com", "password123", "password123"),
        FormData("", "jane@example.com", "password123", "password123"),
        FormData("jane_doe", "janeexample.com", "password123", "password123"),
        FormData("jane_doe", "jane@example.com", "password123", "password321")
    ]

    # 验证表单数据
    for data in form_data:
        try:
            not_empty_validator.validate(data)
            print(f"Form data for {data.username} is valid.")
        except ValueError as e:
            print(f"Validation error for {data.username}: {e}")
# 输出:
# Form data for john_doe is valid.
# Validation error for : All fields must be filled out.
# Validation error for jane_doe: Invalid email format.
# Validation error for jane_doe: Passwords do not match.

代码解释: * FormData 类:定义表单数据结构,包含 username, email, password, confirm_password 四个字段; * Validator 抽象类:定义了验证器的接口,包含 validate 方法和对下一个验证器的引用 _successor; * 具体验证器(NotEmptyValidator, EmailFormatValidator, PasswordMatchValidator): * NotEmptyValidator:检查所有字段是否非空; * EmailFormatValidator:检查邮箱格式是否正确; * PasswordMatchValidator:检查密码和确认密码是否匹配; * 客户端代码: * 构建验证链:NotEmptyValidator -> EmailFormatValidator -> PasswordMatchValidator; * 创建多个表单数据并进行验证;

  • 示例3:基于责任链模式的数据格式校验示例代码,假设我们需要校验数据中的多个字段,如字符串的长度、数值的范围等;
Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional


@dataclass
class Data:
    # Data 数据类:封装了待校验的数据属性,如字符串、分数和日期
    string_value: str
    score: int
    date: str


class _CheckInterface(ABC):
    # _CheckInterface 接口:定义了责任链的基础接口,包括设置下一个检查器(set_next)和执行检查(check)。
    @abstractmethod
    def set_next(self, checker: '_CheckInterface') -> '_CheckInterface':
        pass

    @abstractmethod
    def check(self, data: Data, **kwargs) -> Optional[str]:
        pass


class DataChecker(_CheckInterface):
    # DataChecker 抽象类:实现了责任链的基本逻辑,
    # 通过 set_next 方法构建链条,
    # check 方法逐级调用链条上的检查器,
    # raise_error 方法则抛出校验失败的异常
    _next_checker: Optional[_CheckInterface] = None

    def set_next(self, checker: _CheckInterface) -> _CheckInterface:
        self._next_checker = checker
        return checker

    def check(self, data: Data, **kwargs) -> Optional[str]:
        if self._next_checker:
            return self._next_checker.check(data, **kwargs)
        return None

    def raise_error(self, message: str):
        raise ValueError(f'校验失败: {message}')

    @classmethod
    def check_list(cls) -> 'DataChecker':

        # return CheckStringLength().set_next(CheckScoreRange()).set_next(CheckDateFormat())
        # 或者
        _checker = CheckStringLength()
        tmp = _checker
        for clazz in [CheckScoreRange, CheckDateFormat]:
            tmp = tmp.set_next(clazz())
        return _checker


class CheckStringLength(DataChecker):
    # 具体的检查器:校验字符串长度
    MIN = 5
    MAX = 20

    def check(self, data: Data, **kwargs) -> Optional[str]:
        if not (self.MIN <= len(data.string_value) <= self.MAX):
            self.raise_error(f'字符串长度需要在 {self.MIN}-{self.MAX} 之间')
        return super().check(data, **kwargs)


class CheckScoreRange(DataChecker):
    # 具体的检查器:分数范围
    MIN = 0
    MAX = 100

    def check(self, data: Data, **kwargs) -> Optional[str]:
        if not (self.MIN <= data.score <= self.MAX):
            self.raise_error(f'分值需要在 {self.MIN}-{self.MAX} 之间')
        return super().check(data, **kwargs)


class CheckDateFormat(DataChecker):
    # 具体的检查器:日期格式
    def check(self, data: Data, **kwargs) -> Optional[str]:
        try:
            year, month, day = map(int, data.date.split('-'))
            if len(data.date) != 10:
                raise ValueError
        except ValueError:
            self.raise_error('日期格式不正确,应为 YYYY-MM-DD')
        return super().check(data, **kwargs)


# 示例数据
data = Data(string_value='HelloWorld', score=85, date='2024-08-30')

try:
    checker = DataChecker.check_list()
    checker.check(data)
    print("数据校验通过")
except ValueError as e:
    print(str(e))

扩展与重用:

扩展其他校验时,可以通过继承 DataChecker 类并实现 check 方法来定义新的检查器。例如,如果需要增加电子邮件格式校验,只需创建一个新的检查类并加入到责任链中:

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
class CheckEmailFormat(DataChecker):
    def check(self, data: Data, **kwargs) -> Optional[str]:
        if '@' not in data.string_value:
            self.raise_error('无效的电子邮件格式')
        return super().check(data, **kwargs)


# 在责任链中增加新的检查器
def check_list_with_email() -> DataChecker:
    return CheckStringLength().set_next(CheckEmailFormat()).set_next(CheckScoreRange()).set_next(CheckDateFormat())

通过责任链模式,新的校验逻辑可以轻松集成到现有系统中,且不影响已有的校验器,实现了代码的高度复用和扩展性。

【扩展知识点】:from __future__ import annotations
from __future__ import annotations 是用于延迟类型注解的计算。它让 Python 在运行时才去解析类型注解,而不是在定义时立即解析。这样做可以避免在代码中由于前向引用或循环导入导致的问题。 从 Python 3.10 开始,类型注解默认是延迟计算的,因此在 Python 3.10 及以上版本中不再需要手动引入 from __future__ import annotations。但是,在 Python 3.9 及以下版本中,仍然需要使用该语句来启用这一功能。

  • 在 Python 3.10 及更高版本中,默认启用了 PEP 563,这使得类型注解被推迟到运行时进行求值。这一特性让你不必再手动导入 from __future__ import annotations。但如果你希望在更早的 Python 版本中使用此特性,仍然需要手动导入;
  • 在 Python 3.11 中,PEP 649 被引入,并将替换 PEP 563 的实现,因此无需再手动使用 from __future__ import annotations

5 策略模式(Strategy Pattern)

策略模式(Strategy Pattern)是一种行为设计模式,它定义了一系列算法,并将每个算法封装到独立的类中,使得它们可以互相替换,在 Python 代码中很常见,经常在各种框架中使用,能在不扩展类的情况下向用户提供改变其行为的方式。

关键点:

  • Context:使用策略对象的类,维护指向具体策略的引用,且仅通过策略接口与该对象进行交流;
  • Strategy:策略接口或抽象类,定义了算法的共同行为;
  • Concrete Strategy:具体策略类,实现了不同的算法;

python_features5.1.png

示例:对 Lits 数据提供降序/升序排列算法

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
from __future__ import annotations

from abc import ABC, abstractmethod
from typing import List


class Strategy(ABC):
    """
    策略接口声明了某个算法各个不同版本间所共有的操作。Context 会使用该接口来调用有具体策略定义的算法
    """

    @abstractmethod
    def do_algorithm(self, data: List):
        pass


class Context:
    """
    持有一个 Strategy 对象的引用。Context 类通过 strategy 属性调用策略的算法
    """

    def __init__(self, strategy: Strategy) -> None:
        self._strategy = strategy

    @property
    def strategy(self) -> Strategy:
        return self._strategy

    @strategy.setter
    def strategy(self, strategy: Strategy) -> None:
        self._strategy = strategy

    def do_some_business_logic(self, data: List) -> None:
        # 展示如何使用策略对象来执行算法逻辑。在此例中,它使用策略对数据进行排序
        print("Context: Sorting data using the strategy (not sure how it'll do it)")
        result = self._strategy.do_algorithm(data)
        print(",".join(result))


class ConcreteStrategyA(Strategy):
    """
     实现按升序排序的算法
    """

    def do_algorithm(self, data: List) -> List:
        return sorted(data)


class ConcreteStrategyB(Strategy):
    """
     实现按降序排序的算法
    """

    def do_algorithm(self, data: List) -> List:
        return reversed(sorted(data))


if __name__ == "__main__":
    data = ['a', 'b', 'c', 'd', 'e', 'f']
    context = Context(ConcreteStrategyA())
    print("Client: Strategy is set to normal sorting.")
    context.do_some_business_logic(data)
    print()
    # 通过改变 Context 的 strategy,可以动态地切换使用不同的算法
    print("Client: Strategy is set to reverse sorting.")
    context.strategy = ConcreteStrategyB()
    context.do_some_business_logic(data)

# 输出:
# Client: Strategy is set to normal sorting.
# Context: Sorting data using the strategy (not sure how it'll do it)
# a,b,c,d,e,f
# 
# Client: Strategy is set to reverse sorting.
# Context: Sorting data using the strategy (not sure how it'll do it)
# f,e,d,c,b,a

优点:

  • 灵活性:可以在运行时动态改变策略;
  • 可扩展性:增加新策略不影响现有系统;

应用场景:
策略模式适合场景:当系统有多种算法或行为,并且希望在运行时灵活选择其中一种时。

  • 支付方式选择:根据用户选择的支付方式(如信用卡、PayPal),应用不同的支付策略;
  • 路径规划:地图应用中可以根据不同的策略(最短路径、避开高速、避开收费)进行路径规划;

6 命令模式(Command Pattern)

命令模式(Command Pattern)是一种行为设计模式,它可将请求转换为一个包含与请求相关的所有信息的独立对象。该转换让你能根据不同的请求将方法参数化、延迟请求执行或将其放入队列中,且能实现可撤销操作。

命令模式结构:

  • 命令对象(Command):封装了一个具体的操作和它的参数;
  • 调用者(Invoker):持有命令对象并在某个时间点调用命令;
  • 接收者(Receiver):实际执行命令操作的对象;

python_features6.1.png

示例:

Python
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
from __future__ import annotations

from abc import ABC, abstractmethod


class Command(ABC):
    """
    Command 接口声明了一个执行命令的方法
    """

    @abstractmethod
    def execute(self) -> None:
        pass


class SimpleCommand(Command):
    """
    具体命令类:实现了简单打印任务
    """

    def __init__(self, payload: str) -> None:
        self._payload = payload

    def execute(self) -> None:
        print(f"SimpleCommand: See, I can do simple things like printing"
              f"({self._payload})")


class ComplexCommand(Command):
    """
    具体命令类:处理更复杂的操作,接受 Receiver 对象和一些参数,在 execute 中将任务委托给接收者。
    """

    def __init__(self, receiver: Receiver, a: str, b: str) -> None:
        """
        复杂命令可以通过构造函数接受一个或多个接收对象(receivers)以及任何上下文数据
        """
        self._receiver = receiver
        self._a = a
        self._b = b

    def execute(self) -> None:
        """
        命令执行委托给接收者(receivers)方法
        """
        print("ComplexCommand: Complex stuff should be done by a receiver object", end="")
        self._receiver.do_something(self._a)
        self._receiver.do_something_else(self._b)


class Receiver:
    """
    包含了执行实际任务的业务逻辑
    """

    def do_something(self, a: str) -> None:
        print(f"\nReceiver: Working on ({a}.)", end="")

    def do_something_else(self, b: str) -> None:
        print(f"\nReceiver: Also working on ({b}.)", end="")


class Invoker:
    """
    保存命令对象并在合适的时机调用它们。它通过 set_on_start 和 set_on_finish 方法设置在任务开始前和结束后的命令
    """
    _on_start = None
    _on_finish = None

    def set_on_start(self, command: Command):
        self._on_start = command

    def set_on_finish(self, command: Command):
        self._on_finish = command

    def do_something_important(self) -> None:
        """
        Invoker 不依赖于具体的命令或接收器类。Invoker 通过执行命令间接将请求传递给接收器
        """
        print("Invoker: Does anybody want something done before I begin?")
        if isinstance(self._on_start, Command):
            self._on_start.execute()

        print("Invoker: ...doing something really important...")

        print("Invoker: Does anybody want something done after I finish?")
        if isinstance(self._on_finish, Command):
            self._on_finish.execute()


if __name__ == "__main__":
    """
    通过设置不同的命令和接收者,展示了命令模式的灵活性
    """

    invoker = Invoker()
    invoker.set_on_start(SimpleCommand("Say Hi!"))
    receiver = Receiver()
    invoker.set_on_finish(ComplexCommand(
        receiver, "Send email", "Save report"))

    invoker.do_something_important()

# 输出:
# Invoker: Does anybody want something done before I begin?
# SimpleCommand: See, I can do simple things like printing(Say Hi!)
# Invoker: ...doing something really important...
# Invoker: Does anybody want something done after I finish?
# ComplexCommand: Complex stuff should be done by a receiver object
# Receiver: Working on (Send email.)
# Receiver: Also working on (Save report.)

扩展性:

  • 新的命令可以通过继承 Command 类轻松扩展,无需修改现有代码;
  • Invoker 类可以调用任意命令对象,使得命令链和任务序列化非常灵活;

使用场景:

  • 任务撤销、重做功能;
  • 事务脚本的执行(如数据库操作);
  • 宏命令(多个命令的组合);

7 适配器模式(Adapter Pattern)

适配器模式(Adapter Pattern)是一种结构型设计模式,它允许将一个类的接口转换为客户希望的另一个接口,使得原本不兼容的类可以协同工作。 适配器模式在 Python 代码中很常见。 基于一些遗留代码的系统常常会使用该模式。 在这种情况下, 适配器让遗留代码与当前类得以相互合作。

示例: 假设你有一个旧的类 Adaptee,它的接口不符合新的系统 Target 所要求的接口。通过使用适配器模式,可以在不改变旧系统代码的情况下使其兼容新系统。

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
class Adaptee:
    """
    表示旧系统,有自己的接口 specific_request
    """

    def specific_request(self) -> str:
        return "OldSystem's specific request"


class Target:
    """
    新系统所期望的接口
    """

    def request(self) -> str:
        return "NewSystem's request"


class Adapter(Target, Adaptee):
    """
    适配器类(Adapter),通过多重继承使得 Adaptee(旧系统) 的接口和 Target(新系统) 的接口兼容
    """

    def request(self) -> str:
        return f"Adapter: (TRANSLATED) {self.specific_request()}"


# 新系统中使用统一的接口
def client_code(target: Target) -> None:
    """
    客户端代码,直接与 TargetInterface 交互,无需关心其具体实现
    """
    print(target.request(), end="")


if __name__ == "__main__":
    # Target(新系统) 调用
    target = Target()
    client_code(target)
    print("\n")

    # 使用适配器来适应旧系统
    adapter = Adapter()
    client_code(adapter)

# 输出:
# NewSystem's request
# Adapter: (TRANSLATED) OldSystem's specific request

8 模板方法模式(Template Method Pattern)

模板方法模式(Template Method Pattern)是一种行为设计模式,它定义了一个算法的骨架,并允许子类在不改变算法结构的情况下重定义算法的某些步骤。此模式将代码的复用性和灵活性很好地结合在一起。

示例场景:
假设你要实现一组操作步骤,每个步骤的细节可能会有所不同,但总体的操作流程是相同的。模板方法模式可以将这些步骤的框架定义在一个基类中,而具体的实现则交由子类完成。

  • 当多个类有相似的逻辑结构时,可以使用模板方法模式将相同的逻辑部分提取到基类中,而不同的细节部分则由子类实现;
  • 避免代码重复,提升代码的可维护性和扩展性;

示例代码:

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
from abc import ABC, abstractmethod


class DataProcessor(ABC):
    """
    抽象类,定义了处理数据 process 方法,其中包含读取、处理和保存数据的步骤,具体由子类实现,但模板方法 process 本身应保持不变,定义操作调用组成及顺序。
    """

    def process(self):
        # 模板方法 process:在基类中定义,不可更改,但可以通过继承来实现具体的细节
        self.to_start()
        self.read_data()
        self.process_data()
        self.save_data()
        self.to_finish()

    def to_start(self) -> None:
        print("Start processing data")

    def to_finish(self) -> None:
        print("End of data processing")

    @abstractmethod
    def read_data(self) -> None:
        pass

    @abstractmethod
    def process_data(self) -> None:
        pass

    @abstractmethod
    def save_data(self) -> None:
        pass


class CSVProcessor(DataProcessor):
    """
    具体子类: CSV 数据的读取、处理和保存步骤
    """

    def read_data(self):
        print("Reading CSV data")

    def process_data(self):
        print("Processing CSV data")

    def save_data(self):
        print("Saving CSV data")


class JSONProcessor(DataProcessor):
    """
    具体子类: JSON 数据的读取、处理和保存步骤
    """

    def read_data(self):
        print("Reading JSON data")

    def process_data(self):
        print("Processing JSON data")

    def save_data(self):
        print("Saving JSON data")


def client_code(data_processor: DataProcessor) -> None:
    data_processor.process()


if __name__ == "__main__":
    client_code(CSVProcessor())
    print("\n")
    client_code(JSONProcessor())

# 输出:
# Start processing data
# Reading CSV data
# Processing CSV data
# Saving CSV data
# End of data processing
#
#
# Start processing data
# Reading JSON data
# Processing JSON data
# Saving JSON data
# End of data processing

9 组合模式(Composite Pattern)

组合模式(Composite Pattern)是一种结构型设计模式,它允许你将对象组合成树形结构来表示“部分-整体”的层次结构。组合模式使得用户对单个对象和组合对象的使用具有一致性。

示例场景: 假设你在开发一个文件系统,其中有文件和文件夹。文件夹可以包含文件或其他文件夹。你希望用户能够对文件和文件夹进行统一的操作,如获取大小、添加或删除内容等。

示例代码:

Python
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
from __future__ import annotations

from abc import ABC, abstractmethod
from typing import List


# 组件接口
class Component(ABC):
    """
    声明了组合中简单对象和复杂对象的通用操作
    """

    # 可选(parent):基础组件可以声明一个接口,用于设置和访问树结构中的父级,并提供一些默认实现
    @property
    def parent(self) -> Component:
        return self._parent

    @parent.setter
    def parent(self, component: Component):
        self._parent = component

    def add(self, component: Component) -> None:
        pass

    def remove(self, component: Component) -> None:
        pass

    def is_composite(self) -> bool:
        return False

    @abstractmethod
    def operation(self) -> str:
        # 是所有组件必须实现的方法,定义了组件的具体操作行为
        pass


# 叶子节点
class Leaf(Component):
    """
    Leaf 类表示组合的最终对象(树结构中的叶子节点),叶子节点不能包含其他组件。
    通常,Leaf 对象会执行实际工作,而 Composite 对象只会委托给其子组件。
    """

    def operation(self) -> str:
        # 具体实现 operation() 方法,表示叶子节点的操作
        return "Leaf"


# 组合节点
class Composite(Component):
    """
    组合节点可以包含其他 Component 对象(既可以是叶子节点,也可以是其他组合节点)
    """

    def __init__(self) -> None:
        self._children: List[Component] = []

    # 实现了 add() 和 remove() 方法,用于管理子节点
    def add(self, component: Component) -> None:
        self._children.append(component)
        component.parent = self

    def remove(self, component: Component) -> None:
        self._children.remove(component)
        component.parent = None

    def is_composite(self) -> bool:
        return True

    def operation(self) -> str:
        # operation() 方法递归调用子节点的 operation() 方法,并返回组合的结果
        results = []
        for child in self._children:
            results.append(child.operation())
        return f"Branch({'+'.join(results)})"


def client_code(component: Component) -> None:
    # 演示如何在不知道组件类型的情况下执行操作。这展示了组合模式的强大之处,可以对叶子节点和组合节点进行统一处理
    print(f"RESULT: {component.operation()}", end="")


def client_code2(component1: Component, component2: Component) -> None:
    # 进一步展示了组合模式的灵活性,允许在运行时动态地将子组件添加到组合组件中
    if component1.is_composite():
        component1.add(component2)

    print(f"RESULT: {component1.operation()}", end="")


if __name__ == "__main__":
    # 单个叶子节点: 打印 Leaf,表示这是一个单独的叶子节点
    simple = Leaf()
    print("Client: I've got a simple component:")
    client_code(simple)
    print("\n")
    # 组合树结构: 打印 Branch(Branch(Leaf + Leaf) + Branch(Leaf)),表示一个组合结构,包含多个叶子节点
    tree = Composite()

    branch1 = Composite()
    branch1.add(Leaf())
    branch1.add(Leaf())

    branch2 = Composite()
    branch2.add(Leaf())

    tree.add(branch1)
    tree.add(branch2)

    print("Client: Now I've got a composite tree:")
    client_code(tree)
    print("\n")
    # 动态添加叶子节点: 打印 Branch(Branch(Leaf + Leaf) + Branch(Leaf) + Leaf),展示如何在树结构中动态添加节点,并统一处理
    print("Client: I don't need to check the components classes even when managing the tree:")
    client_code2(tree, simple)

# 输出:
# Client: I've got a simple component:
# RESULT: Leaf
#
# Client: Now I've got a composite tree:
# RESULT: Branch(Branch(Leaf+Leaf)+Branch(Leaf))
#
# Client: I don't need to check the components classes even when managing the tree:
# RESULT: Branch(Branch(Leaf+Leaf)+Branch(Leaf)+Leaf)

10 外观模式(Facade Pattern)

外观模式(Facade Pattern)是一种结构型设计模式,旨在为复杂的子系统提供一个简单的接口,使客户端能够更容易地与子系统交互。外观模式通过封装复杂系统的内部实现,为客户端提供一个统一的接口,从而减少客户端与系统之间的耦合。

外观模式的主要作用:

  • 简化接口:隐藏子系统的复杂性,提供简单易用的接口;
  • 减少依赖:通过引入外观类,客户端不需要直接依赖于复杂系统的各个部分,从而减少了系统的耦合度;

代码示例:

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
from __future__ import annotations


class Subsystem1:
    # 子系统 1:表示复杂系统中的子系统
    def operation1(self) -> str:
        return "Subsystem1: Ready!"

    # ... ...

    def operation_n(self) -> str:
        return "Subsystem1: Go!"


class Subsystem2:
    # 子系统 2:表示复杂系统中的子系统
    def operation1(self) -> str:
        return "Subsystem2: Ready!"

    # ... ...

    def operation_z(self) -> str:
        return "Subsystem2: Go!"


class Facade:
    """
    Facade 类封装了 Subsystem1 和 Subsystem2 的操作,对外提供一个统一的接口 operation,简化了子系统的操作流程
    """

    def __init__(self, subsystem1: Subsystem1, subsystem2: Subsystem2) -> None:
        self._subsystem1 = subsystem1 or Subsystem1()
        self._subsystem2 = subsystem2 or Subsystem2()

    def operation(self) -> str:
        # operation,简化了子系统的操作流程
        results = []
        results.append("Facade initializes subsystems:")
        results.append(self._subsystem1.operation1())
        results.append(self._subsystem2.operation1())
        results.append("Facade orders subsystems to perform the action:")
        results.append(self._subsystem1.operation_n())
        results.append(self._subsystem2.operation_z())
        return "\n".join(results)


def client_code(facade: Facade) -> None:
    """
    客户端代码通过 Facade 提供的简单接口与复杂的子系统协同工作。
    当 Facade 管理子系统的生命周期时,客户端可能根本不知道子系统的存在。 这种方法可以让复杂性得到控制。
    """
    print(facade.operation(), end="")


if __name__ == "__main__":
    subsystem1 = Subsystem1()
    subsystem2 = Subsystem2()
    facade = Facade(subsystem1, subsystem2)
    client_code(facade)

# 输出:
# Facade initializes subsystems:
# Subsystem1: Ready!
# Subsystem2: Ready!
# Facade orders subsystems to perform the action:
# Subsystem1: Go!
# Subsystem2: Go!

扩展性: 当子系统发生变化时,客户端不需要修改,只需调整 Facade 类的实现,从而有效地隔离了客户端与复杂系统的内部结构;

外观模式 与 模板方法区别:

  • 外观模式:
  • 目的: 为复杂子系统提供一个统一的接口,简化客户端的使用;
  • 实现: 外观类封装子系统的细节,对外提供简化的接口;
  • 使用场景: 当系统过于复杂,需要对外提供一个简单接口时;
  • 模板方法:
  • 目的: 定义算法的骨架,将一些步骤的实现延迟到子类;
  • 实现: 在抽象类中定义模板方法,在子类中实现具体步骤;
  • 使用场景: 多个类有相同的算法结构,但具体实现不同;
  • 区别:
  • 使用意图: 外观模式简化接口,模板方法模式定义算法骨架;
  • 实现方式: 外观模式侧重封装子系统,模板方法模式侧重算法步骤的控制;

11 原型模式(Prototype Pattern)

原型模式(Prototype Pattern) 是一种创建型设计模式,它通过复制现有对象来创建新对象,而不是通过构造函数重新创建。这可以避免昂贵的对象创建操作,尤其在初始化代价高昂的情况下使用。

核心思想:

  • 原型模式要求对象实现一个 clone() 方法,以便在不暴露复杂逻辑的情况下复制自身;
  • 当需要大量相似对象时,可以使用原型模式减少创建开销;

代码示例1:

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
import copy
from abc import ABC, abstractmethod


# 定义原型接口
class Prototype(ABC):

    @abstractmethod
    def clone(self):
        pass


# 具体原型类
class ConcretePrototype(Prototype):
    def __init__(self, value: str):
        self.value = value

    def clone(self):
        return copy.deepcopy(self)


# 客户端代码
if __name__ == "__main__":
    prototype1 = ConcretePrototype("prototype1")
    clone1 = prototype1.clone()
    print(f"Original: {prototype1.value}")
    print(f"Clone: {clone1.value}")
    clone1.value = "clone1"
    print(f"Original: {prototype1.value}")
    print(f"Clone: {clone1.value}")

# 输出:
# Original: prototype1
# Clone: prototype1
# Original: prototype1
# Clone: clone1

说明:

  • 原型 (Prototype) 接口将对克隆方法进行声明。 在绝大多数情况下, 其中只会有一个名为 clone 克隆的方法;
  • 具体原型 (Concrete Prototype) 类将实现克隆(clone)方法,除了将原始对象的数据复制到克隆体中之外,该方法有时还需处理克隆过程中的极端情况,例如克隆关联对象和梳理递归依赖等等。这里采用 deepcopy 来确保深层次的复制;
  • 客户端 (Client) 可以复制实现了原型接口的任何对象;

【扩展知识点:copy.deepcopy】
deepcopy 是 Python 中 copy 模块提供的一个函数,它用于深度复制对象。与浅拷贝(copy.copy() )不同,深拷贝会递归地复制对象及其包含的所有子对象。因此,原始对象和深拷贝对象完全独立,修改深拷贝对象不会影响原始对象。

浅拷贝 vs 深拷贝:

  • 浅拷贝:只复制对象的引用,嵌套的子对象仍然引用同一块内存;
  • 深拷贝:不仅复制对象本身,还递归复制所有子对象;
    > 示例:

```python import copy

original = [1, [2, 3]] shallow_copy = copy.copy(original) deep_copy = copy.deepcopy(original)

修改嵌套列表

original[1][0] = 100

print(original) # [1, [100, 3]] print(shallow_copy) # [1, [100, 3]] 受影响 print(deep_copy) # [1, [2, 3]] 不受影响

```

代码示例2:

这段示例代码展示了 Python 中如何通过 copy 模块实现自定义的浅拷贝和深拷贝。

Python
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
import copy


class SelfReferencingEntity:
    """
    表示一个包含父对象的类,可以设置和引用其他对象。
    """

    def __init__(self):
        self.parent = None

    def set_parent(self, parent):
        """
        设置父对象为传入的对象实例。
        """
        self.parent = parent


class SomeComponent:
    """
    一个复杂对象,包含整型值、对象列表和循环引用。
    该类通过自定义的 `__copy__` 和 `__deepcopy__` 来实现浅拷贝和深拷贝。
    """

    def __init__(self, some_int, some_list_of_objects, some_circular_ref):
        self.some_int = some_int  # 一个整型值
        self.some_list_of_objects = some_list_of_objects  # 一个包含对象的列表
        self.some_circular_ref = some_circular_ref  # 一个循环引用对象

    def __copy__(self):
        """
        实现浅拷贝:只复制对象的引用。
        当调用 copy.copy() 时,会调用此方法。
        """
        # 对嵌套的对象进行浅拷贝
        some_list_of_objects = copy.copy(self.some_list_of_objects)
        some_circular_ref = copy.copy(self.some_circular_ref)

        # 克隆对象并使用准备好的嵌套对象副本
        new = self.__class__(
            self.some_int, some_list_of_objects, some_circular_ref
        )
        new.__dict__.update(self.__dict__)  # 更新新对象的属性

        return new

    def __deepcopy__(self, memo=None):
        """
        实现深拷贝:递归复制对象及其嵌套对象。
        当调用 copy.deepcopy() 时,会调用此方法。
        memo 是一个防止循环引用的字典。
        """
        if memo is None:
            memo = {}

        # 对嵌套的对象进行深拷贝
        some_list_of_objects = copy.deepcopy(self.some_list_of_objects, memo)
        some_circular_ref = copy.deepcopy(self.some_circular_ref, memo)

        # 克隆对象并使用准备好的嵌套对象副本
        new = self.__class__(
            self.some_int, some_list_of_objects, some_circular_ref
        )
        new.__dict__ = copy.deepcopy(self.__dict__, memo)

        return new


if __name__ == "__main__":

    # 初始化对象及循环引用
    list_of_objects = [1, {1, 2, 3}, [1, 2, 3]]
    circular_ref = SelfReferencingEntity()
    component = SomeComponent(23, list_of_objects, circular_ref)
    circular_ref.set_parent(component)

    # 浅拷贝对象
    shallow_copied_component = copy.copy(component)

    # 修改浅拷贝中的列表,检查原对象是否受影响
    shallow_copied_component.some_list_of_objects.append("another object")
    if component.some_list_of_objects[-1] == "another object":
        print("浅拷贝的修改影响了原对象")
    else:
        print("浅拷贝的修改未影响原对象")

    # 修改集合,检查浅拷贝是否受影响
    component.some_list_of_objects[1].add(4)
    if 4 in shallow_copied_component.some_list_of_objects[1]:
        print("修改原对象集合影响了浅拷贝")
    else:
        print("修改原对象集合未影响浅拷贝")

    # 深拷贝对象
    deep_copied_component = copy.deepcopy(component)

    # 修改深拷贝中的列表,检查原对象是否受影响
    deep_copied_component.some_list_of_objects.append("one more object")
    if component.some_list_of_objects[-1] == "one more object":
        print("深拷贝的修改影响了原对象")
    else:
        print("深拷贝的修改未影响原对象")

    # 修改集合,检查深拷贝是否受影响
    component.some_list_of_objects[1].add(10)
    if 10 in deep_copied_component.some_list_of_objects[1]:
        print("修改原对象集合影响了深拷贝")
    else:
        print("修改原对象集合未影响深拷贝")

    # 打印循环引用的 ID,证明深拷贝处理了循环引用
    print(
        f"id(deep_copied_component.some_circular_ref.parent): "
        f"{id(deep_copied_component.some_circular_ref.parent)}"
    )
    print(
        f"id(deep_copied_component.some_circular_ref.parent.some_circular_ref.parent): "
        f"{id(deep_copied_component.some_circular_ref.parent.some_circular_ref.parent)}"
    )
    print("^^ 深拷贝处理了循环引用,并没有重复克隆。")

# 输出:
# 浅拷贝的修改影响了原对象
# 修改原对象集合影响了浅拷贝
# 深拷贝的修改未影响原对象
# 修改原对象集合未影响深拷贝
# id(deep_copied_component.some_circular_ref.parent): 140576636458848
# id(deep_copied_component.some_circular_ref.parent.some_circular_ref.parent): 140576636458848
# ^^ 深拷贝处理了循环引用,并没有重复克隆。

代码解释:

  • SelfReferencingEntity 类
    这个类的主要目的是展示循环引用问题,它拥有一个 parent 属性,表示该对象的父级。
  • set_parent 方法:设置当前对象的父级;

  • SomeComponent 类
    此类表示一个带有复杂结构的组件,其中包含一个整数、一个对象列表以及一个循环引用。此类通过实现 __copy____deepcopy__ 方法,展示如何自定义浅拷贝和深拷贝的行为。

  • 构造函数:接受一个整数、一个对象列表和一个循环引用对象作为参数。
  • __copy__ 方法:实现浅拷贝,使用 copy.copy 对象的嵌套属性来创建新实例。
    • copy.copy(self.some_list_of_objects) 创建了对象列表的浅拷贝;
    • 最终返回的 new 是浅拷贝对象,且其属性是对原对象嵌套对象的引用;
  • __deepcopy__ 方法:实现深拷贝,通过 copy.deepcopy 递归地复制嵌套对象,并通过 memo 参数避免循环引用问题。

    • memo 是一个字典,记录已经复制过的对象,防止无限递归;
  • 测试部分
    __main__ 部分中,创建了一个嵌套对象 component 和 circular_ref,并分别展示了浅拷贝和深拷贝的行为差异。

  • 浅拷贝测试:
    • 修改 shallow_copied_component.some_list_of_objects 会影响原始对象 component,因为它们共享同一个引用;
    • 修改 component.some_list_of_objects[1] 中的集合也会影响浅拷贝对象,因为这只是浅层引用;
  • 深拷贝测试:
    • 修改 deep_copied_component.some_list_of_objects 不会影响原对象 component,因为深拷贝创建了完全独立的副本;
    • 循环引用测试:deep_copied_component 的 parent 属性中的引用保持不变,避免了无限循环;

主要原理:

  • 浅拷贝 (copy):
  • 拷贝对象本身,但嵌套的对象(如列表、集合等)仍与原对象共享;
  • 修改嵌套对象会影响拷贝和原对象;
  • 深拷贝 (deepcopy):
  • 递归复制对象及其嵌套的对象,确保它们独立;
  • 使用 memo 参数避免循环引用导致的无限递归;
  • 循环引用:
  • 深拷贝处理了循环引用的情况,通过 memo 确保不会无限递归;
  • 此代码展示了原型模式如何通过浅拷贝和深拷贝来复制复杂对象,同时处理嵌套对象和循环引用问题;

12 状态模式(State Pattern)

状态模式是一种行为设计模式,允许对象在内部状态改变时改变其行为。这种模式对于对象在不同状态下有不同行为的场景非常有用。它通过将状态的行为分离到独立的类中,从而简化了状态的管理,避免了大量的条件分支。

核心概念:

  • Context: 状态的管理对象,维护当前状态并将行为委托给状态对象;
  • State: 表示对象的状态,定义状态下的行为;
  • Concrete States: 具体状态的实现类;

代码示例:

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
from abc import ABC, abstractmethod


class State(ABC):
    """
    定义状态类接口,所有具体状态类需要实现此接口的行为
    """

    @abstractmethod
    def handle(self, context):
        pass


class ConcreteStateReady(State):
    """
    具体状态 Ready 的行为实现
    """

    def handle(self, context):
        print("State Ready: 处理请求,切换到 State Start")
        context.set_state(ConcreteStateStart())


class ConcreteStateStart(State):
    """
    具体状态 Start 的行为实现
    """

    def handle(self, context):
        print("State Start: 处理请求,切换到 State End")
        context.set_state(ConcreteStateEnd())


class ConcreteStateEnd(State):
    """
    具体状态 End 的行为实现
    """

    def handle(self, context):
        print("State End: 处理请求,切换到 State Ready")
        context.set_state(ConcreteStateReady())


class Context:
    """
    上下文类,持有状态,并将请求委托给当前状态处理
    """

    def __init__(self, state: State):
        self._state = state  # 初始化时设置初始状态

    def set_state(self, state: State):
        """
        改变当前状态
        """
        self._state = state

    def request(self):
        """
        将请求委托给当前状态对象处理
        """
        self._state.handle(self)


# 使用状态模式
if __name__ == "__main__":
    context = Context(ConcreteStateReady())  # 设置初始状态为 State Ready
    context.request()  # State Ready: 切换到 State Start
    context.request()  # State Start: 切换到 State End
    context.request()  # State End: 切换到 State Ready

# 输出:
# State Ready: 处理请求,切换到 State Start
# State Start: 处理请求,切换到 State End
# State End: 处理请求,切换到 State Ready

解释:

  • Context 类负责维护当前状态,并在需要时切换状态;
  • State 类是状态的抽象基类,具体状态类继承它并实现各自的 handle() 方法;
  • ConcreteStateA 和 ConcreteStateB 是两个具体状态,它们的 handle() 方法执行特定操作并切换到另一个状态;

通过这种设计,状态的变化是动态的,避免了复杂的 if-else 或 switch 语句,并且每个状态的行为都被封装在对应的类中,使得代码更加清晰和可扩展。

更多设计模式

本站包含各种设计模式及用例(创建新模式、结构性模式、行为模式):https://refactoringguru.cn/design-patterns/python