Python设计模式：工程实践中的模式应用

Python设计模式：工程实践中的模式应用

1. 技术分析

1.1 设计模式分类

1.2 Python实现特点

Python的动态特性和内置机制使得某些传统模式可以更简洁地实现：

• 单例模式 → 模块级实现
• 装饰器模式 → @decorator语法
• 上下文管理器 → with语句

2. 核心功能实现

2.1 单例模式

# 方法1: 模块级单例（推荐） # database.py class Database: def __init__(self): self.connection = None def connect(self): if self.connection is None: print("建立数据库连接") self.connection = "conn_123" return self.connection # 全局实例 _db_instance = Database() def get_database(): return _db_instance # 方法2: 元类单例 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 DatabaseMeta(metaclass=SingletonMeta): def __init__(self): self.connection = None def connect(self): if self.connection is None: print("建立数据库连接") self.connection = "conn_456" return self.connection # 方法3: 装饰器单例 from functools import wraps def singleton(cls): instances = {} @wraps(cls) def get_instance(*args, **kwargs): if cls not in instances: instances[cls] = cls(*args, **kwargs) return instances[cls] return get_instance @singleton class CacheService: def __init__(self): self.cache = {} def get(self, key): return self.cache.get(key) def set(self, key, value): self.cache[key] = value

2.2 工厂模式

from abc import ABC, abstractmethod # 抽象产品 class Model(ABC): @abstractmethod def predict(self, data): pass # 具体产品 class LinearRegressionModel(Model): def __init__(self, lr=0.01): self.lr = lr self.weights = None def predict(self, data): return sum(w * x for w, x in zip(self.weights, data)) def fit(self, X, y): # 简化实现 self.weights = [0.5] * len(X[0]) class DecisionTreeModel(Model): def __init__(self, max_depth=10): self.max_depth = max_depth self.tree = {} def predict(self, data): return 0.5 # 简化实现 # 工厂 class ModelFactory: _registry = {} @classmethod def register(cls, model_type, model_class): cls._registry[model_type] = model_class @classmethod def create(cls, model_type, **kwargs): if model_type not in cls._registry: raise ValueError(f"Unknown model type: {model_type}") return cls._registry[model_type](**kwargs) # 注册模型 ModelFactory.register("linear", LinearRegressionModel) ModelFactory.register("tree", DecisionTreeModel) # 使用 model = ModelFactory.create("linear", lr=0.001) result = model.predict([1.0, 2.0, 3.0])

2.3 策略模式

from abc import ABC, abstractmethod class OptimizationStrategy(ABC): @abstractmethod def optimize(self, data): pass class GradientDescent(OptimizationStrategy): def __init__(self, lr=0.01): self.lr = lr def optimize(self, data): return "GD优化, lr=" + str(self.lr) class AdamOptimizer(OptimizationStrategy): def __init__(self, lr=0.001, beta1=0.9, beta2=0.999): self.lr = lr self.beta1 = beta1 self.beta2 = beta2 def optimize(self, data): return f"Adam优化, lr={self.lr}" class SGDOptimizer(OptimizationStrategy): def __init__(self, lr=0.01, momentum=0.9): self.lr = lr self.momentum = momentum def optimize(self, data): return f"SGD优化, lr={self.lr}, momentum={self.momentum}" class Trainer: def __init__(self, strategy: OptimizationStrategy): self.strategy = strategy def set_strategy(self, strategy: OptimizationStrategy): self.strategy = strategy def train(self, data): return self.strategy.optimize(data) # 使用 trainer = Trainer(GradientDescent(lr=0.01)) print(trainer.train(data)) trainer.set_strategy(AdamOptimizer(lr=0.001)) print(trainer.train(data))

2.4 观察者模式

from collections import defaultdict class EventObserver: def __init__(self): self._handlers = defaultdict(list) def subscribe(self, event, handler): self._handlers[event].append(handler) return lambda: self.unsubscribe(event, handler) def unsubscribe(self, event, handler): if event in self._handlers: self._handlers[event].remove(handler) def emit(self, event, *args, **kwargs): for handler in self._handlers[event]: handler(*args, **kwargs) class TrainingMonitor: def __init__(self): self.observer = EventObserver() self.history = {'loss': [], 'accuracy': []} def on_epoch_end(self, epoch, loss, accuracy): self.history['loss'].append(loss) self.history['accuracy'].append(accuracy) self.observer.emit('epoch_end', epoch=epoch, loss=loss, accuracy=accuracy) # 检查早停 if len(self.history['loss']) > 5: if all(self.history['loss'][-i] >= self.history['loss'][-i-1] for i in range(1, 5)): self.observer.emit('early_stop', epoch=epoch) # 使用 monitor = TrainingMonitor() def save_model(epoch, **kwargs): print(f"保存模型: epoch={epoch}") def send_notification(epoch, loss, **kwargs): print(f"通知: Epoch {epoch}, Loss={loss:.4f}") monitor.observer.subscribe('epoch_end', save_model) monitor.observer.subscribe('epoch_end', send_notification) monitor.observer.subscribe('early_stop', lambda **k: print("早停触发")) # 模拟训练 for epoch in range(10): loss = 1.0 - epoch * 0.1 accuracy = 0.5 + epoch * 0.05 monitor.on_epoch_end(epoch, loss, min(accuracy, 0.99))

2.5 装饰器模式

import time from functools import wraps def log_calls(func): """日志装饰器""" @wraps(func) def wrapper(*args, **kwargs): print(f"调用 {func.__name__}") start = time.perf_counter() result = func(*args, **kwargs) elapsed = time.perf_counter() - start print(f"{func.__name__} 完成, 耗时: {elapsed:.4f}s") return result return wrapper def retry(max_attempts=3, delay=1.0): """重试装饰器""" 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: time.sleep(delay) print(f"重试 {attempt + 1}/{max_attempts}") else: raise return wrapper return decorator def cache_result(func): """缓存装饰器""" @wraps(func) def wrapper(*args, **kwargs): cache_key = str(args) + str(kwargs) if not hasattr(wrapper, '_cache'): wrapper._cache = {} if cache_key not in wrapper._cache: wrapper._cache[cache_key] = func(*args, **kwargs) return wrapper._cache[cache_key] return wrapper @log_calls @retry(max_attempts=3, delay=0.5) @cache_result def fetch_data(url): """带日志、重试和缓存的数据获取""" print(f"从 {url} 获取数据") time.sleep(0.5) return {"status": "success", "data": [1, 2, 3]} # 链式调用 result = fetch_data("http://example.com/api") print(result)

2.6 上下文管理器

class DatabaseConnection: """数据库连接上下文管理器""" def __init__(self, dsn): self.dsn = dsn self.connection = None def __enter__(self): print(f"连接数据库: {self.dsn}") self.connection = f"conn_to_{self.dsn}" return self def __exit__(self, exc_type, exc_val, exc_tb): if exc_type: print(f"数据库错误: {exc_val}") print("关闭数据库连接") self.connection = None return False # 不抑制异常 class Transaction: """事务上下文管理器""" def __init__(self, connection): self.connection = connection self.committed = False def __enter__(self): print("开始事务") return self def __exit__(self, exc_type, exc_val, exc_tb): if exc_type: print(f"事务回滚: {exc_val}") self.rollback() return True # 抑制异常 else: self.commit() return False def commit(self): print("提交事务") self.committed = True def rollback(self): print("回滚事务") # 使用 with DatabaseConnection("mydb") as db: with Transaction(db.connection): print("执行SQL操作...") # raise Exception("测试异常") # 组合使用 with DatabaseConnection("mydb") as db: with Transaction(db.connection) as txn: print("执行SQL操作...")

3. 性能对比

3.1 单例模式实现对比

import timeit # 模块级 vs 元类 module_singleton = """ class _Instance: def __init__(self): self.value = None _instance = _Instance() """ metaclass_singleton = """ 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 Instance(metaclass=SingletonMeta): def __init__(self): self.value = None """ # 测试 module_time = timeit.timeit("_instance", setup=module_singleton, number=100000) meta_time = timeit.timeit("Instance()", setup=metaclass_singleton, number=100000) print(f"模块级: {module_time:.4f}s") print(f"元类: {meta_time:.4f}s")

4. 最佳实践

4.1 模式选择指南

4.2 注意事项

# ✅ 推荐：简单场景用简单实现 # 单例 → 模块级 # 工厂 → 函数+字典注册 # ❌ 避免：过度使用设计模式 # 简单的if-else能解决的问题不需要工厂模式

5. 总结

设计模式应用要点：

1. 适度使用：不要为了模式而模式
2. Python风格：善用Python特性简化实现
3. 组合使用：多个模式可以组合解决复杂问题