Python设计模式：写代码的套路，学会了事半功倍

设计模式不是什么高深的理论它就是前辈们在长期开发中总结出来的套路。就像下棋有定式、写文章有模板一样用设计模式解决问题代码更清晰、更易维护、更可扩展。这篇文章不讲23种设计模式全部只挑Python开发中最常用、最实用的5种模式配合真实场景的代码示例。一、单例模式全局唯一的对象有些类在程序中只需要一个实例比如配置管理、日志记录器、数据库连接池。单例模式就是用来保证一个类只有一个实例的。Python中有多种实现方式从简单到完善依次来看方式1使用模块最推荐Python的模块本身就是单例的。模块第一次被导入时创建实例后续导入都返回同一个实例。python# config.py class Config: def __init__(self): self.host localhost self.port 8080 self.debug True config Config() # 模块级别的实例 # 任何地方导入 from config import config # 永远是同一个实例这是最Pythonic的方式没有之一。方式2使用装饰器pythondef 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 DatabaseConnection: def __init__(self): self.connected False print(创建数据库连接) def connect(self): self.connected True print(连接成功) # 测试 db1 DatabaseConnection() db2 DatabaseConnection() print(db1 is db2) # True方式3使用__new__经典方式pythonclass Logger: _instance None def __new__(cls): if cls._instance is None: cls._instance super().__new__(cls) cls._instance._initialized False return cls._instance def __init__(self): if self._initialized: return self._initialized True self.logs [] print(初始化日志记录器) def log(self, message): self.logs.append(message) print(f[LOG] {message}) # 测试 log1 Logger() log2 Logger() log1.log(第一条日志) log2.log(第二条日志) print(log1.logs log2.logs) # True实际应用场景配置管理类日志记录器数据库连接池缓存管理器线程池二、工厂模式解耦对象的创建和使用当创建对象的逻辑比较复杂或者需要根据条件创建不同类型的对象时把创建逻辑封装到一个工厂里而不是散落在业务代码中。简单工厂pythonclass Dog: def speak(self): return 汪汪 class Cat: def speak(self): return 喵喵 class Duck: def speak(self): return 嘎嘎 def animal_factory(animal_type): 简单工厂根据类型创建不同的动物 animals { dog: Dog, cat: Cat, duck: Duck } animal_class animals.get(animal_type.lower()) if animal_class: return animal_class() raise ValueError(f不支持的动物类型: {animal_type}) # 使用 animal animal_factory(dog) print(animal.speak()) # 汪汪抽象工厂当产品族变多时用抽象工厂管理多个相关的工厂pythonfrom abc import ABC, abstractmethod # 抽象产品 class Button(ABC): abstractmethod def render(self): pass class Checkbox(ABC): abstractmethod def check(self): pass # 具体产品Windows风格 class WindowsButton(Button): def render(self): return 渲染Windows风格按钮 class WindowsCheckbox(Checkbox): def check(self): return 选中Windows风格复选框 # 具体产品Mac风格 class MacButton(Button): def render(self): return 渲染Mac风格按钮 class MacCheckbox(Checkbox): def check(self): return 选中Mac风格复选框 # 抽象工厂 class UIFactory(ABC): abstractmethod def create_button(self): pass abstractmethod def create_checkbox(self): pass # 具体工厂 class WindowsFactory(UIFactory): def create_button(self): return WindowsButton() def create_checkbox(self): return WindowsCheckbox() class MacFactory(UIFactory): def create_button(self): return MacButton() def create_checkbox(self): return MacCheckbox() # 使用 def render_ui(factory: UIFactory): button factory.create_button() checkbox factory.create_checkbox() print(button.render()) print(checkbox.check()) # 根据操作系统选择工厂 import sys if sys.platform win32: factory WindowsFactory() else: factory MacFactory() render_ui(factory)实际应用场景ORM框架中的数据库连接工厂MySQL、PostgreSQL、SQLite日志框架的日志器工厂支付系统的支付渠道工厂微信支付、支付宝三、观察者模式当一个东西变化时通知其他人观察者模式定义了对象间的一对多依赖关系。当一个对象状态改变时所有依赖它的对象都会收到通知并自动更新。Python中实现观察者模式最直接的方式是事件系统pythonclass EventManager: 事件管理器 def __init__(self): self.listeners {} def subscribe(self, event_type, callback): 订阅事件 if event_type not in self.listeners: self.listeners[event_type] [] self.listeners[event_type].append(callback) def unsubscribe(self, event_type, callback): 取消订阅 if event_type in self.listeners: self.listeners[event_type].remove(callback) def publish(self, event_type, dataNone): 发布事件 if event_type in self.listeners: for callback in self.listeners[event_type]: callback(data) # 创建事件管理器全局单例 event_manager EventManager() # 定义一些观察者 def send_email(data): print(f发送邮件到: {data.get(email, )}) def log_activity(data): print(f记录活动: {data.get(action, )}) def update_cache(data): print(f更新缓存: {data.get(user_id, )}) # 订阅事件 event_manager.subscribe(user_registered, send_email) event_manager.subscribe(user_registered, log_activity) event_manager.subscribe(user_registered, update_cache) event_manager.subscribe(user_login, log_activity) # 发布事件 event_manager.publish(user_registered, { email: userexample.com, action: 注册新用户, user_id: 12345 }) event_manager.publish(user_login, { action: 用户登录, user_id: 12345 })使用Python内置的ObservablePython 3.9pythonfrom __future__ import annotations from typing import List, Callable, Any class Observable: def __init__(self): self._observers: List[Callable[[Any], None]] [] def attach(self, observer: Callable[[Any], None]): self._observers.append(observer) def detach(self, observer: Callable[[Any], None]): self._observers.remove(observer) def notify(self, data: Any): for observer in self._observers: observer(data) class WeatherStation(Observable): def __init__(self): super().__init__() self._temperature 0 property def temperature(self): return self._temperature temperature.setter def temperature(self, value): self._temperature value self.notify(value) # 温度变化时通知所有观察者 # 定义观察者 def phone_display(temp): print(f手机显示: 当前温度 {temp}°C) def window_display(temp): print(f窗户显示: 当前温度 {temp}°C) def alert_system(temp): if temp 40: print(⚠️ 高温警报) # 使用 station WeatherStation() station.attach(phone_display) station.attach(window_display) station.attach(alert_system) station.temperature 25 # 触发所有观察者 station.temperature 42 # 触发高温警报实际应用场景GUI事件处理按钮点击、鼠标移动Web框架的请求生命周期钩子消息队列的发布-订阅模型数据变化时自动更新图表/UI实时监控系统告警四、装饰器模式给函数穿上马甲装饰器模式允许在不修改原函数代码的情况下动态地给函数增加功能。Python的语法就是对这个模式的完美支持。基础装饰器pythonimport time from functools import wraps def timer(func): 计时装饰器 wraps(func) def wrapper(*args, **kwargs): start time.perf_counter() result func(*args, **kwargs) elapsed time.perf_counter() - start print(f{func.__name__} 耗时: {elapsed:.3f}秒) return result return wrapper def retry(max_attempts3, delay1): 重试装饰器 def decorator(func): wraps(func) def wrapper(*args, **kwargs): for attempt in range(max_attempts): try: return func(*args, **kwargs) except Exception as e: if attempt max_attempts - 1: raise print(f第{attempt1}次失败{delay}秒后重试...) time.sleep(delay) return None return wrapper return decorator def cache(func): 简单缓存装饰器 cache_dict {} wraps(func) def wrapper(*args): if args in cache_dict: print(从缓存返回结果) return cache_dict[args] result func(*args) cache_dict[args] result return result return wrapper # 组合使用多个装饰器 timer retry(max_attempts3, delay1) cache def fetch_data(url): 模拟从网络获取数据 import requests response requests.get(url, timeout5) return response.text # 调用顺序cache → retry → timer data fetch_data(https://httpbin.org/delay/1)类装饰器pythonclass CountCalls: 统计函数调用次数 def __init__(self, func): self.func func self.count 0 def __call__(self, *args, **kwargs): self.count 1 print(f第 {self.count} 次调用 {self.func.__name__}) return self.func(*args, **kwargs) CountCalls def say_hello(name): print(fHello, {name}) say_hello(张三) say_hello(李四) # 输出第1次调用 say_hello / Hello, 张三 / 第2次调用 say_hello / Hello, 李四实际应用场景权限检查login_required日志记录log性能统计timer事务管理transactional缓存cached重试retry限流rate_limit五、策略模式算法之间自由切换策略模式定义了一系列算法把它们封装起来并且使它们可以互相替换。pythonfrom abc import ABC, abstractmethod from typing import List # 抽象策略 class SortStrategy(ABC): abstractmethod def sort(self, data: List[int]) - List[int]: pass # 具体策略冒泡排序 class BubbleSort(SortStrategy): def sort(self, data: List[int]) - List[int]: arr data.copy() n len(arr) for i in range(n): for j in range(0, n - i - 1): if arr[j] arr[j 1]: arr[j], arr[j 1] arr[j 1], arr[j] print(使用冒泡排序) return arr # 具体策略快速排序 class QuickSort(SortStrategy): def sort(self, data: List[int]) - List[int]: if len(data) 1: return data pivot data[0] left [x for x in data[1:] if x pivot] right [x for x in data[1:] if x pivot] print(使用快速排序) return self.sort(left) [pivot] self.sort(right) # 具体策略内置排序最优 class BuiltInSort(SortStrategy): def sort(self, data: List[int]) - List[int]: print(使用内置排序) return sorted(data) # 上下文 class Sorter: def __init__(self, strategy: SortStrategy): self._strategy strategy def set_strategy(self, strategy: SortStrategy): self._strategy strategy def sort_data(self, data: List[int]) - List[int]: return self._strategy.sort(data) # 使用 data [64, 34, 25, 12, 22, 11, 90] # 默认使用内置排序 sorter Sorter(BuiltInSort()) result sorter.sort_data(data) print(result) # 切换策略 sorter.set_strategy(QuickSort()) result sorter.sort_data(data) print(result) # 再切换 sorter.set_strategy(BubbleSort()) result sorter.sort_data(data) print(result)更实际的例子支付策略pythonfrom abc import ABC, abstractmethod class PaymentStrategy(ABC): abstractmethod def pay(self, amount: float) - bool: pass class WeChatPay(PaymentStrategy): def __init__(self, openid: str): self.openid openid def pay(self, amount: float) - bool: print(f微信支付: {self.openid} 支付 {amount} 元) # 调用微信支付API... return True class Alipay(PaymentStrategy): def __init__(self, user_id: str): self.user_id user_id def pay(self, amount: float) - bool: print(f支付宝支付: {self.user_id} 支付 {amount} 元) # 调用支付宝API... return True class CreditCard(PaymentStrategy): def __init__(self, card_number: str, cvv: str): self.card_number card_number self.cvv cvv def pay(self, amount: float) - bool: print(f信用卡支付: {self.card_number[-4:]} 支付 {amount} 元) return True class Order: def __init__(self, amount: float): self.amount amount self.payment_strategy None def set_payment_method(self, strategy: PaymentStrategy): self.payment_strategy strategy def checkout(self): if not self.payment_strategy: raise ValueError(请先设置支付方式) return self.payment_strategy.pay(self.amount) # 使用 order Order(99.99) # 用户选择微信支付 order.set_payment_method(WeChatPay(oUpF8uMuAJO_M2pxb1Q9zNjWeS6o)) order.checkout() # 用户切换支付宝 order.set_payment_method(Alipay(2088102177732799)) order.checkout()实际应用场景支付系统微信、支付宝、银联文件格式处理JSON、XML、CSV的解析和导出压缩算法ZIP、RAR、7Z数据验证规则邮箱、手机、身份证图像处理算法不同滤镜、裁剪方式设计模式使用原则设计模式虽好但不要滥用。记住这几个原则KISS原则能简单就别搞复杂。不要为了用设计模式而用设计模式YAGNI原则你不需要它的时候就别加开闭原则对扩展开放对修改关闭单一职责一个类只做一件事依赖倒置依赖抽象不依赖具体实现总结模式一句话理解什么时候用单例模式全局唯一实例配置管理、日志器、连接池工厂模式封装对象的创建创建逻辑复杂、需要根据条件创建不同类型观察者模式发布-订阅通知状态变化需要通知多个对象装饰器模式给函数增加功能日志、计时、缓存、权限检查策略模式算法自由切换多种实现方式、需要动态切换建议先写能跑的代码再看是否需要重构设计模式是为了解决特定问题不是目标多读优秀的开源代码看别人怎么用的最重要的是理解模式背后的思想而不是死记硬背用好设计模式你的代码会越来越优雅团队协作也越来越顺畅。