软件设计模式

一、单例模式

1. 目的：确保类只有一个实例，并提供全局访问点

2. 实现方法：

   • 装饰器（简洁且线程不安全，需要加锁进行安全操作）

   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 Logger:
       def __init__(self):
           self.log_level = "INFO"
           
   logger1 = Logger()
   logger2 = Logger()
   print(logger1 is logger2)  # True[1,2](@ref)
   

   • 装饰器（线程安全版本）

     import threading
     
     def singleton(cls):
         instances = {}
         lock = threading.Lock()
         
         def get_instance(*args, **kwargs):
             if cls not in instances:
                 with lock:
                     # 双重检查锁定，确保线程安全
                     if cls not in instances:
                         instances[cls] = cls(*args, **kwargs)
             return instances[cls]
         
         return get_instance
     
     @singleton
     class Logger:
         def __init__(self):
             self.log_level = "INFO"
             
     # 测试代码
     if __name__ == "__main__":
         logger1 = Logger()
         logger2 = Logger()
         print(logger1 is logger2)  # True
     

   • 线程竞态条件下线程不安全示例

     import threading
     import time
     
     def singleton(cls):
         instances = {}
         lock = threading.Lock()
         
         def get_instance(*args, **kwargs):
             if cls not in instances:
                 # 可将锁关闭查看线程不安全
                 with lock:
                     if cls not in instances:
                         # 添加延迟以增加竞争条件概率
                         time.sleep(0.001)
                         instances[cls] = cls(*args, **kwargs)
             return instances[cls]
         
         return get_instance
     
     @singleton
     class Logger:
         def __init__(self):
             # 添加延迟以增加竞争条件概率
             time.sleep(0.001)
             self.log_level = "INFO"
             self.init_count = 0
         
         def increment_init_count(self):
             self.init_count += 1
     
     def test_thread_safety():
         results = []
         
         def test_func(thread_id):
             logger = Logger()
             logger.increment_init_count()
             results.append((thread_id, id(logger), logger.init_count))
         
         threads = []
         for i in range(20):
             t = threading.Thread(target=test_func, args=(i,))
             threads.append(t)
         
         for t in threads:
             t.start()
         
         for t in threads:
             t.join()
         # 通过对对象实例id进行去重操作判断是否重复创建实例导致线程不安全
         unique_instances = len(set(id for _, id, _ in results))
         print(f"创建了 {unique_instances} 个不同的实例")
         print(f"初始化计数: {results[0][2]}")
         
         if unique_instances > 1:
             print("❌ 单例模式被破坏")
         else:
             print("✅ 单例模式正常工作")
     
     if __name__ == "__main__":
         print("测试线程安全单例...")
         test_thread_safety()
     

   • 自定义元类（需要自行加锁实现线程安全，支持继承和多线程）

   import threading
   
   class SingletonMeta(type):
       _instances = {}
       _lock = threading.Lock()
   
       def __call__(cls, *args, **kwargs):
           with cls._lock:
               if cls not in cls._instances:
                   instance = super().__call__(*args, **kwargs)
                   cls._instances[cls] = instance
           return cls._instances[cls]
   
   class Database(metaclass=SingletonMeta):
       def __init__(self):
           self.connection = "Connected"
   
   db1 = Database()
   db2 = Database()
   print(db1 is db2)  # True[3,5](@ref)
   

   • 模块实例（天然单例，但需要提供统一的使用入口，不推荐）

   # singleton.py
   class Config:
       def __init__(self):
           self.settings = {"debug": True}
   
   config = Config()
   
   # 其他文件导入使用
   from singleton import config
   print(config.settings)  # 全局唯一实例[1,2](@ref)
   

二、工厂模式

1. 目的：封装对象创建逻辑，实现创建与使用的解耦

2. 实现方法：

   • 自定义简洁工厂：

   class Animal:
       def speak(self):
           pass
   
   class Dog(Animal):
       def speak(self):
           return "Woof!"
   
   class Cat(Animal):
       def speak(self):
           return "Meow!"
   
   class AnimalFactory:
       @staticmethod
       def create_animal(animal_type):
           if animal_type == "dog":
               return Dog()
           elif animal_type == "cat":
               return Cat()
           else:
               raise ValueError("Invalid type")
   
   dog = AnimalFactory.create_animal("dog")
   print(dog.speak())  # Woof![6,7](@ref)
   

   • 抽象工厂：

   from abc import ABC, abstractmethod
   
   class Button:
       def render(self):
           pass
   
   class WindowsButton(Button):
       def render(self):
           return "Windows风格按钮"
   
   class MacButton(Button):
       def render(self):
           return "Mac风格按钮"
   
   class GUIFactory(ABC):
       @abstractmethod
       def create_button(self):
           pass
   
   class WindowsFactory(GUIFactory):
       def create_button(self):
           return WindowsButton()
   
   class MacFactory(GUIFactory):
       def create_button(self):
           return MacButton()
   
   mac_factory = MacFactory()
   button = mac_factory.create_button()
   print(button.render())  # Mac风格按钮[8](@ref)
   

三、观察者模式

1. 目的：实现一对多的依赖关系，状态变化时自动通知观察者

2. 实现方法：

   • 简单实现：

   # 发布器
   class NewsPublisher:
       def __init__(self):
           self.subscribers = []
   
       def subscribe(self, observer):
           self.subscribers.append(observer)
   
       def notify(self, news):
           for sub in self.subscribers:
               sub.update(news)
   # 订阅器
   class Subscriber:
       def update(self, news):
           print(f"收到新闻：{news}")
   
   publisher = NewsPublisher()
   subscriber = Subscriber()
   # 发布者获取订阅者
   publisher.subscribe(subscriber)
   # 发布者进行通知
   publisher.notify("Python 5.0发布！")
   

   • 基于事件驱动

   from abc import ABC, abstractmethod
   
   class Event:
       pass
   
   class StockEvent(Event):
       def __init__(self, symbol, price):
           self.symbol = symbol
           self.price = price
   
   class Observer(ABC):
       @abstractmethod
       def on_event(self, event):
           pass
   
   class StockTrader(Observer):
       def on_event(self, event):
           if isinstance(event, StockEvent):
               print(f"股票{event.symbol}价格更新：{event.price}")
   
   class StockMarket:
       def __init__(self):
           self.observers = []
   
       def add_observer(self, observer):
           self.observers.append(observer)
   
       def update_price(self, symbol, price):
           event = StockEvent(symbol, price)
           for obs in self.observers:
               obs.on_event(event)
   
   market = StockMarket()
   trader = StockTrader()
   market.add_observer(trader)
   # 触发价格变动事件后自动进行观察者通知
   market.update_price("AAPL", 150.5)
   

四、策略模式

1. 目的：简化不同策略下相同动作

2. 实现方法：

   class PaymentStrategy(ABC):
       @abstractmethod
       def pay(self, amount):
           pass
   
   class CreditCardPayment(PaymentStrategy):
       def pay(self, amount):
           return f"信用卡支付{amount}元"
   
   class AlipayPayment(PaymentStrategy):
       def pay(self, amount):
           return f"支付宝支付{amount}元"
   
   class PaymentContext:
       def __init__(self, strategy):
           self.strategy = strategy
   
       def execute_payment(self, amount):
           return self.strategy.pay(amount)
   
   context = PaymentContext(AlipayPayment())
   print(context.execute_payment(100))  # 支付宝支付100元[11](@ref)
   

五、适配器模式

1. 目的：使原本接口不兼容的类可以一起工作，将一个类的接口转换成客户端期望的另一个接口，常用于：

   • 集成第三方库或API时接口不匹配
   • 将旧系统接口适配到新系统
   • 统一不同数据源的访问接口

2. 实现方法：

   # 假设有一个旧的用户认证系统
   class OldAuthSystem:
       def login_user(self, username, password):
           print(f"Old system: Logging in {username}")
           return {"status": "success", "user_id": 123}
   
   # 新系统期望的认证接口
   class NewAuthInterface:
       def authenticate(self, credentials):
           pass
   
   # 适配器类
   class AuthAdapter(NewAuthInterface):
       def __init__(self, old_system):
           self.old_system = old_system
       
       def authenticate(self, credentials):
           # 将新接口的参数转换为旧系统期望的格式
           username = credentials.get("email", "").split("@")[0]
           password = credentials.get("password")
           return self.old_system.login_user(username, password)
   
   # 使用适配器
   old_auth = OldAuthSystem()
   adapter = AuthAdapter(old_auth)
   result = adapter.authenticate({"email": "user@example.com", "password": "secret"})
   print(result)
   

六、代理模式

1. 目的：为其他对象提供一种代理以控制对这个对象的访问，比如

   • 延迟加载（Lazy Loading）大型资源
   • 访问控制与权限验证
   • 缓存请求结果
   • 记录日志和监控

2. 实现方法：

   from abc import ABC, abstractmethod
   
   # 主题接口
   class ImageLoader(ABC):
       @abstractmethod
       def load_image(self, path):
           pass
   
   # 真实主题
   class RealImageLoader(ImageLoader):
       def load_image(self, path):
           print(f"Loading heavy image from {path}")
           return f"Image data from {path}"
   
   # 代理类
   class ImageLoaderProxy(ImageLoader):
       def __init__(self):
           self.real_loader = None
           self.cache = {}
       
       def load_image(self, path):
           # 检查缓存
           if path in self.cache:
               print(f"Returning cached image for {path}")
               return self.cache[path]
           
           # 延迟初始化真实加载器
           if self.real_loader is None:
               self.real_loader = RealImageLoader()
           
           # 调用真实加载器并缓存结果
           result = self.real_loader.load_image(path)
           self.cache[path] = result
           return result
   
   # 使用代理
   loader = ImageLoaderProxy()
   print(loader.load_image("/images/large.jpg"))  # 第一次加载，从磁盘读取
   print(loader.load_image("/images/large.jpg"))  # 第二次加载，从缓存返回
   

七、责任链模式

1. 目的：使多个对象都有机会处理请求，从而避免请求的发送者和接收者耦合在一起，常用于

   • 中间件处理HTTP请求
   • 权限验证链
   • 数据验证和处理流水线
   • 异常处理链

2. 实现方法：

   from abc import ABC, abstractmethod
   
   # 处理器接口
   class Handler(ABC):
       def __init__(self, successor=None):
           self.successor = successor
       
       @abstractmethod
       def handle_request(self, request):
           pass
   
   # 具体处理器 - 认证
   class AuthenticationHandler(Handler):
       def handle_request(self, request):
           if "auth_token" not in request:
               return {"error": "Authentication required"}
           
           if request["auth_token"] != "valid_token":
               return {"error": "Invalid token"}
           
           print("Authentication passed")
           if self.successor:
               return self.successor.handle_request(request)
           return {"status": "success"}
   
   # 具体处理器 - 授权
   class AuthorizationHandler(Handler):
       def handle_request(self, request):
           if request.get("user_role") != "admin":
               return {"error": "Insufficient permissions"}
           
           print("Authorization passed")
           if self.successor:
               return self.successor.handle_request(request)
           return {"status": "success"}
   
   # 具体处理器 - 验证
   class ValidationHandler(Handler):
       def handle_request(self, request):
           if not request.get("data"):
               return {"error": "No data provided"}
           
           print("Validation passed")
           if self.successor:
               return self.successor.handle_request(request)
           return {"status": "success"}
   
   # 构建责任链
   validation = ValidationHandler()
   authorization = AuthorizationHandler(validation)
   authentication = AuthenticationHandler(authorization)
   
   # 使用责任链处理请求
   request1 = {"auth_token": "valid_token", "user_role": "admin", "data": "some data"}
   result1 = authentication.handle_request(request1)
   print(result1)
   
   request2 = {"auth_token": "invalid_token", "user_role": "admin", "data": "some data"}
   result2 = authentication.handle_request(request2)
   print(result2)
   

八、命令模式

1. 目的：将请求封装为对象，从而可以用不同的请求对客户进行参数化，避免对其他客户端命令的硬嵌入，常用于

   • 实现撤销/重做功能
   • 任务队列和异步处理
   • 宏命令（一组命令的集合）
   • 事务性操作

2. 实现方法：

   from abc import ABC, abstractmethod
   
   # 命令接口
   class Command(ABC):
       @abstractmethod
       def execute(self):
           pass
       
       @abstractmethod
       def undo(self):
           pass
   
   # 接收者 - 数据库操作
   class Database:
       def insert(self, data):
           print(f"Inserting {data} into database")
           return True
       
       def delete(self, id):
           print(f"Deleting record {id} from database")
           return True
       
       def update(self, id, data):
           print(f"Updating record {id} with {data}")
           return True
   
   # 具体命令 - 插入命令
   class InsertCommand(Command):
       def __init__(self, database, data):
           self.database = database
           self.data = data
           self.executed = False
       
       def execute(self):
           result = self.database.insert(self.data)
           self.executed = True
           return result
       
       def undo(self):
           if self.executed:
               print("Undoing insert operation")
               # 实际应用中可能需要记录插入的ID以便删除
               return True
           return False
   
   # 具体命令 - 更新命令
   class UpdateCommand(Command):
       def __init__(self, database, id, data):
           self.database = database
           self.id = id
           self.data = data
           self.old_data = None
           self.executed = False
       
       def execute(self):
           # 实际应用中应该先获取旧数据
           self.old_data = f"old_data_for_{self.id}"
           result = self.database.update(self.id, self.data)
           self.executed = True
           return result
       
       def undo(self):
           if self.executed and self.old_data:
               print(f"Undoing update operation, restoring {self.old_data}")
               return self.database.update(self.id, self.old_data)
           return False
   
   # 调用者
   class Invoker:
       def __init__(self):
           self.history = []
           self.redo_stack = []
       
       def execute_command(self, command):
           result = command.execute()
           if result:
               self.history.append(command)
               self.redo_stack.clear()
           return result
       
       def undo(self):
           if not self.history:
               return False
           
           command = self.history.pop()
           result = command.undo()
           if result:
               self.redo_stack.append(command)
           return result
       
       def redo(self):
           if not self.redo_stack:
               return False
           
           command = self.redo_stack.pop()
           result = command.execute()
           if result:
               self.history.append(command)
           return result
   
   # 使用命令模式
   db = Database()
   invoker = Invoker()
   
   # 执行命令
   insert_cmd = InsertCommand(db, {"name": "John", "age": 30})
   invoker.execute_command(insert_cmd)
   
   update_cmd = UpdateCommand(db, 1, {"name": "John", "age": 31})
   invoker.execute_command(update_cmd)
   
   # 撤销操作
   invoker.undo()
   invoker.undo()
   
   # 重做操作
   invoker.redo()
   invoker.redo()
   

九、状态模式

1. 目的：每当对象在内部状态改变时改变它的行为，常用于

   • 订单状态管理
   • 工作流引擎
   • 游戏角色状态
   • UI组件状态管理

2. 实现方法：

   from abc import ABC, abstractmethod
   
   # 状态接口
   class OrderState(ABC):
       @abstractmethod
       def next(self, order):
           pass
       
       @abstractmethod
       def prev(self, order):
           pass
       
       @abstractmethod
       def get_status(self):
           pass
   
   # 具体状态 - 已创建
   class CreatedState(OrderState):
       def next(self, order):
           order.state = PaidState()
           return "Order paid"
       
       def prev(self, order):
           return "Order is already in initial state"
       
       def get_status(self):
           return "Created"
   
   # 具体状态 - 已支付
   class PaidState(OrderState):
       def next(self, order):
           order.state = ShippedState()
           return "Order shipped"
       
       def prev(self, order):
           order.state = CreatedState()
           return "Order creation restored"
       
       def get_status(self):
           return "Paid"
   
   # 具体状态 - 已发货
   class ShippedState(OrderState):
       def next(self, order):
           order.state = DeliveredState()
           return "Order delivered"
       
       def prev(self, order):
           order.state = PaidState()
           return "Order payment restored"
       
       def get_status(self):
           return "Shipped"
   
   # 具体状态 - 已交付
   class DeliveredState(OrderState):
       def next(self, order):
           return "Order is already in final state"
       
       def prev(self, order):
           order.state = ShippedState()
           return "Order shipment restored"
       
       def get_status(self):
           return "Delivered"
   
   # 上下文 - 订单
   class Order:
       def __init__(self):
           self.state = CreatedState()
       
       def next_state(self):
           return self.state.next(self)
       
       def prev_state(self):
           return self.state.prev(self)
       
       def get_status(self):
           return self.state.get_status()
   
   # 使用状态模式
   order = Order()
   print(f"Current status: {order.get_status()}")  # Created
   
   print(order.next_state())  # Order paid
   print(f"Current status: {order.get_status()}")  # Paid
   
   print(order.next_state())  # Order shipped
   print(f"Current status: {order.get_status()}")  # Shipped
   
   print(order.prev_state())  # Order payment restored
   print(f"Current status: {order.get_status()}")  # Paid
   

十、建造者模式

1. 目的：将一个复杂对象的构建与它的表示分离，使得同样的构建过程可以创建不同的表示，常用于

   • 构建复杂的HTML或XML文档
   • 创建复杂的表单对象
   • 构建SQL查询
   • 配置复杂的第三方API请求

2. 实现方法：

   # 产品 - HTML文档
   class HTMLDocument:
       def __init__(self):
           self.elements = []
       
       def add_element(self, element):
           self.elements.append(element)
       
       def __str__(self):
           return "\n".join(self.elements)
   
   # 建造者接口
   class HTMLBuilder:
       def __init__(self):
           self.document = HTMLDocument()
       
       def build_header(self, text, level=1):
           pass
       
       def build_paragraph(self, text):
           pass
       
       def build_list(self, items, ordered=False):
           pass
       
       def get_result(self):
           return self.document
   
   # 具体建造者 - 标准HTML
   class StandardHTMLBuilder(HTMLBuilder):
       def build_header(self, text, level=1):
           self.document.add_element(f"<h{level}>{text}</h{level}>")
           return self
       
       def build_paragraph(self, text):
           self.document.add_element(f"<p>{text}</p>")
           return self
       
       def build_list(self, items, ordered=False):
           list_type = "ol" if ordered else "ul"
           list_items = "".join([f"<li>{item}</li>" for item in items])
           self.document.add_element(f"<{list_type}>{list_items}</{list_type}>")
           return self
   
   # 具体建造者 - 简洁HTML
   class MinimalHTMLBuilder(HTMLBuilder):
       def build_header(self, text, level=1):
           self.document.add_element(f"#{'#' * (level-1)} {text}")
           return self
       
       def build_paragraph(self, text):
           self.document.add_element(text)
           return self
       
       def build_list(self, items, ordered=False):
           prefix = "1." if ordered else "-"
           list_items = "\n".join([f"{prefix} {item}" for item in items])
           self.document.add_element(list_items)
           return self
   
   # 指导者 - 定义构建过程
   class HTMLDirector:
       def __init__(self, builder):
           self.builder = builder
       
       def build_article(self, title, paragraphs, items):
           return (self.builder
                   .build_header(title, 1)
                   .build_paragraph("Introduction:")
                   .build_list(items)
                   .build_paragraph(paragraphs[0])
                   .build_header("Conclusion", 2)
                   .build_paragraph(paragraphs[1])
                   .get_result())
   
   # 使用建造者模式
   # 构建标准HTML
   standard_builder = StandardHTMLBuilder()
   director = HTMLDirector(standard_builder)
   article = director.build_article(
       "Design Patterns",
       ["Design patterns are reusable solutions...", "In conclusion..."],
       ["Creational", "Structural", "Behavioral"]
   )
   print("Standard HTML:")
   print(article)
   print()
   
   # 构建简洁HTML
   minimal_builder = MinimalHTMLBuilder()
   director = HTMLDirector(minimal_builder)
   article = director.build_article(
       "Design Patterns",
       ["Design patterns are reusable solutions...", "In conclusion..."],
       ["Creational", "Structural", "Behavioral"]
   )
   print("Minimal HTML:")
   print(article)