设计模式_Python

2025-07-14 7761 words 16 minutes

Contents

Nature's Beauty

Explore the wonders of the world

参考资料

https://refactoringguru.cn/design-patterns/python

设计模式的存在就是为了抵御需求变更，这个真理我直到工作之后才能明白。你必须把一个软件的架构设计得如此之好，才能在需求大规模变更之后，还能在整体上让你的代码是漂亮的、易于修改的、高性能的、并且是安全的。每一次改动都不能是打补丁，你总是需要重构来使得你的代码在任何一刻都在整体上是好的。为了达到这个目标，就需要熟练掌握并使用设计模式来开发项目。 —— vczh

创建型模式

工厂

它将对象的创建与使用分离，提高了代码的灵活性和可维护性，是面向对象设计中非常常用的模式之一
在工厂模式中，我们创建对象时不会对客户端暴露创建逻辑，而是通过使用一个共同的接口来指向新创建的对象

实例：支付系统

 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


from __future__ import annotations
from abc import ABC, abstractmethod


class PaymentProcessor(ABC):
    """
    Product，产品接口，支付处理器接口
    """
    @abstractmethod
    def process_payment(self, amount: float) -> str:
        pass

class CreditCardProcessor(PaymentProcessor):
    """
    ConcreteProducts，具体产品，信用卡支付处理器
    """
    def process_payment(self, amount: float) -> str:
        return f"Processing ${amount:.2f} via Credit Card"

class PayPalProcessor(PaymentProcessor):
    """
    ConcreteProducts，具体产品，PayPal 支付处理器
    """
    def process_payment(self, amount: float) -> str:
        return f"Processing ${amount:.2f} via PayPal"

class BankTransferProcessor(PaymentProcessor):
    """
    ConcreteProducts，具体产品，银行转账支付处理器
    """
    def process_payment(self, amount: float) -> str:
        return f"Processing ${amount:.2f} via Bank Transfer"

class PaymentProcessorFactory(ABC):
    """
    Creator，抽象工厂，支付处理器工厂
    """
    @abstractmethod
    def create_processor(self) -> PaymentProcessor:
        pass

    def execute_payment(self, amount: float) -> str:
        processor = self.create_processor()
        return processor.process_payment(amount)

class CreditCardFactory(PaymentProcessorFactory):
    """
    ConcreteCreators，具体工厂，信用卡支付工厂
    """
    def create_processor(self) -> PaymentProcessor:
        return CreditCardProcessor()

class PayPalFactory(PaymentProcessorFactory):
    """
    ConcreteCreators，具体工厂，PayPal 支付工厂
    """
    def create_processor(self) -> PaymentProcessor:
        return PayPalProcessor()

class BankTransferFactory(PaymentProcessorFactory):
    """
    ConcreteCreators，具体工厂，银行转账支付工厂
    """
    def create_processor(self) -> PaymentProcessor:
        return BankTransferProcessor()


def process_client_payment(factory: PaymentProcessorFactory, amount: float):
    """
    客户端代码，无需知道具体产品类
    """
    print(f"Client: Initiating payment...")
    result = factory.execute_payment(amount)
    print(result)


if __name__ == "__main__":
    # 使用信用卡支付
    process_client_payment(CreditCardFactory(), 100.50)

    # 使用PayPal支付
    process_client_payment(PayPalFactory(), 75.25)

    # 使用银行转账支付
    process_client_payment(BankTransferFactory(), 200.00)


# 运行结果
# Client: Initiating payment...
# Processing $100.50 via Credit Card
# Client: Initiating payment...
# Processing $75.25 via PayPal
# Client: Initiating payment...
# Processing $200.00 via Bank Transfer

由上可知，工厂模式需要实现：

Creator (抽象工厂)
抽象基类，定义工厂方法和业务操作，返回产品对象
ConcreteCreator (具体工厂)
实现各自的工厂方法，返回具体产品实例
Product (产品接口)
定义产品对象的接口
ConcreteProduct (具体产品)
实现产品接口的具体类

工厂模式的优点：

解耦：客户端代码不需要知道具体支付处理器的实现细节
可维护性：每种支付方式的逻辑封装在各自的类中
可扩展性：新增支付方式只需添加新的具体产品和工厂类

抽象工厂

它能创建一系列相关或相互依赖的对象（产品族），而无需指定它们的具体类

实例：跨平台 UI 组件库

  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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149


from __future__ import annotations
from abc import ABC, abstractmethod


class Button(ABC):
    """
    AbstractProduct，抽象产品，按钮
    """
    @abstractmethod
    def render(self) -> str:
        pass

    @abstractmethod
    def on_click(self) -> str:
        pass

class Checkbox(ABC):
    """
    AbstractProduct，抽象产品，复选框
    """
    @abstractmethod
    def render(self) -> str:
        pass

    @abstractmethod
    def on_check(self) -> str:
        pass

class WindowsButton(Button):
    """
    ConcreteProduct，Windows风格的具体产品，按钮
    """
    def render(self) -> str:
        return "渲染一个Windows风格的按钮"

    def on_click(self) -> str:
        return "处理Windows按钮点击事件"

class WindowsCheckbox(Checkbox):
    """
    ConcreteProduct，Windows风格的具体产品，复选框
    """
    def render(self) -> str:
        return "渲染一个Windows风格的复选框"

    def on_check(self) -> str:
        return "处理Windows复选框选中事件"

class MacOSButton(Button):
    """
    ConcreteProduct，MacOS风格的具体产品，按钮
    """
    def render(self) -> str:
        return "渲染一个MacOS风格的按钮"

    def on_click(self) -> str:
        return "处理MacOS按钮点击事件"

class MacOSCheckbox(Checkbox):
    """
    ConcreteProduct，MacOS风格的具体产品，复选框
    """
    def render(self) -> str:
        return "渲染一个MacOS风格的复选框"

    def on_check(self) -> str:
        return "处理MacOS复选框选中事件"

class GUIFactory(ABC):
    """
    AbstractFactory，抽象工厂
    """
    @abstractmethod
    def create_button(self) -> Button:
        pass

    @abstractmethod
    def create_checkbox(self) -> Checkbox:
        pass

class WindowsFactory(GUIFactory):
    """
    ConcreteFactory，具体工厂，Windows 工厂
    """
    def create_button(self) -> Button:
        return WindowsButton()

    def create_checkbox(self) -> Checkbox:
        return WindowsCheckbox()

class MacOSFactory(GUIFactory):
    """
    ConcreteFactory，具体工厂，MacOS 工厂
    """
    def create_button(self) -> Button:
        return MacOSButton()

    def create_checkbox(self) -> Checkbox:
        return MacOSCheckbox()


class Application:
    """
    客户端代码，不依赖具体 UI 组件类
    """
    def __init__(self, factory: GUIFactory):
        self.factory = factory
        self.button = None
        self.checkbox = None

    def create_ui(self):
        self.button = self.factory.create_button()
        self.checkbox = self.factory.create_checkbox()

    def render_ui(self):
        print(self.button.render())
        print(self.checkbox.render())

    def simulate_user_interaction(self):
        print(self.button.on_click())
        print(self.checkbox.on_check())


if __name__ == "__main__":
    print("启动Windows风格的应用程序:")
    app = Application(WindowsFactory())
    app.create_ui()
    app.render_ui()
    app.simulate_user_interaction()

    print("\n启动MacOS风格的应用程序:")
    app = Application(MacOSFactory())
    app.create_ui()
    app.render_ui()
    app.simulate_user_interaction()


# 输出结果：
# 启动Windows风格的应用程序:
# 渲染一个Windows风格的按钮
# 渲染一个Windows风格的复选框
# 处理Windows按钮点击事件
# 处理Windows复选框选中事件
#
# 启动MacOS风格的应用程序:
# 渲染一个MacOS风格的按钮
# 渲染一个MacOS风格的复选框
# 处理MacOS按钮点击事件
# 处理MacOS复选框选中事件

由上可知，抽象工厂模式需要实现：

AbstractFactory (抽象工厂)
声明创建一系列产品族的方法
ConcreteFactory (具体工厂)
实现抽象工厂的方法，创建具体产品族
AbstractProduct (抽象产品)
为每种产品声明接口
ConcreteProduct (具体产品)
实现抽象产品接口的具体类

抽象工厂模式的适用场景：

跨平台 UI 工具包(如 Qt、Java AWT/Swing)
支持多种数据库的数据库访问层
不同外观风格的主题系统
游戏引擎中不同渲染风格的资源创建

抽象工厂模式的优点：

一致性：确保同一风格的所有 UI 组件一起工作
可扩展性：添加新平台(如 Linux)只需添加新的工厂和产品类
解耦：客户端代码不依赖具体 UI 组件类
可维护性：产品创建逻辑集中在工厂中

对比

对比维度	工厂方法模式	抽象工厂模式
核心目的	创建单一产品对象	创建多个相关或依赖的产品族(产品系列)
工厂角色	一个抽象工厂类，每个具体工厂类只负责创建一个具体产品	一个抽象工厂类，每个具体工厂类负责创建一族相关产品
产品维度	只处理一种产品	处理多种相关产品，强调产品之间的兼容性
扩展方向	通过添加新的具体工厂来支持新产品	通过添加新的具体工厂来支持新的产品族
适用场景	需要创建单一类型对象，但具体类型可能在子类中决定	需要创建多个相关或依赖的对象，确保这些对象能一起工作

工厂方法模式关注的是单一产品的创建，而抽象工厂模式关注的是产品族的创建
选择哪种模式取决于你的系统是需要创建单一类型对象还是需要创建一系列相关对象
在实际开发中，这两种模式经常结合使用。例如，抽象工厂中的每个创建方法可能会使用工厂方法来实现

生成器

它允许你分步骤创建复杂对象
该模式的主要目的是将一个复杂对象的 构建与表示分离，使得同样的构建过程可以创建不同的表示

实例：构建电脑配置

  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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150


from __future__ import annotations
from abc import ABC, abstractmethod


class Computer:
    """
    Product，产品类，电脑
    """
    def __init__(self):
        self.cpu = None
        self.ram = None
        self.storage = None
        self.gpu = None
        self.motherboard = None

    def __str__(self):
        return (f"Computer Configuration:\n"
                f"  CPU: {self.cpu}\n"
                f"  RAM: {self.ram}\n"
                f"  Storage: {self.storage}\n"
                f"  GPU: {self.gpu}\n"
                f"  Motherboard: {self.motherboard}\n")


class ComputerBuilder(ABC):
    """
    Builder，抽象生成器
    """
    @abstractmethod
    def set_cpu(self, cpu: str) -> None:
        pass

    @abstractmethod
    def set_ram(self, ram: str) -> None:
        pass

    @abstractmethod
    def set_storage(self, storage: str) -> None:
        pass

    @abstractmethod
    def set_gpu(self, gpu: str) -> None:
        pass

    @abstractmethod
    def set_motherboard(self, motherboard: str) -> None:
        pass

    @abstractmethod
    def get_computer(self) -> Computer:
        pass


class GamingComputerBuilder(ComputerBuilder):
    """
    ConcreteBuilder，具体生成器，游戏电脑生成器
    """
    def __init__(self):
        self.computer = Computer()

    def set_cpu(self, cpu: str) -> None:
        self.computer.cpu = f"High-end {cpu}"

    def set_ram(self, ram: str) -> None:
        self.computer.ram = f"Fast {ram} RAM"

    def set_storage(self, storage: str) -> None:
        self.computer.storage = f"SSD {storage}"

    def set_gpu(self, gpu: str) -> None:
        self.computer.gpu = f"Powerful {gpu}"

    def set_motherboard(self, motherboard: str) -> None:
        self.computer.motherboard = f"Gaming {motherboard}"

    def get_computer(self) -> Computer:
        return self.computer


class OfficeComputerBuilder(ComputerBuilder):
    """
    ConcreteBuilder，具体生成器，办公电脑生成器
    """
    def __init__(self):
        self.computer = Computer()

    def set_cpu(self, cpu: str) -> None:
        self.computer.cpu = f"Standard {cpu}"

    def set_ram(self, ram: str) -> None:
        self.computer.ram = f"Basic {ram} RAM"

    def set_storage(self, storage: str) -> None:
        self.computer.storage = f"HDD {storage}"

    def set_gpu(self, gpu: str) -> None:
        self.computer.gpu = "Integrated graphics"

    def set_motherboard(self, motherboard: str) -> None:
        self.computer.motherboard = f"Standard {motherboard}"

    def get_computer(self) -> Computer:
        return self.computer


class ComputerDirector:
    """
    Director，指挥者
    """
    def __init__(self, builder: ComputerBuilder):
        self.builder = builder

    def build_basic_computer(self):
        self.builder.set_cpu("i5")
        self.builder.set_ram("8GB")
        self.builder.set_storage("500GB")
        self.builder.set_motherboard("ATX")

    def build_advanced_computer(self):
        self.builder.set_cpu("i9")
        self.builder.set_ram("32GB")
        self.builder.set_storage("2TB")
        self.builder.set_gpu("RTX 3080")
        self.builder.set_motherboard("Extended ATX")


if __name__ == "__main__":
    print("Building a gaming computer:")
    gaming_builder = GamingComputerBuilder()
    director = ComputerDirector(gaming_builder)
    director.build_advanced_computer()
    gaming_computer = gaming_builder.get_computer()
    print(gaming_computer)

    print("Building an office computer:")
    office_builder = OfficeComputerBuilder()
    director = ComputerDirector(office_builder)
    director.build_basic_computer()
    office_computer = office_builder.get_computer()
    print(office_computer)

    # 也可以不通过指挥者来构建产品，而直接使用 Builder
    print("Custom office computer:")
    office_builder = OfficeComputerBuilder()
    office_builder.set_cpu("i7")
    office_builder.set_ram("16GB")
    office_builder.set_storage("1TB SSD")
    office_builder.set_motherboard("Micro ATX")
    custom_office_computer = office_builder.get_computer()
    print(custom_office_computer)

由上可知，生成器模式需要实现：

Builder（抽象生成器）
定义了创建产品各个部分的抽象接口
ConcreteBuilder（具体生成器）
构造和装配产品的各个部件，明确所创建的产品表示
Product（产品）
表示被构造的复杂对象
Director（指挥者）
构建一个使用 Builder 接口的对象，控制构建过程

生成器模式的适用场景：
适用于那些需要多个步骤或部分来构建的对象，而且这些步骤或部分可能有多种组合方式的情况

生成器模式的优点：

分步构建对象：允许你分步骤构建一个复杂对象，而不是一次性构建完成
单一职责原则：将复杂对象的构建代码与其业务逻辑分离
更好的控制：对构建过程有更精细的控制

结构型模式

适配器

它允许不兼容的接口之间能够协同工作
就像现实世界中的电源适配器可以让不同国家标准的插头工作一样，适配器模式在软件中充当两个不兼容接口之间的桥梁

实例：假设我们有一个电商系统，原本只支持 PayPal 支付，现在通过适配器模式集成新的 Stripe 支付服务

 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


class PaymentProcessor:
    """
    Target，原有系统使用的支付接口
    """
    def pay(self, amount: float) -> str:
        return f"Processing ${amount} via default payment processor"

class StripePayment:
    """
    Adaptee，新的 Stripe 支付服务，但是不兼容原有的接口
    """
    def make_payment(self, amount_in_cents: int, currency: str) -> str:
        return f"Processed {amount_in_cents/100} {currency} via Stripe"

class StripeAdapter(PaymentProcessor, StripePayment):
    """
    Adapter，适配器
    """
    def pay(self, amount: float) -> str:
        # 转换金额单位为分，并默认使用USD
        return self.make_payment(int(amount * 100), "USD")

def process_payment(processor: PaymentProcessor, amount: float):
    """
    客户端代码，处理支付
    """
    print(processor.pay(amount))


if __name__ == "__main__":
    print("Using default payment processor:")
    default_processor = PaymentProcessor()
    process_payment(default_processor, 100.50)

    print("\nUsing Stripe payment via adapter:")
    stripe_adapter = StripeAdapter()
    process_payment(stripe_adapter, 100.50)

# 运行结果
# Using default payment processor:
# Processing $100.5 via default payment processor
#
# Using Stripe payment via adapter:
# Processed 100.5 USD via Stripe

由上可知，适配器模式需要实现：

Target
需要进行改造的目标类，定义了客户端使用的接口
Adaptee
新的待适配功能，但接口不兼容
Adapter
适配器，多重继承 Target 和 Adaptee，并实现了 Target 的request()方法
在内部调用 Adaptee 的方法并对其进行转换

适配器模式适用场景：

集成第三方库：当你需要使用某个类，但其接口与你的代码不兼容时
复用旧代码：在重构系统时，让新接口能够使用旧代码
统一多个类的接口：当你有多个类似功能的类但有不同接口时

行为模式

策略

它定义了一系列算法，并将每个算法封装起来，使它们可以相互替换
策略模式让算法的变化独立于使用算法的客户端

实例：支持多种支付方式的电商支付系统

 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


from __future__ import annotations
from abc import ABC, abstractmethod
from typing import List


class PaymentContext:
    """
    Context，支付上下文，负责处理支付请求
    """

    def __init__(self, strategy: PaymentStrategy) -> None:
        # 初始化时，传入策略，选择某种支付方式
        self._strategy = strategy

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

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

    def execute_payment(self, amount: float) -> None:
        """
        执行支付
        """
        print(f"准备支付金额: {amount}元")
        result = self._strategy.process_payment(amount)
        print(f"支付结果: {result}")


class PaymentStrategy(ABC):
    """
    Strategy，支付策略接口
    """

    @abstractmethod
    def process_payment(self, amount: float) -> str:
        pass


class AlipayStrategy(PaymentStrategy):
    """
    ConcreteStrategy，支付宝支付
    """

    def process_payment(self, amount: float) -> str:
        return f"使用支付宝成功支付{amount}元"

class WechatPayStrategy(PaymentStrategy):
    """
    ConcreteStrategy，微信支付
    """

    def process_payment(self, amount: float) -> str:
        return f"使用微信支付成功支付{amount}元"

class CreditCardStrategy(PaymentStrategy):
    """
    ConcreteStrategy，信用卡支付
    """

    def process_payment(self, amount: float) -> str:
        return f"使用信用卡成功支付{amount}元"


if __name__ == "__main__":
    # 用户选择支付宝支付
    context = PaymentContext(AlipayStrategy())
    context.execute_payment(100.50)
    print()

    # 用户选择微信支付
    context.strategy = WechatPayStrategy()
    context.execute_payment(200.75)
    print()

    # 用户选择信用卡支付
    context.strategy = CreditCardStrategy()
    context.execute_payment(150.00)


# 运行结果
# 准备支付金额: 100.5元
# 支付结果: 使用支付宝成功支付100.5元

# 准备支付金额: 200.75元
# 支付结果: 使用微信支付成功支付200.75元

# 准备支付金额: 150.0元
# 支付结果: 使用信用卡成功支付150.0元

由上可知，策略模式需要实现：

Context(上下文)
维护对策略对象的引用，并提供接口让策略对象访问其数据
通过构造函数或 setter 方法接收具体策略
需要一个方法委托策略对象执行算法，例如上述的 execute_payment()
Strategy(策略接口)
抽象类或接口，定义所有支持的算法的公共接口
定义需要实现的抽象方法，例如上述的 process_payment()
ConcreteStrategy(具体策略)
实现策略接口的具体算法类

策略模式的优点：

避免多重条件语句：替代大量的 if-else 或 switch-case 语句
开闭原则：无需修改上下文即可引入新策略
运行时切换算法：可以在运行时切换对象使用的算法

观察者

它定义了一种 一对多的依赖关系
当一个对象（称为"主题"）的状态发生改变时，所有依赖于它的对象（称为"观察者"）都会自动得到通知并被更新

实例：股票价格监控系统

  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
130


from abc import ABC, abstractmethod
from typing import List


class StockSubject(ABC):
    """
    Subject，主题接口
    """
    @abstractmethod
    def attach(self, observer: 'StockObserver') -> None:
        pass

    @abstractmethod
    def detach(self, observer: 'StockObserver') -> None:
        pass

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

class Stock(StockSubject):
    """
    ConcreteSubject，具体股票主题
    """
    def __init__(self, symbol: str, price: float):
        self.symbol = symbol
        self._price = price
        self._observers: List['StockObserver'] = []

    @property
    def price(self):
        return self._price

    @price.setter
    def price(self, value: float):
        if self._price != value:
            # 在价格变动时，进行通知
            self._price = value
            self.notify()

    def attach(self, observer: 'StockObserver') -> None:
        if observer not in self._observers:
            self._observers.append(observer)

    def detach(self, observer: 'StockObserver') -> None:
        if observer in self._observers:
            self._observers.remove(observer)

    def notify(self) -> None:
        print(f"\n{self.symbol} price changed to {self._price}. Notifying observers...")
        # 遍历主题(Subject)中注册的所有观察者(Observer)
        for observer in self._observers:
            # 对每个观察者调用其update()方法
            # 将主题自身(self)作为参数传递给观察者的update()方法
            observer.update(self)

class StockObserver(ABC):
    """
    Observer，观察者接口
    """
    @abstractmethod
    def update(self, stock: Stock) -> None:
        pass

class Investor(StockObserver):
    """
    ConcreteObserver，具体观察者：普通投资者
    """
    def __init__(self, name: str):
        self.name = name
        self.interested_stocks = {}

    def add_interest(self, stock: Stock, buy_threshold: float = None, sell_threshold: float = None):
        self.interested_stocks[stock.symbol] = {
            'stock': stock,
            'buy_threshold': buy_threshold,
            'sell_threshold': sell_threshold
        }
        stock.attach(self)

    def remove_interest(self, stock_symbol: str):
        if stock_symbol in self.interested_stocks:
            stock = self.interested_stocks[stock_symbol]['stock']
            stock.detach(self)
            del self.interested_stocks[stock_symbol]

    def update(self, stock: Stock) -> None:
        info = self.interested_stocks[stock.symbol]
        current_price = stock.price

        # 如果股票当前价格小于买入阈值，则通知观察者进行买入
        if info['buy_threshold'] is not None and current_price <= info['buy_threshold']:
            print(f"{self.name}: {stock.symbol} price dropped to {current_price} (<= {info['buy_threshold']}). BUY opportunity!")

        # 如果股票当前价格大于卖出阈值，则通知观察者进行卖出
        if info['sell_threshold'] is not None and current_price >= info['sell_threshold']:
            print(f"{self.name}: {stock.symbol} price rose to {current_price} (>= {info['sell_threshold']}). SELL opportunity!")


# 使用示例
if __name__ == "__main__":
    # 创建股票，即主题
    apple = Stock("AAPL", 150.0)
    google = Stock("GOOGL", 2800.0)

    # 创建投资者，即观察者
    alice = Investor("Alice")
    bob = Investor("Bob")

    # 设置投资兴趣
    alice.add_interest(apple, buy_threshold=145.0, sell_threshold=160.0)
    alice.add_interest(google, buy_threshold=2750.0)

    bob.add_interest(apple, sell_threshold=155.0)
    bob.add_interest(google, buy_threshold=2700.0, sell_threshold=2900.0)

    # 模拟价格变化
    apple.price = 148.0  # 小跌，无通知
    apple.price = 144.0  # 跌破Alice的买入阈值
    google.price = 2755.0  # 在Alice的买入阈值附近
    google.price = 2905.0  # 突破Bob的卖出阈值

    # Alice不再关注Google股票
    alice.remove_interest("GOOGL")
    google.price = 2950.0  # Alice不会收到通知

    # Bob不再关注任何股票
    bob.remove_interest("AAPL")
    bob.remove_interest("GOOGL")
    apple.price = 142.0  # Bob不会收到通知

由上可知，观察者模式需要实现：

Subject (主题)
提供添加、删除以及通知观察者的方法
Observer (观察者)
定义一个更新接口，用于在主题状态改变时接收通知
ConcreteSubject (具体主题)
维护主题的状态和观察者列表，当状态改变时通知所有观察者
ConcreteObserver (具体观察者)
实现观察者接口，保持与主题状态的一致性

观察者模式的优缺点：

优点
- 开闭原则：可以引入新的观察者而不需要修改主题代码
- 可以在运行时建立对象之间的关系
- 支持广播通信
缺点
- 观察者不知道彼此的存在，可能导致意外的更新
- 如果通知顺序很重要，观察者模式可能无法满足需求
- 如果观察者处理不当，可能导致性能问题

迭代器

它提供了一种顺序访问聚合对象元素的方法，而不需要暴露其底层表示
迭代器模式将遍历逻辑从集合中分离出来，使得 集合可以专注于数据存储，而迭代器专注于遍历

实例：文件系统遍历器，可以递归地遍历目录结构

 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


import os
from collections.abc import Iterable, Iterator
from typing import Any, List


class FileSystemIterator(Iterator):
    """
    迭代器类
    """
    def __init__(self, path: str, recursive: bool = False, file_only: bool = False):
        self._paths = [path]
        self._recursive = recursive
        self._file_only = file_only
        self._current_index = 0
        self._current_items = []

    def _get_items(self, path: str) -> List[str]:
        try:
            items = os.listdir(path)
            full_paths = [os.path.join(path, item) for item in items]
            if self._file_only:
                return [p for p in full_paths if os.path.isfile(p)]
            return full_paths
        except PermissionError:
            return []

    def __next__(self) -> str:
        if not self._current_items and not self._paths:
            raise StopIteration()

        if not self._current_items:
            current_path = self._paths.pop(0)
            self._current_items = self._get_items(current_path)

            if self._recursive:
                # 添加子目录以便后续遍历
                for item in self._current_items:
                    if os.path.isdir(item):
                        self._paths.append(item)

            return self.__next__()

        return self._current_items.pop(0)

class FileSystemCollection(Iterable):
    """
    集合类
    """
    def __init__(self, path: str):
        if not os.path.exists(path):
            raise ValueError(f"Path {path} does not exist")
        self._path = path

    def __iter__(self) -> FileSystemIterator:
        return FileSystemIterator(self._path)

    def get_recursive_iterator(self, file_only: bool = False) -> FileSystemIterator:
        return FileSystemIterator(self._path, recursive=True, file_only=file_only)

# 使用示例
if __name__ == "__main__":
    # 假设当前目录下有一些文件和子目录
    collection = FileSystemCollection(".")

    print("Flat traversal (files and directories):")
    # 打印当前同级目录下的所有文件和文件夹
    for item in collection:
        print(item)

    print("\nRecursive traversal (files only):")
    # 递归打印当前目录以及子目录下的所有文件
    for item in collection.get_recursive_iterator(file_only=True):
        print(item)

迭代器模式的优点：

单一职责原则：将遍历算法与集合分离，使集合和迭代器可以独立变化
并行遍历：可以同时使用多个迭代器遍历同一集合
延迟加载：当需要处理不适合全部加载到内存的大型数据集时，可以延迟计算或加载元素

命令

它将请求或简单操作转换为对象
这种转换让你能够根据不同的请求参数化方法，延迟或排队执行请求，以及支持可撤销操作

实例：智能家居控制系统

  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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144


from abc import ABC, abstractmethod
from typing import List


class Command(ABC):
    """
    Command，命令接口
    """
    @abstractmethod
    def execute(self):
        pass

    @abstractmethod
    def undo(self):
        pass

class Light:
    """
    Receiver，命令接收者，实际执行操作的对象，灯
    """
    def on(self):
        print("Light is ON")

    def off(self):
        print("Light is OFF")

class Thermostat:
    """
    Receiver，命令接收者，恒温器
    """
    def set_temperature(self, temp):
        print(f"Thermostat set to {temp}°C")
        self.temperature = temp

# Concrete Commands
class LightOnCommand(Command):
    """
    ConcreteCommand，具体命令，开灯
    """
    def __init__(self, light: Light):
        self.light = light

    def execute(self):
        self.light.on()

    def undo(self):
        self.light.off()

class LightOffCommand(Command):
    """
    ConcreteCommand，具体命令，关灯
    """
    def __init__(self, light: Light):
        self.light = light

    def execute(self):
        self.light.off()

    def undo(self):
        self.light.on()

class ThermostatSetCommand(Command):
    """
    ConcreteCommand，具体命令，恒温器设置温度
    """
    def __init__(self, thermostat: Thermostat, temperature: int):
        self.thermostat = thermostat
        self.temperature = temperature
        self.previous_temp = None

    def execute(self):
        self.previous_temp = getattr(self.thermostat, 'temperature', None)
        self.thermostat.set_temperature(self.temperature)

    def undo(self):
        if self.previous_temp is not None:
            self.thermostat.set_temperature(self.previous_temp)

class RemoteControl:
    """
    Invoker，命令调用者，遥控器
    """
    def __init__(self):
        self.on_commands: List[Command] = []
        self.off_commands: List[Command] = []
        self.undo_command = None

    def set_command(self, slot: int, on_command: Command, off_command: Command):
        self.on_commands.insert(slot, on_command)
        self.off_commands.insert(slot, off_command)

    def press_on_button(self, slot: int):
        if slot < len(self.on_commands) and self.on_commands[slot]:
            command = self.on_commands[slot]
            command.execute()
            self.undo_command = command

    def press_off_button(self, slot: int):
        if slot < len(self.off_commands) and self.off_commands[slot]:
            command = self.off_commands[slot]
            command.execute()
            self.undo_command = command

    def press_undo_button(self):
        if self.undo_command:
            print("-- Undo last command --")
            self.undo_command.undo()
            self.undo_command = None


if __name__ == "__main__":
    # 创建灯、恒温器，即命令接收者 Receiver
    living_room_light = Light()
    kitchen_light = Light()
    thermostat = Thermostat()

    # 创建具体命令
    living_room_light_on = LightOnCommand(living_room_light)
    living_room_light_off = LightOffCommand(living_room_light)
    kitchen_light_on = LightOnCommand(kitchen_light)
    kitchen_light_off = LightOffCommand(kitchen_light)
    thermostat_day = ThermostatSetCommand(thermostat, 22)
    thermostat_night = ThermostatSetCommand(thermostat, 18)

    # 创建遥控器，即命令调用者 Invoker
    remote = RemoteControl()
    remote.set_command(0, living_room_light_on, living_room_light_off)
    remote.set_command(1, kitchen_light_on, kitchen_light_off)
    remote.set_command(2, thermostat_day, thermostat_night)

    # 测试遥控器
    print("--- Testing Living Room Light ---")
    remote.press_on_button(0)
    remote.press_off_button(0)
    remote.press_undo_button()

    print("\n--- Testing Kitchen Light ---")
    remote.press_on_button(1)
    remote.press_undo_button()

    print("\n--- Testing Thermostat ---")
    remote.press_on_button(2)  # 设置为白天温度
    remote.press_off_button(2) # 设置为夜晚温度
    remote.press_undo_button() # 撤销夜晚温度设置

由上可知，命令模式需要实现：

Command 接口
定义执行和重做操作的抽象方法
ConcreteCommand
具体的命令实现，委托给 Receiver 对象处理
Receiver
命令接受者，能够执行相关操作，是实际执行命令的角色
Invoker
命令调用者

命令模式的应用场景：

GUI 按钮和菜单项：每个按钮点击或菜单选择都可以封装为一个命令对象
事务系统：实现原子操作，支持回滚
任务队列：将命令对象放入队列，在不同时间执行
网络请求：封装网络请求为命令对象，支持重试机制
游戏开发：实现游戏中的用户输入、AI 行为等

命令模式的优点：

解耦：将调用操作的对象与知道如何执行操作的对象分离
可扩展：可以轻松添加新命令而无需更改现有代码
支持撤销/重做：通过实现 undo() 方法可以轻松实现撤销功能