设计模式
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 的修改在所有实例中都会反映出来。
| 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模块本身是线程安全的,导入时只会执行一次初始化;
使用锁适用于需要严格控制访问的场景,双重检查锁更高效,而模块级别单例则简单直接。
线程安全:示例
| 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 |
|---|
| # singleton.py
class Singleton:
pass
singleton = Singleton()
|
| Python |
|---|
| # main.py
from singleton import singleton
singleton1 = singleton
singleton2 = singleton
print(singleton1 is singleton2) # 输出: True
|
说明: 在 Python 中,模块本身是线程安全的,导入时只会执行一次初始化,后续导入不会再次初始化。
2 工厂模式(Factory Pattern)
同步可参考个人另一博文:《Python: 开始使用工厂模式设计》
工厂模式是一种创建型设计模式,用于定义一个接口或抽象类来创建对象,但将实例化推迟到子类中。其核心思想是通过工厂方法来处理对象的创建,而不是在代码中直接调用构造函数。这使得代码更加灵活和可扩展,尤其是在需要通过不同条件创建不同类型对象时。
以下是常见工厂实现方式:
2.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
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 |
|---|
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 |
|---|
| 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 |
|---|
| 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:具体策略类,实现了不同的算法;

示例:对 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 |
|---|
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